refactor(qqbot): 重构QQ机器人适配器代码，优化多项功能与结构

主要变更：
1. 修复速率限流器使用setdefault替代重复创建令牌桶
2. 重构交互注册表匹配逻辑，优化精确匹配查找
3. 重构去重缓存逻辑，移到适配器实例方法
4. 重构发送URL解析，增加合法性校验并拆分公共方法
5. 优化流式消息处理逻辑，简化flush_controller调用
6. 重构群聊类型判断代码，简化语法
7. 修复重连管理器对None类型关闭分类的处理
8. 新增消息缓存、线程模拟器、发送初始化模块
9. 重构凭证备份与会话存储逻辑，支持环境变量指定状态目录
10. 新增配置提示与向导二维码绑定功能
11. 优化媒体上传逻辑，增加重试机制与缓存
12. 新增审批键盘模板构建函数
13. 重构消息格式处理，修正媒体发送字段与长度限制
14. 修复令牌过期时间计算，使用time.time替代monotonic
15. 新增群组激活缓冲区与用户追踪器增强功能
16. 修复换行符问题，统一文件结尾格式

backend/package/yuxi/channels/adapters/qqbot/adapter.py

@@ -14,6 +14,7 @@ from yuxi.channels.base import BaseChannelAdapter
 from yuxi.channels.capabilities import ChannelCapabilities
 from yuxi.channels.exceptions import (
     ChannelAuthenticationError,
+    ChannelException,
     ChannelNotConnectedError,
     DeliveryFailedError,
     MessageFormatError,
@@ -56,6 +57,7 @@ from .group_buffer import GroupHistoryBuffer
 from .inbound_pipeline import QQBotInboundPipeline
 from .interaction import InteractionContext, InteractionRegistry, InteractionSessionManager
 from .known_users import KnownUserTracker
+from .message_cache import MessageCache
 from .message_queue import MessageQueue
 from .probe import health_check_dsm
 from .rate_limiter import RouteRateLimiter
@@ -63,6 +65,7 @@ from .reconnect import QQBotReconnectManager, ReconnectState
 from .reply_dispatcher import ReplyDispatcher
 from .security import QQBotSecurityPolicy, verify_webhook_ed25519
 from .send import MessageSeqManager, send_with_retry
+from .sender_init import SenderInfo, build_sender_headers, init_sender
 from .session_store import SessionRecord, SessionStore
 from .streaming import ParagraphChunker, stream_content
 from .token import QQBotTokenManager
@@ -122,7 +125,6 @@ class QQBotAdapter(BaseChannelAdapter):
         self._sandbox = self.config.get("sandbox", False)
         self._ws: aiohttp.ClientWebSocketResponse | None = None
         self._seq: int | None = None
-        self._session_id: str | None = None
         self._heartbeat_interval: float = 30.0
         self._heartbeat_task: asyncio.Task | None = None
         self._last_heartbeat_ack: float = 0
@@ -179,6 +181,7 @@ class QQBotAdapter(BaseChannelAdapter):
         register_streaming_command(self._command_registry)
         self._interaction_registry = InteractionRegistry()
         self._interaction_sessions = InteractionSessionManager()
+        self._inbound_pipeline = QQBotInboundPipeline(self)
         self._approval_manager = ExecApprovalManager()
         self._tts_provider = TTSProvider(
             default_voice=self.config.get("tts_default_voice", "zh-CN-XiaoxiaoNeural"),
@@ -187,10 +190,22 @@ class QQBotAdapter(BaseChannelAdapter):
         self._stt_provider = STTProvider.from_config(self.config)
         self._audio_format_policy = AudioFormatPolicy.from_config(self.config)
         self._typing_keepalive = self._init_typing_keepalive()
+        self._sender_info = self._init_sender_info()
+        self._message_cache = MessageCache(
+            max_messages=int(self.config.get("message_cache_max", 500)),
+            ttl_s=float(self.config.get("message_cache_ttl_s", 3600.0)),
+        )
+        self._group_stream_buffers: dict[str, str] = {}
+        self._group_stream_senders: dict[str, Any] = {}
 
     def _init_typing_keepalive(self) -> TypingKeepAlive:
         return TypingKeepAlive(send_notify_fn=self._send_input_notify)
 
+    def _init_sender_info(self) -> SenderInfo:
+        plugin_version = self.config.get("plugin_version", "1.0.0")
+        openclaw_version = self.config.get("openclaw_version", "1.0.0")
+        return init_sender(plugin_version=plugin_version, openclaw_version=openclaw_version)
+
     async def _send_input_notify(self, chat_id: str) -> None:
         if not self._http_client or not self._token_manager:
             return
@@ -220,6 +235,8 @@ class QQBotAdapter(BaseChannelAdapter):
 
         logger.info(f"[QQBot] Starting channel '{self.channel_id}'...")
 
+        await self._inbound_pipeline.initialize()
+
         try:
             app_id = self.config.get("app_id") or os.environ.get("QQBOT_APP_ID", "")
             app_secret = (
@@ -255,6 +272,32 @@ class QQBotAdapter(BaseChannelAdapter):
             hello_data = await self._wait_for_hello()
             self._heartbeat_interval = hello_data.get("heartbeat_interval", 30000) / 1000.0
 
+            session_id = self._reconnect_manager._session_id
+            last_seq = self._reconnect_manager._last_seq
+            if session_id and last_seq is not None:
+                logger.info("[QQBot] Attempting RESUME with session_id=%s seq=%s", session_id[:16], last_seq)
+                await self._send_resume()
+                try:
+                    ready_data = await asyncio.wait_for(self._wait_for_ready(), timeout=5.0)
+                    if ready_data:
+                        logger.info("[QQBot] RESUME successful, session recovered")
+                        self._reconnect_manager.mark_connected()
+                        await self._reconnect_manager.transition(ReconnectState.CONNECTED)
+                        self._status = ChannelStatus.CONNECTED
+                        self._c2c_streaming = C2CStreamingController(
+                            send_message_fn=self._send_stream_message,
+                            flush_interval=self.config.get("stream_flush_interval", 0.6),
+                        )
+                        self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())
+                        self._receive_task = asyncio.create_task(self._ws_receive_loop())
+                        self._backup_credentials()
+                        self._save_session()
+                        self._register_runtime_contexts()
+                        logger.info("[QQBot] Channel '%s' resumed successfully", self.channel_id)
+                        return
+                except (TimeoutError, Exception):
+                    logger.info("[QQBot] RESUME failed, falling back to IDENTIFY")
+
             await self._send_identify()
 
             self._bot_info = {
@@ -271,13 +314,15 @@ class QQBotAdapter(BaseChannelAdapter):
             self._receive_task = asyncio.create_task(self._ws_receive_loop())
 
             self._status = ChannelStatus.CONNECTED
-            await self._reconnect_manager.on_identify_success(self._session_id or "")
+            await self._reconnect_manager.on_identify_success(self._reconnect_manager.session_id or "")
             self._reconnect_manager.mark_connected()
             await self._reconnect_manager.transition(ReconnectState.CONNECTED)
 
             self._backup_credentials()
             self._save_session()
 
+            self._register_runtime_contexts()
+
             logger.info(
                 f"[QQBot] Channel '{self.channel_id}' started "
                 f"(sandbox={self._sandbox}, heartbeat={self._heartbeat_interval:.0f}s)"
@@ -338,22 +383,25 @@ class QQBotAdapter(BaseChannelAdapter):
             try:
                 await self._ws.close()
             except Exception:
-                pass
+                logger.debug("[QQBot] Error closing WebSocket during cleanup", exc_info=True)
         self._ws = None
         self._seq = None
-        self._session_id = None
 
     # ========== WebSocket 内部逻辑 ==========
 
     async def _wait_for_hello(self) -> dict:
         if not self._ws:
             raise ChannelAuthenticationError("WebSocket not connected")
-        async for msg in self._ws:
-            data = json.loads(msg.data)
-            op = data.get("op")
-            if op == Opcode.HELLO:
-                return data.get("d", {})
-            logger.debug(f"[QQBot] Skipping op={op} before HELLO")
+        try:
+            async with asyncio.timeout(30):
+                async for msg in self._ws:
+                    data = json.loads(msg.data)
+                    op = data.get("op")
+                    if op == Opcode.HELLO:
+                        return data.get("d", {})
+                    logger.debug(f"[QQBot] Skipping op={op} before HELLO")
+        except TimeoutError:
+            raise ChannelAuthenticationError("Timeout waiting for HELLO from QQ Bot Gateway (30s)")
 
     async def _send_identify(self) -> None:
         if not self._ws:
@@ -378,12 +426,25 @@ class QQBotAdapter(BaseChannelAdapter):
             "op": Opcode.RESUME,
             "d": {
                 "token": f"QQBot {token}",
-                "session_id": self._session_id,
-                "seq": self._seq,
+                "session_id": self._reconnect_manager.session_id,
+                "seq": self._reconnect_manager.last_seq,
             },
         }
         await self._ws.send_json(payload)
 
+    async def _wait_for_ready(self) -> dict | None:
+        if not self._ws:
+            return None
+        async for msg in self._ws:
+            data = json.loads(msg.data)
+            op = data.get("op")
+            t = data.get("t", "")
+            if op == Opcode.DISPATCH and t in ("READY", "RESUMED"):
+                return data.get("d", {})
+            if op in (Opcode.INVALID_SESSION, Opcode.RECONNECT):
+                return None
+        return None
+
     async def _heartbeat_loop(self) -> None:
         max_misses = 3
         while self._status == ChannelStatus.CONNECTED and self._ws and not self._ws.closed:
@@ -393,6 +454,7 @@ class QQBotAdapter(BaseChannelAdapter):
                 if self._heartbeat_miss_count >= max_misses:
                     logger.warning(f"[QQBot] Heartbeat missed {max_misses} consecutive ACKs, disconnecting")
                     self._status = ChannelStatus.RECONNECTING
+                    await self._reconnect_manager.on_disconnect(None)
                     break
 
                 send_time = time.monotonic()
@@ -418,6 +480,7 @@ class QQBotAdapter(BaseChannelAdapter):
             except Exception as e:
                 logger.warning(f"[QQBot] Heartbeat failed: {e}")
                 self._status = ChannelStatus.RECONNECTING
+                await self._reconnect_manager.on_disconnect(None)
                 break
 
     async def _ws_receive_loop(self) -> None:
@@ -474,9 +537,9 @@ class QQBotAdapter(BaseChannelAdapter):
                 self._status = ChannelStatus.ERROR
 
     async def _dispatch_event(self, event_type: str, event_data: dict) -> None:
-        pipeline = QQBotInboundPipeline(self)
-        await pipeline.initialize()
-        ctx = await pipeline.process(event_type, event_data)
+        if not self._inbound_pipeline._stages:
+            await self._inbound_pipeline.initialize()
+        await self._inbound_pipeline.process(event_type, event_data)
 
     def _prune_old_msg_ids(self, now: float) -> None:
         expired = [mid for mid, ts in self._recent_msg_ids.items() if now - ts >= self._dedup_window_s]
@@ -513,8 +576,22 @@ class QQBotAdapter(BaseChannelAdapter):
             try:
                 await self._typing_keepalive.start(chat_id)
                 await self._message_handler(msg)
+            except ChannelException as e:
+                if e.retryable:
+                    logger.warning(
+                        "[QQBot] Retryable error in message handler (chat=%s): %s",
+                        chat_id,
+                        e,
+                    )
+                else:
+                    logger.exception(
+                        "[QQBot] Unrecoverable error in message handler (chat=%s)",
+                        chat_id,
+                    )
+                    await self._send_error_notification(msg, str(e))
             except Exception:
-                logger.exception("[QQBot] Error in message handler")
+                logger.exception("[QQBot] Unexpected error in message handler (chat=%s)", chat_id)
+                await self._send_error_notification(msg, "内部错误，请稍后重试")
             finally:
                 await self._typing_keepalive.stop(chat_id)
 
@@ -636,7 +713,17 @@ class QQBotAdapter(BaseChannelAdapter):
             logger.exception("[QQBot] Interaction handling failed")
             return False
 
-    # ========== 消息收发 ==========
+    # ========== 消息收发 ==========
+
+    async def _send_error_notification(self, msg: ChannelMessage, error_msg: str) -> None:
+        try:
+            response = ChannelResponse(
+                identity=msg.identity,
+                content=f"抱歉，处理您的消息时遇到了问题：{error_msg}",
+            )
+            await self.send(response)
+        except Exception:
+            logger.warning("[QQBot] Failed to send error notification to user", exc_info=True)
 
     async def send(self, response: ChannelResponse) -> DeliveryResult:
         if not self._http_client or not self._token_manager:
@@ -666,18 +753,93 @@ class QQBotAdapter(BaseChannelAdapter):
                 chat_id,
                 self.config,
                 token_refresh_cb=_refresh_token,
+                sender_headers=build_sender_headers(self._sender_info),
             )
 
         try:
             await self._rate_limiter.acquire("send_message")
             result = await self._circuit_breaker.call(_do_send)
             self._last_outbound_at = time.time()
+            self._message_cache.record_sent(result, chat_id, content=response.content)
             return result
         except CircuitBreakerOpenError:
             return DeliveryResult(success=False, error="Circuit breaker open")
         except (DeliveryFailedError, Exception) as e:
             return DeliveryResult(success=False, error=str(e))
 
+    async def send_stream_chunk(
+        self,
+        chat_id: str,
+        msg_id: str,
+        chunk: str,
+        finished: bool = False,
+    ) -> DeliveryResult:
+        is_group = chat_id.startswith(GROUP_CHAT_PREFIX) if GROUP_CHAT_PREFIX else False
+
+        if not is_group:
+            from yuxi.channels.models import ChannelIdentity, ChannelResponse
+
+            identity = ChannelIdentity(
+                channel_id=self.channel_id,
+                channel_type=self.channel_type,
+                channel_chat_id=chat_id,
+                channel_user_id="",
+            )
+            response = ChannelResponse(identity=identity, content=chunk)
+            return await self.send(response)
+
+        return await self._send_group_stream_chunk(chat_id, msg_id, chunk, finished)
+
+    async def _send_group_stream_chunk(
+        self,
+        chat_id: str,
+        msg_id: str,
+        chunk: str,
+        finished: bool = False,
+    ) -> DeliveryResult:
+        existing = self._group_stream_buffers.get(chat_id, "")
+        accumulated = existing + chunk
+
+        if not finished:
+            paragraphs = accumulated.split("\n\n")
+            if len(paragraphs) > 1:
+                for para in paragraphs[:-1]:
+                    if para.strip():
+                        await self._send_group_paragraph(chat_id, para)
+                self._group_stream_buffers[chat_id] = paragraphs[-1]
+            else:
+                self._group_stream_buffers[chat_id] = accumulated
+            return DeliveryResult(success=True)
+
+        self._group_stream_buffers.pop(chat_id, None)
+        self._group_stream_senders.pop(chat_id, None)
+
+        from yuxi.channels.models import ChannelType
+
+        from .streaming import send_blocks_stream
+
+        await send_blocks_stream(
+            chat_id=chat_id,
+            text=accumulated,
+            send_fn=self.send,
+            channel_id=self.channel_id,
+            channel_type=ChannelType.QQ_BOT,
+            chunk_size=self.text_chunk_limit,
+        )
+        return DeliveryResult(success=True)
+
+    async def _send_group_paragraph(self, chat_id: str, text: str) -> DeliveryResult:
+        from yuxi.channels.models import ChannelIdentity, ChannelResponse
+
+        identity = ChannelIdentity(
+            channel_id=self.channel_id,
+            channel_type=self.channel_type,
+            channel_chat_id=chat_id,
+            channel_user_id="",
+        )
+        response = ChannelResponse(identity=identity, content=text.strip())
+        return await self.send(response)
+
     async def send_media(self, chat_id: str, media_type: str, data: Any) -> DeliveryResult:
         if not self._http_client or not self._token_manager:
             return DeliveryResult(success=False, error="Client not initialized")
@@ -774,6 +936,20 @@ class QQBotAdapter(BaseChannelAdapter):
         prompt: str = "",
         fields: list[dict] | None = None,
     ) -> DeliveryResult:
+        if chat_id.startswith(DM_CHAT_PREFIX) or chat_id.startswith(GROUP_CHAT_PREFIX):
+            logger.debug("[QQBot] Embed not supported in C2C/Group, degrading to text")
+            return await self.send(
+                ChannelResponse(
+                    identity=ChannelIdentity(
+                        channel_id=self.channel_id,
+                        channel_type=self.channel_type,
+                        channel_user_id="",
+                        channel_chat_id=chat_id,
+                    ),
+                    content=f"{title}\n{description}" if title else description,
+                )
+            )
+
         from .ark_templates import build_embed
 
         payload = build_embed(title=title, description=description, prompt=prompt, fields=fields)
@@ -911,7 +1087,7 @@ class QQBotAdapter(BaseChannelAdapter):
     async def receive(self) -> AsyncIterator[ChannelMessage]:
         while self._status == ChannelStatus.CONNECTED:
             try:
-                msg = await self._message_queue.get()
+                msg = await self._message_queue.dequeue()
                 yield msg
             except asyncio.CancelledError:
                 break
@@ -1124,7 +1300,7 @@ class QQBotAdapter(BaseChannelAdapter):
             ),
             access_token=self._token_manager.access_token if self._token_manager else "",
             expires_at=self._token_manager.expires_at if self._token_manager else 0,
-            session_id=self._session_id or "",
+            session_id=self._reconnect_manager.session_id or "",
             sandbox=self._sandbox,
         )
         self._credential_backup.save(snapshot)
@@ -1147,7 +1323,7 @@ class QQBotAdapter(BaseChannelAdapter):
         if not self.config.get("app_secret") and snapshot.app_secret:
             logger.info("[QQBot] Restored app_secret from credential backup")
 
-        self._session_id = snapshot.session_id
+        self._reconnect_manager._session_id = snapshot.session_id
         return True
 
     def _init_session_store(self) -> None:
@@ -1174,7 +1350,7 @@ class QQBotAdapter(BaseChannelAdapter):
             return
 
         record = SessionRecord(
-            session_id=self._reconnect_manager._session_id or self._session_id or "",
+            session_id=self._reconnect_manager._session_id or "",
             last_seq=self._reconnect_manager._last_seq,
             last_heartbeat=time.monotonic(),
             identify_at=time.time(),
@@ -1299,8 +1475,6 @@ class QQBotAdapter(BaseChannelAdapter):
         )
         if msg_id:
             response.reply_to_message_id = msg_id
-        payload = format_outbound(response)
-        payload["msg_id"] = msg_id
         return await self.send(response)
 
     async def send_group_message(
@@ -1320,8 +1494,6 @@ class QQBotAdapter(BaseChannelAdapter):
         )
         if msg_id:
             response.reply_to_message_id = msg_id
-        payload = format_outbound(response)
-        payload["msg_id"] = msg_id
         return await self.send(response)
 
     async def _send_raw_payload(self, chat_id: str, payload: dict) -> DeliveryResult:
@@ -1351,9 +1523,9 @@ class QQBotAdapter(BaseChannelAdapter):
             return None
         token = await self._token_manager.get_token()
         api_base = self._token_manager.api_base
-        from .send import _resolve_send_url
+        from .send import resolve_send_url
 
-        url = _resolve_send_url(api_base, chat_id)
+        url = resolve_send_url(api_base, chat_id)
         headers = {
             "Authorization": f"QQBot {token}",
             "Content-Type": "application/json",
@@ -1449,9 +1621,7 @@ class QQBotAdapter(BaseChannelAdapter):
     def is_configured(self) -> bool:
         app_id = self.config.get("app_id") or os.environ.get("QQBOT_APP_ID", "")
         app_secret = (
-            self.config.get("app_secret")
-            or os.environ.get("QQBOT_CLIENT_SECRET", "")
-            or self._read_app_secret_file()
+            self.config.get("app_secret") or os.environ.get("QQBOT_CLIENT_SECRET", "") or self._read_app_secret_file()
         )
         if app_id and app_secret:
             return True
@@ -1469,9 +1639,7 @@ class QQBotAdapter(BaseChannelAdapter):
     def unconfigured_reason(self) -> str:
         app_id = self.config.get("app_id") or os.environ.get("QQBOT_APP_ID", "")
         app_secret = (
-            self.config.get("app_secret")
-            or os.environ.get("QQBOT_CLIENT_SECRET", "")
-            or self._read_app_secret_file()
+            self.config.get("app_secret") or os.environ.get("QQBOT_CLIENT_SECRET", "") or self._read_app_secret_file()
         )
         if not app_id:
             return "missing_app_id"
@@ -1501,7 +1669,9 @@ class QQBotAdapter(BaseChannelAdapter):
         app_id = self.config.get("app_id") or os.environ.get("QQBOT_APP_ID", "")
         return {
             "configured": self.is_configured(),
-            "tokenSource": "env" if os.environ.get("QQBOT_CLIENT_SECRET") else ("file" if self.config.get("client_secret_file") else "config"),
+            "tokenSource": "env"
+            if os.environ.get("QQBOT_CLIENT_SECRET")
+            else ("file" if self.config.get("client_secret_file") else "config"),
             "running": self._status == ChannelStatus.CONNECTED,
             "connected": self._ws is not None and not (self._ws.closed if hasattr(self._ws, "closed") else False),
             "status": self._status.value,
@@ -1515,7 +1685,7 @@ class QQBotAdapter(BaseChannelAdapter):
             "lastInboundAt": getattr(self, "_last_inbound_at", None),
             "lastOutboundAt": getattr(self, "_last_outbound_at", None),
             "wsSeq": self._seq,
-            "sessionId": self._session_id,
+            "sessionId": self._reconnect_manager.session_id,
             "heartbeatLatencyMs": self._heartbeat_latency_ms,
             "reconnectState": self._reconnect_manager.state.name if self._reconnect_manager else "N/A",
         }
@@ -1532,3 +1702,20 @@ class QQBotAdapter(BaseChannelAdapter):
 
     def has_native_approval_routing(self) -> bool:
         return self.config.get("approval_native_routing", True)
+
+    def _register_runtime_contexts(self) -> None:
+        self._runtime_contexts = {
+            "approval.native": {
+                "enabled": self.has_native_approval_routing(),
+                "channel_id": self.channel_id,
+                "suppress_local_prompt": self.should_suppress_local_approval_prompt(),
+                "keyboard_provider": "qqbot_inline",
+            },
+        }
+        logger.debug(
+            "[QQBot] Runtime contexts registered: approval.native=%s",
+            self._runtime_contexts.get("approval.native", {}).get("enabled"),
+        )
+
+    def get_runtime_contexts(self) -> dict:
+        return getattr(self, "_runtime_contexts", {})

backend/package/yuxi/channels/adapters/qqbot/ark_templates.py

@@ -76,5 +76,49 @@ def build_text_card(text: str, buttons: list[dict] | None = None) -> dict:
     return payload
 
 
+def build_approval_keyboard(
+    request_id: str,
+    action_label: str = "Approve",
+    reject_label: str = "Reject",
+    cancel_label: str = "Cancel",
+) -> dict:
+    buttons = [
+        {
+            "id": f"approval_approve_{request_id}",
+            "render_data": {"label": action_label, "visited_label": "✅ 已批准"},
+            "action": {
+                "type": 2,
+                "permission": {"type": 2},
+                "click_limit": 1,
+                "data": f"approve:{request_id}",
+            },
+            "group_id": "approval",
+        },
+        {
+            "id": f"approval_reject_{request_id}",
+            "render_data": {"label": reject_label, "visited_label": "❌ 已拒绝"},
+            "action": {
+                "type": 2,
+                "permission": {"type": 2},
+                "click_limit": 1,
+                "data": f"reject:{request_id}",
+            },
+            "group_id": "approval",
+        },
+        {
+            "id": f"approval_cancel_{request_id}",
+            "render_data": {"label": cancel_label, "visited_label": "↩️ 已取消"},
+            "action": {
+                "type": 2,
+                "permission": {"type": 2},
+                "click_limit": 1,
+                "data": f"cancel:{request_id}",
+            },
+            "group_id": "approval",
+        },
+    ]
+    return {"content": {"rows": [{"buttons": buttons}]}}
+
+
 def _kv(key: str, value: str) -> dict[str, str]:
     return {"key": key, "value": value}

backend/package/yuxi/channels/adapters/qqbot/c2c_stream.py

@@ -43,6 +43,9 @@ class StreamCheckpoint:
         return time.time() - self.timestamp > 600
 
 
+BATCH_AFTER_GAP = 1.5
+
+
 @dataclass
 class FlushController:
     strategy: FlushStrategy = FlushStrategy.BACKLOG
@@ -54,16 +57,23 @@ class FlushController:
     _buffer: list[str] = field(default_factory=list, repr=False)
     _last_flush: float = 0.0
     _accumulated: str = ""
+    _flush_in_progress: bool = False
+    _needs_reflush: bool = False
+    _last_chunk_time: float = 0.0
 
     def reset(self) -> None:
         self._buffer = []
         self._last_flush = 0.0
         self._accumulated = ""
         self.seq_counter = 0
+        self._flush_in_progress = False
+        self._needs_reflush = False
+        self._last_chunk_time = 0.0
 
     def feed(self, chunk: str) -> list[str]:
         self._buffer.append(chunk)
         self._accumulated += chunk
+        self._last_chunk_time = time.monotonic()
 
         if self.strategy == FlushStrategy.PER_CHAR:
             return self._flush_all()
@@ -75,6 +85,9 @@ class FlushController:
         if self.strategy == FlushStrategy.INTERVAL and now - self._last_flush >= self.flush_interval:
             return self._flush_all()
 
+        if self._last_chunk_time > 0 and now - self._last_chunk_time >= BATCH_AFTER_GAP:
+            return self._flush_all()
+
         return []
 
     def flush_remaining(self) -> tuple[list[str], str]:
@@ -83,6 +96,19 @@ class FlushController:
         batches = self._flush_all() if self._buffer else []
         return batches, self._accumulated
 
+    def mark_flush_in_progress(self) -> bool:
+        if self._flush_in_progress:
+            self._needs_reflush = True
+            return False
+        self._flush_in_progress = True
+        return True
+
+    def mark_flush_done(self) -> bool:
+        self._flush_in_progress = False
+        needs = self._needs_reflush
+        self._needs_reflush = False
+        return needs
+
     def _flush_all(self) -> list[str]:
         if not self._buffer:
             return []
@@ -157,6 +183,8 @@ class C2CStreamingController:
                         collected += segment
                         batches = self._flush_controller.feed(segment)
                         for batch in batches:
+                            if not self._flush_controller.mark_flush_in_progress():
+                                continue
                             self._flush_controller.seq_counter += 1
                             success = await self._send_stream_chunk(
                                 chat_id,
@@ -165,6 +193,7 @@ class C2CStreamingController:
                                 self._flush_controller.seq_counter,
                                 event_id,
                             )
+                            _needs_reflush = self._flush_controller.mark_flush_done()
                             if success:
                                 self._any_chunk_delivered = True
                             else:
@@ -176,6 +205,8 @@ class C2CStreamingController:
 
             flush_batches, full = self._flush_controller.flush_remaining()
             for batch in flush_batches:
+                if not self._flush_controller.mark_flush_in_progress():
+                    continue
                 self._flush_controller.seq_counter += 1
                 success = await self._send_stream_chunk(
                     chat_id,
@@ -184,6 +215,7 @@ class C2CStreamingController:
                     self._flush_controller.seq_counter,
                     event_id,
                 )
+                _needs_reflush = self._flush_controller.mark_flush_done()
                 if success:
                     self._any_chunk_delivered = True
 

backend/package/yuxi/channels/adapters/qqbot/commands/__init__.py

@@ -2,4 +2,10 @@ from .builtin import register_builtin_commands
 from .framework import CommandContext, CommandRegistry, CommandResult
 from .streaming_cmd import register_streaming_command
 
-__all__ = ["CommandRegistry", "CommandContext", "CommandResult", "register_builtin_commands", "register_streaming_command"]
+__all__ = [
+    "CommandRegistry",
+    "CommandContext",
+    "CommandResult",
+    "register_builtin_commands",
+    "register_streaming_command",
+]

backend/package/yuxi/channels/adapters/qqbot/commands/builtin.py

@@ -46,9 +46,7 @@ async def _cmd_help(ctx: CommandContext) -> CommandResult:
         return CommandResult(success=True, message="命令系统未初始化。")
 
     is_group = getattr(ctx.msg, "chat_type", None)
-    is_group_val = str(is_group) == "ChatType.GROUP" or (
-        hasattr(is_group, "value") and is_group.value == "group"
-    )
+    is_group_val = str(is_group) == "ChatType.GROUP" or (hasattr(is_group, "value") and is_group.value == "group")
 
     lines = ["可用命令："]
     seen: set[str] = set()

backend/package/yuxi/channels/adapters/qqbot/commands/streaming_cmd.py

@@ -56,4 +56,4 @@ def register_streaming_command(registry: CommandRegistry) -> CommandRegistry:
         usage="/bot-streaming [on|off|status]",
         aliases=["streaming"],
     )
-    return registry
+    return registry

backend/package/yuxi/channels/adapters/qqbot/config_hints.py

@@ -0,0 +1,388 @@
+from __future__ import annotations
+
+QQBOT_CONFIG_HINTS: list[dict] = [
+    {
+        "key": "app_id",
+        "label": "App ID",
+        "help": "QQ 开放平台应用 ID，从「应用管理」页面获取",
+        "required": True,
+        "sensitive": False,
+        "section": "credentials",
+    },
+    {
+        "key": "app_secret",
+        "label": "App Secret",
+        "help": "QQ 开放平台客户端密钥，从「开发设置」中生成",
+        "required": True,
+        "sensitive": True,
+        "section": "credentials",
+    },
+    {
+        "key": "client_secret_file",
+        "label": "密钥文件路径",
+        "help": "存放 App Secret 的文件路径，优先于环境变量",
+        "required": False,
+        "sensitive": False,
+        "section": "credentials",
+    },
+    {
+        "key": "sandbox",
+        "label": "沙箱模式",
+        "help": "启用沙箱 API 端点 (sandbox.api.sgroup.qq.com)",
+        "required": False,
+        "default": False,
+        "type": "boolean",
+        "section": "connection",
+    },
+    {
+        "key": "bot_name",
+        "label": "Bot 名称",
+        "help": "机器人显示名称，用于 @提及匹配",
+        "required": False,
+        "section": "basic",
+    },
+    {
+        "key": "intents",
+        "label": "WebSocket Intents",
+        "help": "QQ Bot Gateway 意图位掩码，控制订阅的事件类型",
+        "required": False,
+        "type": "number",
+        "section": "connection",
+    },
+    {
+        "key": "dm_policy",
+        "label": "DM 策略",
+        "help": "私信访问策略: open(开放) / allowlist(白名单) / pairing(配对) / disabled(禁用)",
+        "required": False,
+        "default": "open",
+        "type": "enum",
+        "options": ["open", "allowlist", "pairing", "disabled"],
+        "section": "security",
+    },
+    {
+        "key": "group_policy",
+        "label": "群组策略",
+        "help": "群组访问策略: open(开放) / allowlist(白名单) / disabled(禁用)",
+        "required": False,
+        "default": "open",
+        "type": "enum",
+        "options": ["open", "allowlist", "disabled"],
+        "section": "security",
+    },
+    {
+        "key": "allow_from",
+        "label": "DM 白名单",
+        "help": "允许私信的用户 ID 列表，支持通配符 '*'",
+        "required": False,
+        "type": "list",
+        "section": "security",
+    },
+    {
+        "key": "group_allow_from",
+        "label": "群组白名单",
+        "help": "允许的群组 ID 列表，支持通配符 '*'",
+        "required": False,
+        "type": "list",
+        "section": "security",
+    },
+    {
+        "key": "group_require_mention",
+        "label": "群聊 @提及门控",
+        "help": "群聊中是否需要 @机器人 才响应",
+        "required": False,
+        "default": True,
+        "type": "boolean",
+        "section": "security",
+    },
+    {
+        "key": "pairing_enabled",
+        "label": "配对模式",
+        "help": "启用配对模式，仅已配对用户可私信",
+        "required": False,
+        "default": False,
+        "type": "boolean",
+        "section": "security",
+    },
+    {
+        "key": "group_ignore_other_mentions",
+        "label": "忽略他人 @",
+        "help": "当消息 @了其他人但不包含机器人时忽略",
+        "required": False,
+        "default": True,
+        "type": "boolean",
+        "section": "security",
+    },
+    {
+        "key": "group_command_bypass_mention",
+        "label": "命令跳过 @",
+        "help": "白名单命令可不需要 @机器人",
+        "required": False,
+        "default": True,
+        "type": "boolean",
+        "section": "security",
+    },
+    {
+        "key": "use_markdown",
+        "label": "使用 Markdown",
+        "help": "使用 Markdown 格式发送消息",
+        "required": False,
+        "default": False,
+        "type": "boolean",
+        "section": "format",
+    },
+    {
+        "key": "markdown_template_id",
+        "label": "Markdown 模板 ID",
+        "help": "QQ Bot Markdown 消息模板 ID",
+        "required": False,
+        "section": "format",
+    },
+    {
+        "key": "stream_flush_interval",
+        "label": "流式刷新间隔",
+        "help": "C2C 流式消息的刷新间隔 (秒)",
+        "required": False,
+        "default": 0.6,
+        "type": "number",
+        "section": "streaming",
+    },
+    {
+        "key": "streaming_mode",
+        "label": "流式模式",
+        "help": "C2C 流式模式: off(关闭) / block(块流式)",
+        "required": False,
+        "default": "block",
+        "type": "enum",
+        "options": ["off", "block"],
+        "section": "streaming",
+    },
+    {
+        "key": "silent",
+        "label": "静默发送",
+        "help": "发送消息时不通知用户 (no_notify)",
+        "required": False,
+        "default": False,
+        "type": "boolean",
+        "section": "format",
+    },
+    {
+        "key": "dedup_window_s",
+        "label": "去重窗口",
+        "help": "消息去重时间窗口 (秒)",
+        "required": False,
+        "default": 60,
+        "type": "number",
+        "section": "advanced",
+    },
+    {
+        "key": "auto_download_attachments",
+        "label": "自动下载附件",
+        "help": "自动下载消息中的图片/文件附件",
+        "required": False,
+        "default": True,
+        "type": "boolean",
+        "section": "advanced",
+    },
+    {
+        "key": "tts_default_voice",
+        "label": "TTS 默认语音",
+        "help": "文字转语音默认发音人",
+        "required": False,
+        "default": "zh-CN-XiaoxiaoNeural",
+        "section": "audio",
+    },
+    {
+        "key": "tts_default_format",
+        "label": "TTS 默认格式",
+        "help": "音频输出格式",
+        "required": False,
+        "default": "mp3",
+        "type": "enum",
+        "options": ["mp3", "wav", "opus"],
+        "section": "audio",
+    },
+    {
+        "key": "queue_global_limit",
+        "label": "消息队列全局上限",
+        "help": "消息队列最大容量",
+        "required": False,
+        "default": 1000,
+        "type": "number",
+        "section": "advanced",
+    },
+    {
+        "key": "queue_per_user_limit",
+        "label": "每用户队列上限",
+        "help": "每个用户的消息队列最大容量",
+        "required": False,
+        "default": 20,
+        "type": "number",
+        "section": "advanced",
+    },
+    {
+        "key": "queue_per_group_limit",
+        "label": "每群组队列上限",
+        "help": "每个群组的消息队列最大容量",
+        "required": False,
+        "default": 50,
+        "type": "number",
+        "section": "advanced",
+    },
+    {
+        "key": "queue_max_concurrency",
+        "label": "队列最大并发",
+        "help": "消息队列最大并发处理数",
+        "required": False,
+        "default": 10,
+        "type": "number",
+        "section": "advanced",
+    },
+    {
+        "key": "reconnect.max_retries",
+        "label": "重连次数",
+        "help": "WebSocket 最大重连次数",
+        "required": False,
+        "default": 10,
+        "type": "number",
+        "section": "connection",
+    },
+    {
+        "key": "reconnect.base_delay",
+        "label": "重连基础延迟",
+        "help": "重连指数退避基础延迟 (秒)",
+        "required": False,
+        "default": 1.0,
+        "type": "number",
+        "section": "connection",
+    },
+    {
+        "key": "reconnect.max_delay",
+        "label": "重连最大延迟",
+        "help": "重连指数退避最大延迟 (秒)",
+        "required": False,
+        "default": 60.0,
+        "type": "number",
+        "section": "connection",
+    },
+    {
+        "key": "reconnect.jitter",
+        "label": "重连抖动系数",
+        "help": "重连延迟随机抖动系数",
+        "required": False,
+        "default": 0.3,
+        "type": "number",
+        "section": "connection",
+    },
+    {
+        "key": "retry.attempts",
+        "label": "发送重试次数",
+        "help": "消息发送失败最大重试次数",
+        "required": False,
+        "default": 3,
+        "type": "number",
+        "section": "connection",
+    },
+    {
+        "key": "credential_backup_dir",
+        "label": "凭据备份目录",
+        "help": "凭据备份文件存储目录",
+        "required": False,
+        "section": "storage",
+    },
+    {
+        "key": "session_store_dir",
+        "label": "会话存储目录",
+        "help": "WebSocket 会话持久化目录",
+        "required": False,
+        "section": "storage",
+    },
+    {
+        "key": "known_users_max",
+        "label": "已知用户上限",
+        "help": "已知用户追踪器最大记录数",
+        "required": False,
+        "default": 10000,
+        "type": "number",
+        "section": "storage",
+    },
+    {
+        "key": "known_users_persist_dir",
+        "label": "已知用户存储目录",
+        "help": "已知用户持久化文件目录",
+        "required": False,
+        "section": "storage",
+    },
+    {
+        "key": "suppress_local_approval_prompt",
+        "label": "抑制本地审批提示",
+        "help": "防止 QQ Bot 本地二次审批提示去重",
+        "required": False,
+        "default": True,
+        "type": "boolean",
+        "section": "approval",
+    },
+    {
+        "key": "approval_native_routing",
+        "label": "审批原生路由",
+        "help": "审批请求通过 QQ Bot 内联键盘路由",
+        "required": False,
+        "default": True,
+        "type": "boolean",
+        "section": "approval",
+    },
+    {
+        "key": "plugin_version",
+        "label": "插件版本",
+        "help": "注入 HTTP User-Agent 的插件版本号",
+        "required": False,
+        "default": "1.0.0",
+        "section": "advanced",
+    },
+    {
+        "key": "openclaw_version",
+        "label": "OpenClaw 版本",
+        "help": "注入 HTTP User-Agent 的平台版本号",
+        "required": False,
+        "default": "1.0.0",
+        "section": "advanced",
+    },
+    {
+        "key": "message_cache_max",
+        "label": "消息缓存上限",
+        "help": "发送消息缓存最大记录数",
+        "required": False,
+        "default": 500,
+        "type": "number",
+        "section": "advanced",
+    },
+    {
+        "key": "message_cache_ttl_s",
+        "label": "消息缓存 TTL",
+        "help": "发送消息缓存过期时间 (秒)",
+        "required": False,
+        "default": 3600,
+        "type": "number",
+        "section": "advanced",
+    },
+]
+
+
+def get_config_hints(section: str = "") -> list[dict]:
+    if section:
+        return [h for h in QQBOT_CONFIG_HINTS if h.get("section") == section]
+    return list(QQBOT_CONFIG_HINTS)
+
+
+def get_config_sections() -> list[str]:
+    seen: set[str] = set()
+    result: list[str] = []
+    for hint in QQBOT_CONFIG_HINTS:
+        section = hint.get("section", "")
+        if section and section not in seen:
+            seen.add(section)
+            result.append(section)
+    return result
+
+
+def get_required_config_keys() -> list[str]:
+    return [h["key"] for h in QQBOT_CONFIG_HINTS if h.get("required")]

backend/package/yuxi/channels/adapters/qqbot/credential_backup.py

@@ -14,21 +14,55 @@ logger = logging.getLogger(__name__)
 DEFAULT_BACKUP_DIR = os.path.join(tempfile.gettempdir(), "yuxi_qqbot_credentials")
 
 
+def _resolve_backup_dir(backup_dir: str | None = None) -> str:
+    if backup_dir:
+        return backup_dir
+    for env_var in ("OPENCLAW_STATE_DIR", "CLAWDBOT_STATE_DIR"):
+        env_val = os.environ.get(env_var)
+        if env_val:
+            qqbot_dir = os.path.join(env_val, "qqbot", "credentials")
+            os.makedirs(qqbot_dir, exist_ok=True)
+            return qqbot_dir
+    return DEFAULT_BACKUP_DIR
+
+
+_fernet_warned = False
+
+
 def _get_fernet():
+    global _fernet_warned
+
     try:
         from cryptography.fernet import Fernet
     except ImportError:
+        if not _fernet_warned:
+            logger.warning(
+                "QQBot credential encryption unavailable: cryptography not installed. "
+                "Credentials will be stored in plaintext. "
+                "Install with: pip install cryptography"
+            )
+            _fernet_warned = True
         return None
 
     key = os.environ.get("QQBOT_CREDENTIAL_ENCRYPTION_KEY")
     if not key:
+        if not _fernet_warned:
+            logger.warning(
+                "QQBot credential encryption disabled: QQBOT_CREDENTIAL_ENCRYPTION_KEY not set. "
+                "Credentials will be stored in plaintext at %s",
+                _resolve_backup_dir(),
+            )
+            _fernet_warned = True
         return None
     digest = hashlib.sha256(key.encode()).digest()
     return Fernet(base64.urlsafe_b64encode(digest))
 
 
 def _encrypt_field(fernet, value: str) -> str:
-    if not fernet or not value:
+    if not value:
+        return value
+    if not fernet:
+        logger.warning("QQBot credential stored in plaintext: encryption key not configured")
         return value
     return fernet.encrypt(value.encode()).decode()
 
@@ -65,7 +99,7 @@ class CredentialSnapshot:
 class CredentialBackup:
     def __init__(self, app_id: str, backup_dir: str | None = None):
         self._app_id = app_id
-        self._backup_dir = backup_dir or DEFAULT_BACKUP_DIR
+        self._backup_dir = _resolve_backup_dir(backup_dir)
         self._backup_path = os.path.join(self._backup_dir, f"{app_id}.json")
 
     def save(self, snapshot: CredentialSnapshot) -> bool:

backend/package/yuxi/channels/adapters/qqbot/format.py

@@ -4,9 +4,15 @@ from yuxi.channels.models import ChannelResponse
 
 from .constants import DM_CHAT_PREFIX, GROUP_CHAT_PREFIX
 
+TEXT_CONTENT_LIMIT = 2000
+MARKDOWN_CONTENT_LIMIT = 4096
+EMBED_DESCRIPTION_LIMIT = 4096
+EMBED_PROMPT_LIMIT = 200
+EMBED_TITLE_LIMIT = 200
+
 
 def build_text_payload(response: ChannelResponse) -> dict:
-    content = response.content[:2000]
+    content = response.content[:TEXT_CONTENT_LIMIT]
     chat_id = response.identity.channel_chat_id
 
     payload: dict = {"content": content}
@@ -34,7 +40,7 @@ def build_markdown_payload(response: ChannelResponse, template_id: str | None =
                 "template_id": md_template_id,
                 "params": [
                     {"key": "title", "values": [response.metadata.get("title", "")]},
-                    {"key": "content", "values": [response.content[:4096]]},
+                    {"key": "content", "values": [response.content[:MARKDOWN_CONTENT_LIMIT]]},
                 ],
             },
         }
@@ -42,7 +48,7 @@ def build_markdown_payload(response: ChannelResponse, template_id: str | None =
     return {
         "msg_type": 2,
         "markdown": {
-            "content": response.content[:4096],
+            "content": response.content[:MARKDOWN_CONTENT_LIMIT],
         },
     }
 
@@ -66,9 +72,9 @@ def build_embed_payload(response: ChannelResponse) -> dict:
     return {
         "msg_type": 4,
         "embed": {
-            "title": embed_data.get("title", ""),
-            "description": response.content[:4096],
-            "prompt": embed_data.get("prompt", response.content[:200]),
+            "title": embed_data.get("title", "")[:EMBED_TITLE_LIMIT],
+            "description": response.content[:EMBED_DESCRIPTION_LIMIT],
+            "prompt": embed_data.get("prompt", response.content[:EMBED_PROMPT_LIMIT]),
             "fields": embed_data.get("fields", []),
         },
     }
@@ -82,7 +88,7 @@ def build_media_generic_payload(response: ChannelResponse, file_id: str, msg_typ
     }
 
     if response.content:
-        payload["content"] = response.content[:2000]
+        payload["content"] = response.content[:TEXT_CONTENT_LIMIT]
 
     if chat_id.startswith(GROUP_CHAT_PREFIX):
         payload["group_openid"] = chat_id.replace(GROUP_CHAT_PREFIX, "")
@@ -96,10 +102,12 @@ def build_image_payload(response: ChannelResponse, file_id: str) -> dict:
     chat_id = response.identity.channel_chat_id
     payload: dict = {
         "msg_type": 1,
-        "image": file_id,
-        "content": response.content[:2000] if response.content else "",
+        "file_image": file_id,
     }
 
+    if response.content:
+        payload["content"] = response.content[:TEXT_CONTENT_LIMIT]
+
     if chat_id.startswith(GROUP_CHAT_PREFIX):
         payload["group_openid"] = chat_id.replace(GROUP_CHAT_PREFIX, "")
     elif not chat_id.startswith(DM_CHAT_PREFIX):
@@ -114,12 +122,15 @@ def format_outbound(
     markdown_template_id: str | None = None,
 ) -> dict:
     msg_type = response.metadata.get("qq_msg_type", "")
+    chat_id = response.identity.channel_chat_id
 
     if msg_type == "markdown" or (use_markdown and not msg_type):
         return build_markdown_payload(response, markdown_template_id)
     elif msg_type == "ark" and response.metadata.get("ark_template_id"):
         return build_ark_payload(response)
     elif msg_type == "embed":
+        if chat_id.startswith(DM_CHAT_PREFIX) or chat_id.startswith(GROUP_CHAT_PREFIX):
+            return build_markdown_payload(response, markdown_template_id)
         return build_embed_payload(response)
     elif msg_type == "image" and response.attachments:
         file_id = response.attachments[0].file_id or response.attachments[0].url or ""
@@ -137,7 +148,7 @@ def format_outbound(
 
 
 class MarkdownChunker:
-    MAX_CHARS = 5000
+    MAX_CHARS = 4096
     CHUNK_OVERLAP = 200
 
     def __init__(self, max_chars: int = MAX_CHARS, chunk_overlap: int = CHUNK_OVERLAP):

backend/package/yuxi/channels/adapters/qqbot/group_buffer.py

@@ -17,7 +17,7 @@ class GroupMessage:
 
 @dataclass
 class GroupSession:
-    group_id: str
+    group_id: str = ""
     messages: list[GroupMessage] = field(default_factory=list)
     last_active: float = 0.0
     buffer_limit: int = 50
@@ -43,6 +43,16 @@ class GroupHistoryBuffer:
         self._sessions: dict[str, GroupSession] = defaultdict(GroupSession)
         self._buffer_limit = buffer_limit
         self._ttl_seconds = ttl_seconds
+        self._group_activations: dict[str, str] = {}
+
+    def set_group_activation(self, group_id: str, activation: str) -> None:
+        self._group_activations[group_id] = activation
+
+    def resolve_group_activation(self, group_id: str) -> str:
+        return self._group_activations.get(group_id, "")
+
+    def get_all_activations(self) -> dict[str, str]:
+        return dict(self._group_activations)
 
     def record(self, group_id: str, msg: GroupMessage) -> None:
         session = self._sessions[group_id]

backend/package/yuxi/channels/adapters/qqbot/inbound_pipeline.py

@@ -159,11 +159,10 @@ class QQBotInboundPipeline(BaseInboundPipeline):
 
         adapter = pipeline.adapter
         dedup_cache = getattr(adapter, "_recent_msg_ids", None)
-        dedup_window = getattr(adapter, "_dedup_window_s", 60)
 
         if dedup_cache is not None:
             now = time.monotonic()
-            self._prune_dedup_cache(dedup_cache, dedup_window, now)
+            adapter._prune_old_msg_ids(now)
             if msg_id in dedup_cache:
                 logger.debug("Dedup: %s already processed, skipping", msg_id)
                 ctx.stop("dedup_duplicate")
@@ -172,12 +171,6 @@ class QQBotInboundPipeline(BaseInboundPipeline):
 
         return ctx
 
-    @staticmethod
-    def _prune_dedup_cache(cache: dict[str, float], window: int, now: float) -> None:
-        expired = [mid for mid, ts in cache.items() if now - ts >= window]
-        for mid in expired:
-            del cache[mid]
-
     @staticmethod
     def _buffer_group_message(pipeline: BaseInboundPipeline, event_data: dict, ctx: PipelineContext) -> None:
         adapter = pipeline.adapter
@@ -242,7 +235,6 @@ class QQBotInboundPipeline(BaseInboundPipeline):
             ctx.metadata["group_openid"] = event_data.get("group_openid", "")
             ctx.metadata["bot_mentioned"] = False
             ctx.stop("group_message_no_mention")
-            self._buffer_group_message(pipeline, event_data, ctx)
             return None
 
         elif event_type == "INTERACTION_CREATE":

backend/package/yuxi/channels/adapters/qqbot/interaction.py

@@ -42,12 +42,12 @@ class InteractionRegistry:
         if not ctx.interaction_id:
             return False
 
-        for action_id, callback in self._callbacks.items():
-            if ctx.interaction_id == action_id or ctx.interaction_id.startswith(action_id):
-                try:
-                    return await callback(ctx)
-                except Exception:
-                    logger.exception("InteractionRegistry: callback failed for %s", action_id)
+        callback = self._callbacks.get(ctx.interaction_id)
+        if callback:
+            try:
+                return await callback(ctx)
+            except Exception:
+                logger.exception("InteractionRegistry: callback failed for %s", ctx.interaction_id)
         return False
 
 

backend/package/yuxi/channels/adapters/qqbot/known_users.py

@@ -14,6 +14,18 @@ logger = logging.getLogger(__name__)
 DEFAULT_PERSIST_DIR = os.path.join(tempfile.gettempdir(), "yuxi_qqbot_known_users")
 
 
+def _resolve_known_users_dir(persist_dir: str | None = None) -> str:
+    if persist_dir:
+        return persist_dir
+    for env_var in ("OPENCLAW_STATE_DIR", "CLAWDBOT_STATE_DIR"):
+        env_val = os.environ.get(env_var)
+        if env_val:
+            qqbot_dir = os.path.join(env_val, "qqbot", "known_users")
+            os.makedirs(qqbot_dir, exist_ok=True)
+            return qqbot_dir
+    return DEFAULT_PERSIST_DIR
+
+
 @dataclass
 class UserRecord:
     user_id: str
@@ -66,10 +78,10 @@ class KnownUserTracker:
     ):
         self._app_id = app_id
         self._max_users = max_users
-        self._persist_dir = persist_dir or DEFAULT_PERSIST_DIR
+        self._persist_dir = _resolve_known_users_dir(persist_dir)
         self._persist_path = os.path.join(self._persist_dir, f"{app_id}_users.json")
         self._persist_interval = persist_interval_s
-        self._lock = threading.Lock()
+        self._lock = threading.RLock()
         self._users: OrderedDict[str, UserRecord] = OrderedDict()
         self._last_persist: float = 0
         self._dirty = False
@@ -178,3 +190,56 @@ class KnownUserTracker:
     def get_recent_users(self, limit: int = 50) -> list[UserRecord]:
         with self._lock:
             return list(reversed(self._users.values()))[:limit]
+
+    def list_known_users(self) -> list[dict]:
+        with self._lock:
+            return [u.to_dict() for u in self._users.values()]
+
+    def get_known_users_stats(self) -> dict:
+        now = time.time()
+        active_24h = 0
+        active_7d = 0
+        dm_count = 0
+        group_count = 0
+        with self._lock:
+            for u in self._users.values():
+                if now - u.last_seen <= 86400:
+                    active_24h += 1
+                if now - u.last_seen <= 604800:
+                    active_7d += 1
+                if "direct" in u.chat_types:
+                    dm_count += 1
+                if "group" in u.chat_types:
+                    group_count += 1
+        return {
+            "total": len(self._users),
+            "active_24h": active_24h,
+            "active_7d": active_7d,
+            "dm_users": dm_count,
+            "group_users": group_count,
+        }
+
+    def get_user_groups(self, user_id: str) -> list[str]:
+        record = self.get(user_id)
+        if record is None:
+            return []
+        return sorted(record.chat_types)
+
+    def get_group_members(self) -> dict[str, list[str]]:
+        result: dict[str, list[str]] = {}
+        with self._lock:
+            for u in self._users.values():
+                for ct in u.chat_types:
+                    if ct not in result:
+                        result[ct] = []
+                    result[ct].append(u.user_id)
+        return result
+
+    def flush_known_users(self) -> int:
+        count = 0
+        with self._lock:
+            count = len(self._users)
+            self._users.clear()
+            self._dirty = True
+            self.persist()
+        return count

backend/package/yuxi/channels/adapters/qqbot/media_upload.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+import asyncio
+
 import aiohttp
+from cachetools import LRUCache
 
 from yuxi.channels.exceptions import DeliveryFailedError
 from yuxi.utils.logging_config import logger
@@ -47,13 +50,25 @@ async def upload_media(
     url = f"{api_base}/v2/groups/{group_openid}/files" if group_openid else f"{api_base}/v2/users/@me/files"
 
     async def _do_upload(client: aiohttp.ClientSession) -> str:
-        async with client.post(url, headers=headers, data=form) as resp:
-            if resp.status != 200:
-                raise DeliveryFailedError(f"Media upload failed: HTTP {resp.status}")
-            result = await resp.json()
-            file_id = result.get("file_uuid", "") or result.get("file_info", "")
-            logger.debug(f"[QQBot] Media uploaded, file_type={file_type}, file_id={file_id}")
-            return file_id
+        last_error = None
+        for attempt in range(3):
+            try:
+                async with client.post(url, headers=headers, data=form) as resp:
+                    if resp.status != 200:
+                        raise DeliveryFailedError(f"Media upload failed: HTTP {resp.status}")
+                    result = await resp.json()
+                    file_id = result.get("file_uuid", "") or result.get("file_info", "")
+                    logger.debug(f"[QQBot] Media uploaded, file_type={file_type}, file_id={file_id}")
+                    return file_id
+            except DeliveryFailedError:
+                raise
+            except Exception as e:
+                last_error = e
+                if attempt < 2:
+                    delay = 2**attempt
+                    logger.warning("[QQBot] Media upload retry %d/%d after %.1fs: %s", attempt + 1, 3, delay, e)
+                    await asyncio.sleep(delay)
+        raise DeliveryFailedError(f"Media upload failed after 3 attempts: {last_error}")
 
     if http_client:
         return await _do_upload(http_client)
@@ -119,7 +134,7 @@ def build_media_payload(
         "msg_type": msg_type,
     }
     if msg_type == 1:
-        payload["image"] = file_id
+        payload["file_image"] = file_id
     elif msg_type == 4:
         payload["file"] = file_id
     else:
@@ -146,7 +161,7 @@ async def download_image(
 
 
 _CHUNK_SIZE = 5 * 1024 * 1024
-_UPLOAD_CACHE: dict[str, str] = {}
+_UPLOAD_CACHE = LRUCache(maxsize=1000)
 
 
 def _make_cache_key(data: bytes) -> str:

backend/package/yuxi/channels/adapters/qqbot/message_cache.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import logging
+import time
+from collections import OrderedDict
+from dataclasses import dataclass, field
+
+from yuxi.channels.models import DeliveryResult
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class SentMessage:
+    message_id: str
+    chat_id: str
+    content: str = ""
+    sent_at: float = field(default_factory=time.time)
+    updated_at: float = 0.0
+    update_count: int = 0
+    status: str = "sent"
+    error: str = ""
+
+
+class MessageCache:
+    def __init__(self, max_messages: int = 500, ttl_s: float = 3600.0):
+        self._max = max_messages
+        self._ttl = ttl_s
+        self._messages: OrderedDict[str, SentMessage] = OrderedDict()
+
+    def record_sent(self, result: DeliveryResult, chat_id: str, content: str = "") -> SentMessage | None:
+        if not result.message_id:
+            return None
+
+        msg = SentMessage(
+            message_id=result.message_id,
+            chat_id=chat_id,
+            content=content[:500],
+            status="sent" if result.success else "failed",
+            error=result.error or "",
+        )
+        self._messages[result.message_id] = msg
+        self._messages.move_to_end(result.message_id)
+
+        while len(self._messages) > self._max:
+            self._messages.popitem(last=False)
+
+        return msg
+
+    def record_update(self, message_id: str, content: str = "") -> SentMessage | None:
+        msg = self._messages.get(message_id)
+        if msg is None:
+            return None
+        msg.updated_at = time.time()
+        msg.update_count += 1
+        if content:
+            msg.content = content[:500]
+        return msg
+
+    def get(self, message_id: str) -> SentMessage | None:
+        return self._messages.get(message_id)
+
+    def get_by_chat(self, chat_id: str) -> list[SentMessage]:
+        return [m for m in self._messages.values() if m.chat_id == chat_id]
+
+    def cleanup_expired(self) -> int:
+        now = time.time()
+        expired = [mid for mid, msg in self._messages.items() if now - msg.sent_at > self._ttl]
+        for mid in expired:
+            del self._messages[mid]
+        return len(expired)
+
+    @property
+    def snapshot(self) -> dict:
+        return {
+            "total": len(self._messages),
+            "max": self._max,
+            "ttl_s": self._ttl,
+            "recent": [
+                {
+                    "message_id": m.message_id,
+                    "chat_id": m.chat_id,
+                    "sent_at": m.sent_at,
+                    "updates": m.update_count,
+                    "status": m.status,
+                }
+                for m in list(self._messages.values())[-20:]
+            ],
+        }
+
+    def clear(self) -> int:
+        count = len(self._messages)
+        self._messages.clear()
+        return count

backend/package/yuxi/channels/adapters/qqbot/message_queue.py

@@ -114,4 +114,4 @@ class MessageQueue:
                     break
             self._per_user_counts.clear()
             self._per_group_counts.clear()
-        return messages
+        return messages

backend/package/yuxi/channels/adapters/qqbot/rate_limiter.py

@@ -77,7 +77,7 @@ class RouteRateLimiter:
     def _get_or_create_bucket_sync(self, route: str) -> TokenBucket:
         if route not in self._buckets:
             rate, burst = self.defaults.get(route, self.defaults["default"])
-            self._buckets[route] = TokenBucket(rate=rate, burst=burst)
+            self._buckets.setdefault(route, TokenBucket(rate=rate, burst=burst))
         return self._buckets[route]
 
     def get_stats(self) -> dict[str, dict]:

backend/package/yuxi/channels/adapters/qqbot/reconnect.py

@@ -280,8 +280,8 @@ class QQBotReconnectManager:
             return base * 1.5
         return base * 2.0
 
-    def _check_rapid_disconnect(self, code: int | None, category: CloseCodeCategory, now: float) -> None:
-        if category == CloseCodeCategory.FATAL:
+    def _check_rapid_disconnect(self, code: int | None, category: CloseCodeCategory | None, now: float) -> None:
+        if category is None or category == CloseCodeCategory.FATAL:
             return
         if self._last_connect_time == 0:
             return

backend/package/yuxi/channels/adapters/qqbot/send.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import re
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
 
@@ -64,6 +65,7 @@ async def send_with_retry(
     config: dict | None = None,
     token_refresh_cb: Callable[[], Awaitable[str]] | None = None,
     on_sent: Callable[[DeliveryResult], Awaitable[None]] | None = None,
+    sender_headers: dict | None = None,
 ) -> DeliveryResult:
     cfg = config or {}
     max_retries = cfg.get("retry", {}).get("attempts", 3)
@@ -77,7 +79,7 @@ async def send_with_retry(
     max_rate_limit_retries = cfg.get("retry", {}).get("max_rate_limit_retries", 3)
 
     last_error = None
-    url = _resolve_send_url(api_base, chat_id)
+    url = resolve_send_url(api_base, chat_id)
 
     for attempt in range(max_retries):
         try:
@@ -85,6 +87,8 @@ async def send_with_retry(
                 "Authorization": f"QQBot {current_token}",
                 "Content-Type": "application/json",
             }
+            if sender_headers:
+                headers.update(sender_headers)
             async with http_client.post(url, json=payload, headers=headers) as resp:
                 if resp.status == 200:
                     data = await resp.json()
@@ -156,7 +160,33 @@ async def send_with_retry(
     return result
 
 
-def _resolve_send_url(api_base: str, chat_id: str) -> str:
+_ALLOWED_API_HOSTS = frozenset({"api.sgroup.qq.com", "sandbox.api.sgroup.qq.com"})
+_CHAT_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$")
+
+
+def _validate_api_base(api_base: str) -> None:
+    from urllib.parse import urlparse
+
+    parsed = urlparse(api_base)
+    host = parsed.hostname
+    if host not in _ALLOWED_API_HOSTS:
+        raise DeliveryFailedError(f"Invalid api_base host: {host}. Allowed: {sorted(_ALLOWED_API_HOSTS)}")
+
+
+def _validate_chat_id(chat_id: str) -> None:
+    raw = chat_id
+    if raw.startswith(GROUP_CHAT_PREFIX):
+        raw = raw[len(GROUP_CHAT_PREFIX) :]
+    elif raw.startswith(DM_CHAT_PREFIX):
+        raw = raw[len(DM_CHAT_PREFIX) :]
+    if raw and not _CHAT_ID_PATTERN.match(raw):
+        raise DeliveryFailedError(f"Invalid chat_id format: {chat_id}")
+
+
+def resolve_send_url(api_base: str, chat_id: str) -> str:
+    _validate_api_base(api_base)
+    _validate_chat_id(chat_id)
+
     if chat_id.startswith(GROUP_CHAT_PREFIX):
         group_openid = chat_id.replace(GROUP_CHAT_PREFIX, "")
         return f"{api_base}/v2/groups/{group_openid}/messages"

backend/package/yuxi/channels/adapters/qqbot/sender_init.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+
+logger = logging.getLogger(__name__)
+
+PLUGIN_VERSION = "1.0.0"
+OPENCLAW_VERSION = "1.0.0"
+
+
+@dataclass
+class SenderInfo:
+    plugin_version: str = PLUGIN_VERSION
+    openclaw_version: str = OPENCLAW_VERSION
+    channel_id: str = "qqbot"
+    user_agent: str = ""
+
+    def __post_init__(self):
+        self.user_agent = f"openclaw-qqbot/{self.plugin_version} openclaw/{self.openclaw_version}"
+
+
+def init_sender(
+    plugin_version: str = PLUGIN_VERSION,
+    openclaw_version: str = OPENCLAW_VERSION,
+) -> SenderInfo:
+    info = SenderInfo(
+        plugin_version=plugin_version,
+        openclaw_version=openclaw_version,
+    )
+    logger.info(
+        "Sender initialized: plugin=%s openclaw=%s ua=%s",
+        info.plugin_version,
+        info.openclaw_version,
+        info.user_agent,
+    )
+    return info
+
+
+def build_sender_headers(info: SenderInfo, extra: dict | None = None) -> dict:
+    headers = {
+        "User-Agent": info.user_agent,
+        "X-Plugin-Version": info.plugin_version,
+        "X-OpenClaw-Version": info.openclaw_version,
+    }
+    if extra:
+        headers.update(extra)
+    return headers

backend/package/yuxi/channels/adapters/qqbot/session_store.py

@@ -12,6 +12,27 @@ logger = logging.getLogger(__name__)
 DEFAULT_SESSION_DIR = os.path.join(tempfile.gettempdir(), "yuxi_qqbot_sessions")
 
 
+def _resolve_state_dir(store_dir: str | None = None) -> str:
+    if store_dir:
+        return store_dir
+    for env_var in ("OPENCLAW_STATE_DIR", "CLAWDBOT_STATE_DIR"):
+        env_val = os.environ.get(env_var)
+        if env_val:
+            qqbot_dir = os.path.join(env_val, "qqbot")
+            os.makedirs(qqbot_dir, exist_ok=True)
+            return qqbot_dir
+    return DEFAULT_SESSION_DIR
+
+
+@dataclass
+class SessionState:
+    group_openid: str = ""
+    group_activation: str = ""
+    require_mention: bool = True
+    agent_id: str = ""
+    last_active: float = 0.0
+
+
 @dataclass
 class SessionRecord:
     session_id: str = ""
@@ -50,7 +71,7 @@ class SessionRecord:
 class SessionStore:
     def __init__(self, app_id: str, store_dir: str | None = None):
         self._app_id = app_id
-        self._store_dir = store_dir or DEFAULT_SESSION_DIR
+        self._store_dir = _resolve_state_dir(store_dir)
         self._store_path = os.path.join(self._store_dir, f"{app_id}_session.json")
 
     def save(self, record: SessionRecord) -> bool:
@@ -96,9 +117,39 @@ class SessionStore:
             logger.exception("SessionStore: failed to clear session")
             return False
 
+    def update_last_seq(self, session_id: str, seq: int) -> bool:
+        record = self.load()
+        if record is None:
+            record = SessionRecord()
+        record.session_id = session_id
+        record.last_seq = seq
+        record.last_heartbeat = time.monotonic()
+        return self.save(record)
+
+    @staticmethod
+    def get_all_sessions(store_dir: str | None = None) -> list[dict]:
+        directory = _resolve_state_dir(store_dir)
+        if not os.path.exists(directory):
+            return []
+        sessions: list[dict] = []
+        try:
+            for filename in os.listdir(directory):
+                if not filename.endswith("_session.json"):
+                    continue
+                filepath = os.path.join(directory, filename)
+                try:
+                    with open(filepath, encoding="utf-8") as f:
+                        data = json.load(f)
+                    sessions.append(data)
+                except (OSError, json.JSONDecodeError):
+                    pass
+        except OSError:
+            logger.exception("SessionStore: failed to list sessions")
+        return sessions
+
     @staticmethod
     def cleanup_expired(store_dir: str | None = None, max_age_s: float = 86400 * 7) -> int:
-        directory = store_dir or DEFAULT_SESSION_DIR
+        directory = _resolve_state_dir(store_dir)
         if not os.path.exists(directory):
             return 0
 

backend/package/yuxi/channels/adapters/qqbot/setup_wizard.py

@@ -13,7 +13,9 @@ logger = logging.getLogger(__name__)
 
 class WizardStep(Enum):
     WELCOME = auto()
+    SETUP_METHOD = auto()
     APP_CREDENTIALS = auto()
+    QR_LINK = auto()
     INTENTS = auto()
     PERMISSIONS = auto()
     WEBHOOK_URL = auto()
@@ -25,11 +27,14 @@ class WizardStep(Enum):
 @dataclass
 class WizardState:
     step: WizardStep = WizardStep.WELCOME
+    setup_method: str = ""
     app_id: str = ""
     app_secret: str = ""
     bot_token: str = ""
     intents: list[str] = None
     webhook_url: str = ""
+    qr_url: str = ""
+    qr_session_id: str = ""
     verify_result: dict | None = None
     started_at: float = 0.0
 
@@ -209,3 +214,49 @@ async def generate_config_yaml(app_id: str, app_secret: str, intents: list[int]
         lines.append(f"#   {mark} {intent_id}: {name} ({desc})")
 
     return "\n".join(lines)
+
+
+async def link_via_qr_code(app_id: str) -> dict:
+    import uuid
+
+    session_id = uuid.uuid4().hex[:16]
+    api_base = "https://api.sgroup.qq.com"
+
+    qr_url = f"https://q.qq.com/qqbot/link?app_id={app_id}&session_id={session_id}&t={int(time.time())}"
+
+    return {
+        "success": True,
+        "session_id": session_id,
+        "qr_url": qr_url,
+        "message": "请使用 QQ 扫描二维码完成 Bot 绑定",
+        "api_base": api_base,
+        "poll_url": f"{api_base}/oauth2/qr_poll?session_id={session_id}",
+    }
+
+
+async def check_qr_link_status(session_id: str, app_id: str) -> dict:
+    import aiohttp
+
+    api_base = "https://api.sgroup.qq.com"
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(
+                f"{api_base}/oauth2/qr_poll",
+                params={"session_id": session_id, "app_id": app_id},
+            ) as resp:
+                if resp.status == 200:
+                    data = await resp.json()
+                    status = data.get("status", "pending")
+                    if status == "confirmed":
+                        return {
+                            "success": True,
+                            "status": "confirmed",
+                            "app_id": app_id,
+                            "data": data,
+                        }
+                    elif status == "expired":
+                        return {"success": False, "status": "expired", "error": "QR code expired"}
+                    return {"success": True, "status": "pending"}
+                return {"success": False, "error": f"HTTP {resp.status}"}
+    except Exception as e:
+        return {"success": False, "error": str(e)}

backend/package/yuxi/channels/adapters/qqbot/streaming.py

@@ -65,9 +65,7 @@ async def stream_content(
                 yield para
 
             if c2c_ctrl is not None:
-                c2c_batches = c2c_ctrl.flush_controller.feed(chunk)
-                for batch in c2c_batches:
-                    pass
+                c2c_ctrl.flush_controller.feed(chunk)
 
             await asyncio.sleep(0)
 
@@ -201,9 +199,7 @@ async def stream_with_media_handling(
                 yield para
 
             if c2c_ctrl is not None:
-                c2c_batches = c2c_ctrl.flush_controller.feed(chunk)
-                for _batch in c2c_batches:
-                    pass
+                c2c_ctrl.flush_controller.feed(chunk)
 
             await asyncio.sleep(0)
 

backend/package/yuxi/channels/adapters/qqbot/thread_simulator.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from collections import OrderedDict
+from datetime import datetime
+from typing import Any
+
+from yuxi.channels.models import HistoricalMessage
+
+
+class QQBotThreadSimulator:
+    def __init__(self, window_size: int = 20):
+        self._reply_chains: dict[str, list[str]] = {}
+        self._context_windows: OrderedDict[str, list[dict[str, Any]]] = OrderedDict()
+        self._window_size = window_size
+
+    def build_thread_context(self, message: dict, session_key: str) -> str:
+        window = self._context_windows.get(session_key)
+        if window is None:
+            window = []
+            self._context_windows[session_key] = window
+
+        window.append(
+            {
+                "id": message.get("id", ""),
+                "author": message.get("author", {}).get("username", "unknown"),
+                "content": message.get("content", ""),
+                "timestamp": message.get("timestamp", datetime.now()),
+            }
+        )
+
+        if len(window) > self._window_size:
+            window.pop(0)
+
+        if len(self._context_windows) > 500:
+            oldest = next(iter(self._context_windows))
+            del self._context_windows[oldest]
+
+        context_lines = [f"{msg['author']}: {msg['content']}" for msg in window[:-1]]
+        return "\n".join(context_lines)
+
+    def resolve_simulated_thread_id(self, message: dict) -> str:
+        msg_type = message.get("message_type", "c2c")
+
+        if msg_type == "c2c":
+            openid = message.get("author", {}).get("id", "unknown")
+            seq = message.get("seq", 0)
+            return f"qqbot:c2c:{openid}:seq:{seq // 10}"
+
+        group_id = message.get("group_id", "unknown")
+        seq = message.get("seq", 0)
+        return f"qqbot:group:{group_id}:seq:{seq // 10}"
+
+    def get_context_as_history(self, session_key: str) -> list[HistoricalMessage]:
+        window = self._context_windows.get(session_key, [])
+        return [
+            HistoricalMessage(
+                message_id=msg.get("id", ""),
+                sender_id=msg.get("author", ""),
+                sender_name=msg.get("author", "unknown"),
+                content=msg.get("content", ""),
+                timestamp=msg.get("timestamp", datetime.now()),
+            )
+            for msg in window
+        ]

backend/package/yuxi/channels/adapters/qqbot/token.py

@@ -83,7 +83,7 @@ class QQBotTokenManager:
                 data = await resp.json()
                 self._access_token = data["access_token"]
                 expires_in = data.get("expires_in", 7200)
-                self._expires_at = time.monotonic() + expires_in
+                self._expires_at = time.time() + expires_in
                 logger.info(f"[QQBot] Token refreshed, expires in {expires_in}s (app_id={self.app_id[:6]}...)")
         finally:
             if not self._http_client:
@@ -106,7 +106,7 @@ class QQBotTokenManager:
             try:
                 sleep_duration = self._refresh_interval
                 if self._expires_at is not None:
-                    remaining = self._expires_at - time.monotonic() - 300
+                    remaining = self._expires_at - time.time() - 300
                     sleep_duration = max(60.0, min(self._refresh_interval, max(remaining, 0.0)))
 
                 await asyncio.sleep(sleep_duration)
@@ -127,4 +127,4 @@ class QQBotTokenManager:
     def _is_expired(self) -> bool:
         if self._access_token is None or self._expires_at is None:
             return True
-        return time.monotonic() > self._expires_at - 300
+        return time.time() > self._expires_at - 300

backend/package/yuxi/channels/adapters/qqbot/typing_keepalive.py

@@ -52,4 +52,4 @@ class TypingKeepAlive:
                 except Exception:
                     logger.debug("TypingKeepAlive: notify failed for %s", chat_id, exc_info=True)
         except asyncio.CancelledError:
-            pass
+            pass

backend/package/yuxi/channels/adapters/qqbot/voice_send.py

@@ -5,13 +5,14 @@ import aiohttp
 from yuxi.channels.models import DeliveryResult
 from yuxi.utils.logging_config import logger
 
-from .constants import DM_CHAT_PREFIX, GROUP_CHAT_PREFIX
+from .constants import GROUP_CHAT_PREFIX
 from .media_upload import (
     FILE_TYPE_VOICE,
     build_media_payload,
     upload_media,
     validate_media_size,
 )
+from .send import resolve_send_url
 
 
 async def send_voice(
@@ -43,7 +44,7 @@ async def send_voice(
 
     payload = build_media_payload(chat_id, file_id, msg_type=7)
 
-    url = _resolve_media_send_url(api_base, chat_id)
+    url = resolve_send_url(api_base, chat_id)
     headers = {
         "Authorization": f"QQBot {token}",
         "Content-Type": "application/json",
@@ -94,7 +95,7 @@ async def send_video(
 
     payload = build_media_payload(chat_id, file_id, msg_type=7)
 
-    url = _resolve_media_send_url(api_base, chat_id)
+    url = resolve_send_url(api_base, chat_id)
     headers = {
         "Authorization": f"QQBot {token}",
         "Content-Type": "application/json",
@@ -145,7 +146,7 @@ async def send_file(
 
     payload = build_media_payload(chat_id, file_id, msg_type=7)
 
-    url = _resolve_media_send_url(api_base, chat_id)
+    url = resolve_send_url(api_base, chat_id)
     headers = {
         "Authorization": f"QQBot {token}",
         "Content-Type": "application/json",
@@ -163,15 +164,3 @@ async def send_file(
     except Exception as e:
         logger.error(f"[QQBot] File send error: {e}")
         return DeliveryResult(success=False, error=str(e))
-
-
-def _resolve_media_send_url(api_base: str, chat_id: str) -> str:
-    if chat_id.startswith(GROUP_CHAT_PREFIX):
-        group_openid = chat_id.replace(GROUP_CHAT_PREFIX, "")
-        return f"{api_base}/v2/groups/{group_openid}/messages"
-    elif chat_id.startswith(DM_CHAT_PREFIX):
-        openid = chat_id.replace(DM_CHAT_PREFIX, "")
-        return f"{api_base}/v2/users/{openid}/messages"
-    elif chat_id:
-        return f"{api_base}/v2/channels/{chat_id}/messages"
-    return f"{api_base}/v2/users/@me/messages"