chore: bump version to 4.9.4 in pyproject.toml and __init__.py

Backend serializes monitoring timestamps as naive ISO strings without
timezone designator. JavaScript's new Date() treats such strings as
local time, causing displayed times to be off by the user's UTC offset.
Add parseUTCTimestamp() utility that appends 'Z' to ensure correct UTC
interpretation.

README_CN.md

@@ -34,8 +34,6 @@
 
 ---
 
-## 什么是 LangBot？
-
 LangBot 是一个**开源的生产级平台**，用于构建 AI 驱动的即时通信机器人。它将大语言模型（LLM）连接到各种聊天平台，帮助你创建能够对话、执行任务、并集成到现有工作流程中的智能 Agent。
 
 ### 核心能力
@@ -43,7 +41,7 @@ LangBot 是一个**开源的生产级平台**，用于构建 AI 驱动的即时
 - **AI 对话与 Agent** — 多轮对话、工具调用、多模态、流式输出。自带 RAG（知识库），深度集成 [Dify](https://dify.ai)、[Coze](https://coze.com)、[n8n](https://n8n.io)、[Langflow](https://langflow.org) 等 LLMOps 平台。
 - **全平台支持** — 一套代码，覆盖 QQ、微信、企业微信、飞书、钉钉、Discord、Telegram、Slack、LINE、KOOK 等平台。
 - **生产就绪** — 访问控制、限速、敏感词过滤、全面监控与异常处理，已被多家企业采用。
-- **插件生态** — 数百个插件，事件驱动架构，组件扩展，适配 [MCP 协议](https://modelcontextprotocol.io/)。
+- **插件生态** — 数百个插件，跨进程的事件驱动架构，组件扩展，适配 [MCP 协议](https://modelcontextprotocol.io/)。
 - **Web 管理面板** — 通过浏览器直观地配置、管理和监控机器人，无需手动编辑配置文件。
 - **多流水线架构** — 不同机器人用于不同场景，具备全面的监控和异常处理能力。
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "langbot"
-version = "4.9.0"
+version = "4.9.4"
 description = "Production-grade platform for building agentic IM bots"
 readme = "README.md"
 license-files = ["LICENSE"]
@@ -64,7 +64,7 @@ dependencies = [
     "chromadb>=1.0.0,<2.0.0",
     "qdrant-client (>=1.15.1,<2.0.0)",
     "pyseekdb==1.1.0.post3",
-    "langbot-plugin==0.3.0",
+    "langbot-plugin==0.3.5",
     "asyncpg>=0.30.0",
     "line-bot-sdk>=3.19.0",
     "tboxsdk>=0.0.10",

src/langbot/__init__.py

@@ -1,3 +1,3 @@
 """LangBot - Production-grade platform for building agentic IM bots"""
 
-__version__ = '4.9.0'
+__version__ = '4.9.4'

src/langbot/libs/openclaw_weixin_api/__init__.py

@@ -0,0 +1,3 @@
+from .client import OpenClawWeixinClient as OpenClawWeixinClient
+from .types import ApiError as ApiError
+from .types import LoginResult as LoginResult

src/langbot/libs/openclaw_weixin_api/client.py

@@ -0,0 +1,807 @@
+"""Async HTTP client for the OpenClaw WeChat API.
+
+Implements the iLink Bot API protocol.
+Reference: https://github.com/epiral/weixin-bot
+
+Endpoints: getUpdates (long-poll), sendMessage, getUploadUrl, getConfig, sendTyping.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import io
+import logging
+import os
+import struct
+import typing
+import uuid
+from typing import Optional
+from urllib.parse import quote
+
+import aiohttp
+
+from .types import (
+    ApiError,
+    CDNMedia,
+    FileItem,
+    GetConfigResponse,
+    GetUpdatesResponse,
+    GetUploadUrlResponse,
+    ImageItem,
+    LoginResult,
+    MessageItem,
+    QRCodeResponse,
+    QRStatusResponse,
+    RefMessage,
+    TextItem,
+    VideoItem,
+    VoiceItem,
+    WeixinMessage,
+)
+
+logger = logging.getLogger('openclaw-weixin-sdk')
+
+DEFAULT_BASE_URL = 'https://ilinkai.weixin.qq.com'
+CDN_BASE_URL = 'https://novac2c.cdn.weixin.qq.com/c2c'
+
+CHANNEL_VERSION = '1.0.0'
+
+DEFAULT_API_TIMEOUT = 15
+DEFAULT_LONG_POLL_TIMEOUT = 40
+DEFAULT_CONFIG_TIMEOUT = 10
+DEFAULT_QR_POLL_TIMEOUT = 35
+
+SESSION_EXPIRED_ERRCODE = -14
+
+DEFAULT_BOT_TYPE = '3'
+
+# Maximum text length per message chunk (WeChat limit)
+MAX_TEXT_CHUNK_SIZE = 2000
+
+
+def _random_wechat_uin() -> str:
+    """Generate the X-WECHAT-UIN header: random uint32 -> decimal string -> base64."""
+    rand_bytes = os.urandom(4)
+    uint32_val = struct.unpack('>I', rand_bytes)[0]
+    return base64.b64encode(str(uint32_val).encode('utf-8')).decode('utf-8')
+
+
+def _build_base_info() -> dict:
+    """Build the base_info payload included in every API request."""
+    return {'channel_version': CHANNEL_VERSION}
+
+
+def _chunk_text(text: str, max_size: int = MAX_TEXT_CHUNK_SIZE) -> list[str]:
+    """Split long text into chunks that fit within WeChat's message size limit."""
+    if len(text) <= max_size:
+        return [text]
+    chunks = []
+    while text:
+        chunks.append(text[:max_size])
+        text = text[max_size:]
+    return chunks
+
+
+class OpenClawWeixinClient:
+    """Async client for the OpenClaw WeChat HTTP JSON API."""
+
+    def __init__(self, base_url: str, token: str):
+        self.base_url = base_url.rstrip('/')
+        self.token = token
+        self._session: Optional[aiohttp.ClientSession] = None
+
+    async def _get_session(self) -> aiohttp.ClientSession:
+        if self._session is None or self._session.closed:
+            self._session = aiohttp.ClientSession()
+        return self._session
+
+    async def close(self):
+        if self._session and not self._session.closed:
+            await self._session.close()
+
+    def _build_headers(self) -> dict[str, str]:
+        headers = {
+            'Content-Type': 'application/json',
+            'AuthorizationType': 'ilink_bot_token',
+            'X-WECHAT-UIN': _random_wechat_uin(),
+        }
+        if self.token:
+            headers['Authorization'] = f'Bearer {self.token}'
+        return headers
+
+    async def _post(self, endpoint: str, payload: dict, timeout: float = DEFAULT_API_TIMEOUT) -> dict:
+        """Make a POST request and return the JSON response.
+
+        Raises ApiError on HTTP errors or when the response contains a non-zero errcode.
+        """
+        payload['base_info'] = _build_base_info()
+
+        session = await self._get_session()
+        url = f'{self.base_url}/{endpoint}'
+        headers = self._build_headers()
+
+        async with session.post(
+            url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=timeout)
+        ) as resp:
+            if resp.status != 200:
+                text = await resp.text()
+                raise ApiError(
+                    f'OpenClaw API error {resp.status}: {text}',
+                    status=resp.status,
+                )
+            data = await resp.json(content_type=None)
+
+        # Check for application-level errors in the response body
+        errcode = data.get('errcode') or data.get('ret')
+        if errcode and errcode != 0:
+            raise ApiError(
+                data.get('errmsg') or f'API errcode {errcode}',
+                status=200,
+                code=errcode,
+                payload=data,
+            )
+
+        return data
+
+    async def get_updates(
+        self, get_updates_buf: str = '', timeout: float = DEFAULT_LONG_POLL_TIMEOUT
+    ) -> GetUpdatesResponse:
+        """Long-poll for new messages.
+
+        Note: This method does NOT raise ApiError for errcode responses —
+        it returns them in the GetUpdatesResponse so the caller can handle
+        session expiry and other errors with full context.
+        """
+        try:
+            # Bypass the errcode check in _post since get_updates needs
+            # to return error info (e.g. session expired) to the caller.
+            payload: dict = {'get_updates_buf': get_updates_buf}
+            payload['base_info'] = _build_base_info()
+
+            session = await self._get_session()
+            url = f'{self.base_url}/ilink/bot/getupdates'
+            headers = self._build_headers()
+
+            async with session.post(
+                url,
+                json=payload,
+                headers=headers,
+                timeout=aiohttp.ClientTimeout(total=timeout),
+            ) as resp:
+                if resp.status != 200:
+                    text = await resp.text()
+                    raise ApiError(
+                        f'OpenClaw API error {resp.status}: {text}',
+                        status=resp.status,
+                    )
+                data = await resp.json(content_type=None)
+
+        except (asyncio.TimeoutError, aiohttp.ServerTimeoutError):
+            return GetUpdatesResponse(ret=0, msgs=[], get_updates_buf=get_updates_buf)
+        except ApiError:
+            raise
+        except Exception as e:
+            if 'timeout' in str(e).lower():
+                return GetUpdatesResponse(ret=0, msgs=[], get_updates_buf=get_updates_buf)
+            raise
+
+        return _parse_get_updates_response(data)
+
+    async def send_message(
+        self,
+        to_user_id: str,
+        item_list: list[MessageItem],
+        context_token: str = '',
+    ) -> None:
+        """Send a message to a user."""
+        items_payload = [_message_item_to_dict(item) for item in item_list]
+
+        payload = {
+            'msg': {
+                'from_user_id': '',
+                'to_user_id': to_user_id,
+                'client_id': f'langbot-{uuid.uuid4().hex[:16]}',
+                'message_type': WeixinMessage.TYPE_BOT,
+                'message_state': WeixinMessage.STATE_FINISH,
+                'item_list': items_payload,
+                'context_token': context_token or None,
+            }
+        }
+        await self._post('ilink/bot/sendmessage', payload)
+
+    async def send_text(self, to_user_id: str, text: str, context_token: str = '') -> None:
+        """Send a plain text message, automatically chunking if too long."""
+        chunks = _chunk_text(text)
+        for chunk in chunks:
+            item = MessageItem(type=MessageItem.TEXT, text_item=TextItem(text=chunk))
+            await self.send_message(to_user_id, [item], context_token)
+
+    async def get_config(self, ilink_user_id: str, context_token: str = '') -> GetConfigResponse:
+        """Get bot config including typing_ticket."""
+        data = await self._post(
+            'ilink/bot/getconfig',
+            {'ilink_user_id': ilink_user_id, 'context_token': context_token or None},
+            timeout=DEFAULT_CONFIG_TIMEOUT,
+        )
+        return GetConfigResponse(
+            ret=data.get('ret'),
+            errmsg=data.get('errmsg'),
+            typing_ticket=data.get('typing_ticket'),
+        )
+
+    async def send_typing(self, ilink_user_id: str, typing_ticket: str, status: int = 1) -> None:
+        """Send typing indicator. status: 1=typing, 2=cancel."""
+        await self._post(
+            'ilink/bot/sendtyping',
+            {
+                'ilink_user_id': ilink_user_id,
+                'typing_ticket': typing_ticket,
+                'status': status,
+            },
+            timeout=DEFAULT_CONFIG_TIMEOUT,
+        )
+
+    async def stop_typing(self, ilink_user_id: str, typing_ticket: str) -> None:
+        """Cancel the typing indicator for a user."""
+        await self.send_typing(ilink_user_id, typing_ticket, status=2)
+
+    async def download_media(
+        self,
+        media: CDNMedia,
+    ) -> bytes:
+        """Download and decrypt a file from the WeChat CDN.
+
+        Args:
+            media: CDNMedia object with encrypt_query_param and aes_key.
+
+        Returns:
+            Decrypted file bytes.
+        """
+        from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+        from cryptography.hazmat.primitives.padding import PKCS7
+
+        if not media.encrypt_query_param:
+            raise ApiError('CDN media has no encrypt_query_param', status=0)
+        if not media.aes_key:
+            raise ApiError('CDN media has no aes_key', status=0)
+
+        # Derive 16-byte AES key
+        # aes_key is base64-encoded; the decoded content may be:
+        #   - raw 16 bytes (direct AES key)
+        #   - 32-char hex string (decode hex to get 16 bytes)
+        raw = base64.b64decode(media.aes_key)
+        if len(raw) == 16:
+            aes_key = raw
+        elif len(raw) == 32:
+            # Hex-encoded 16-byte key
+            aes_key = bytes.fromhex(raw.decode('utf-8'))
+        else:
+            raise ApiError(f'Invalid AES key length: {len(raw)} (expected 16 or 32)', status=0)
+
+        # Download encrypted bytes from CDN
+        session = await self._get_session()
+        cdn_url = f'{CDN_BASE_URL}/download?encrypted_query_param={quote(media.encrypt_query_param, safe="")}'
+
+        async with session.get(cdn_url, timeout=aiohttp.ClientTimeout(total=120)) as resp:
+            if resp.status != 200:
+                text = await resp.text()
+                raise ApiError(f'CDN download failed: {resp.status} {text}', status=resp.status)
+            encrypted = await resp.read()
+
+        # Decrypt AES-128-ECB with PKCS7 padding
+        cipher = Cipher(algorithms.AES(aes_key), modes.ECB())
+        decryptor = cipher.decryptor()
+        padded = decryptor.update(encrypted) + decryptor.finalize()
+
+        unpadder = PKCS7(128).unpadder()
+        return unpadder.update(padded) + unpadder.finalize()
+
+    async def upload_media(
+        self,
+        file_bytes: bytes,
+        to_user_id: str,
+        media_type: int,
+    ) -> CDNMedia:
+        """Encrypt and upload media to WeChat CDN.
+
+        Args:
+            file_bytes: Raw file bytes to upload.
+            to_user_id: Recipient user ID.
+            media_type: 1=IMAGE, 2=VIDEO, 3=FILE, 4=VOICE.
+
+        Returns:
+            CDNMedia with encrypt_query_param and aes_key for use in sendMessage.
+        """
+        import hashlib
+
+        from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+        from cryptography.hazmat.primitives.padding import PKCS7
+
+        # 1. Generate random 16-byte AES key
+        raw_key = os.urandom(16)
+        aes_key_hex = raw_key.hex()  # 32-char hex string
+
+        # 2. Encode key for CDNMedia: base64(hex_string) — same for all media types
+        # Matches official SDK: Buffer.from(aeskey_hex).toString("base64")
+        encoded_key = base64.b64encode(aes_key_hex.encode('utf-8')).decode('utf-8')
+
+        # 3. Encrypt file with AES-128-ECB + PKCS7
+        padder = PKCS7(128).padder()
+        padded = padder.update(file_bytes) + padder.finalize()
+        cipher = Cipher(algorithms.AES(raw_key), modes.ECB())
+        encryptor = cipher.encryptor()
+        encrypted = encryptor.update(padded) + encryptor.finalize()
+
+        # 4. Get upload URL
+        raw_md5 = hashlib.md5(file_bytes).hexdigest()
+        filekey = os.urandom(16).hex()  # 32-char hex, matches official SDK
+
+        upload_resp = await self.get_upload_url(
+            filekey=filekey,
+            media_type=media_type,
+            to_user_id=to_user_id,
+            rawsize=len(file_bytes),
+            rawfilemd5=raw_md5,
+            filesize=len(encrypted),
+            aeskey=aes_key_hex,  # hex string, as expected by the API
+        )
+
+        if not upload_resp.upload_param:
+            raise ApiError('Failed to get upload URL', status=0)
+
+        # 5. Upload to CDN
+        # upload_param is an opaque token from the server — pass it as-is
+        session = await self._get_session()
+        cdn_url = f'{CDN_BASE_URL}/upload?encrypted_query_param={quote(upload_resp.upload_param, safe="")}&filekey={quote(filekey, safe="")}'
+        logger.debug(
+            'CDN upload: url=%s raw_size=%d encrypted_size=%d md5=%s aeskey=%s',
+            cdn_url,
+            len(file_bytes),
+            len(encrypted),
+            raw_md5,
+            encoded_key,
+        )
+
+        async with session.post(
+            cdn_url,
+            data=encrypted,
+            headers={'Content-Type': 'application/octet-stream'},
+            timeout=aiohttp.ClientTimeout(total=120),
+        ) as resp:
+            if resp.status != 200:
+                text = await resp.text()
+                logger.error('CDN upload failed: status=%d url=%s body=%s', resp.status, cdn_url, text[:500])
+                raise ApiError(f'CDN upload failed: {resp.status} {text}', status=resp.status)
+            download_param = resp.headers.get('x-encrypted-param', '')
+
+        if not download_param:
+            raise ApiError('CDN upload succeeded but no x-encrypted-param returned', status=0)
+
+        return CDNMedia(
+            encrypt_query_param=download_param,
+            aes_key=encoded_key,
+            encrypt_type=1,
+        )
+
+    async def send_image(
+        self,
+        to_user_id: str,
+        image_bytes: bytes,
+        context_token: str = '',
+    ) -> None:
+        """Upload an image to CDN and send it."""
+        media = await self.upload_media(image_bytes, to_user_id, media_type=1)
+        item = MessageItem(
+            type=MessageItem.IMAGE,
+            image_item=ImageItem(
+                media=media,
+                aeskey=media.aes_key,
+            ),
+        )
+        await self.send_message(to_user_id, [item], context_token)
+
+    async def send_file(
+        self,
+        to_user_id: str,
+        file_bytes: bytes,
+        file_name: str,
+        context_token: str = '',
+    ) -> None:
+        """Upload a file to CDN and send it."""
+        import hashlib
+
+        media = await self.upload_media(file_bytes, to_user_id, media_type=3)
+        item = MessageItem(
+            type=MessageItem.FILE,
+            file_item=FileItem(
+                media=media,
+                file_name=file_name,
+                md5=hashlib.md5(file_bytes).hexdigest(),
+                len=str(len(file_bytes)),
+            ),
+        )
+        await self.send_message(to_user_id, [item], context_token)
+
+    async def send_voice(
+        self,
+        to_user_id: str,
+        voice_bytes: bytes,
+        playtime: int = 0,
+        context_token: str = '',
+    ) -> None:
+        """Upload a voice message to CDN and send it."""
+        media = await self.upload_media(voice_bytes, to_user_id, media_type=4)
+        item = MessageItem(
+            type=MessageItem.VOICE,
+            voice_item=VoiceItem(
+                media=media,
+                playtime=playtime,
+            ),
+        )
+        await self.send_message(to_user_id, [item], context_token)
+
+    async def get_upload_url(
+        self,
+        filekey: str,
+        media_type: int,
+        to_user_id: str,
+        rawsize: int,
+        rawfilemd5: str,
+        filesize: int,
+        thumb_rawsize: Optional[int] = None,
+        thumb_rawfilemd5: Optional[str] = None,
+        thumb_filesize: Optional[int] = None,
+        aeskey: Optional[str] = None,
+    ) -> GetUploadUrlResponse:
+        """Get a pre-signed CDN upload URL."""
+        payload: dict = {
+            'filekey': filekey,
+            'media_type': media_type,
+            'to_user_id': to_user_id,
+            'rawsize': rawsize,
+            'rawfilemd5': rawfilemd5,
+            'filesize': filesize,
+            'no_need_thumb': True,
+        }
+        if thumb_rawsize is not None:
+            payload['thumb_rawsize'] = thumb_rawsize
+        if thumb_rawfilemd5 is not None:
+            payload['thumb_rawfilemd5'] = thumb_rawfilemd5
+        if thumb_filesize is not None:
+            payload['thumb_filesize'] = thumb_filesize
+        if aeskey is not None:
+            payload['aeskey'] = aeskey
+
+        data = await self._post('ilink/bot/getuploadurl', payload)
+        logger.debug('get_upload_url response: %s', data)
+        return GetUploadUrlResponse(
+            upload_param=data.get('upload_param'),
+            thumb_upload_param=data.get('thumb_upload_param'),
+        )
+
+    # -----------------------------------------------------------------------
+    # QR Code Login
+    # -----------------------------------------------------------------------
+
+    async def fetch_qrcode(self, bot_type: str = DEFAULT_BOT_TYPE) -> QRCodeResponse:
+        """Fetch a QR code for WeChat login authorization (GET, no auth needed)."""
+        session = await self._get_session()
+        url = f'{self.base_url}/ilink/bot/get_bot_qrcode?bot_type={bot_type}'
+
+        async with session.get(url, timeout=aiohttp.ClientTimeout(total=DEFAULT_API_TIMEOUT)) as resp:
+            if resp.status != 200:
+                text = await resp.text()
+                raise ApiError(
+                    f'Failed to fetch QR code: {resp.status} {text}',
+                    status=resp.status,
+                )
+            data = await resp.json(content_type=None)
+
+        logger.debug(
+            'fetch_qrcode response: qrcode=%s, img=%s', data.get('qrcode'), bool(data.get('qrcode_img_content'))
+        )
+
+        return QRCodeResponse(
+            qrcode=data.get('qrcode'),
+            qrcode_img_content=data.get('qrcode_img_content'),
+        )
+
+    async def _fetch_qr_image_base64(self, url: str) -> str:
+        """Generate a QR code image from the URL and return a data URI string.
+
+        The qrcode_img_content URL points to an HTML page (not a raw image),
+        so we generate the QR code locally using the qrcode library.
+        """
+        import qrcode
+
+        qr = qrcode.QRCode(error_correction=qrcode.constants.ERROR_CORRECT_L)
+        qr.add_data(url)
+        qr.make(fit=True)
+        img = qr.make_image(fill_color='black', back_color='white')
+
+        buf = io.BytesIO()
+        img.save(buf, format='PNG')
+        b64 = base64.b64encode(buf.getvalue()).decode('utf-8')
+        return f'data:image/png;base64,{b64}'
+
+    async def poll_qrcode_status(self, qrcode: str) -> QRStatusResponse:
+        """Long-poll the QR code scan status (GET with iLink-App-ClientVersion header)."""
+        session = await self._get_session()
+        url = f'{self.base_url}/ilink/bot/get_qrcode_status?qrcode={quote(qrcode, safe="")}'
+        headers = {'iLink-App-ClientVersion': '1'}
+
+        try:
+            async with session.get(
+                url, headers=headers, timeout=aiohttp.ClientTimeout(total=DEFAULT_QR_POLL_TIMEOUT)
+            ) as resp:
+                if resp.status != 200:
+                    text = await resp.text()
+                    raise ApiError(
+                        f'Failed to poll QR status: {resp.status} {text}',
+                        status=resp.status,
+                    )
+                data = await resp.json(content_type=None)
+                logger.debug('QR status poll response: %s', data)
+        except (asyncio.TimeoutError, aiohttp.ServerTimeoutError):
+            return QRStatusResponse(status='wait')
+
+        return QRStatusResponse(
+            status=data.get('status'),
+            bot_token=data.get('bot_token'),
+            ilink_bot_id=data.get('ilink_bot_id'),
+            baseurl=data.get('baseurl'),
+            ilink_user_id=data.get('ilink_user_id'),
+        )
+
+    async def login(
+        self,
+        max_retries: int = 5,
+        poll_timeout_ms: int = 480_000,
+        on_qrcode: Optional[typing.Callable[[str, str], typing.Any]] = None,
+        on_status: Optional[typing.Callable[[str], typing.Any]] = None,
+    ) -> LoginResult:
+        """Complete QR code login flow with auto-retry on expiry.
+
+        Args:
+            max_retries: Max number of QR code refreshes on expiry.
+            poll_timeout_ms: Timeout per QR code in milliseconds.
+            on_qrcode: Callback(qr_image_base64, qr_url) called each time a
+                        new QR code is fetched. Use this to display the QR code.
+            on_status: Callback(status_str) called on each status poll change.
+
+        Returns:
+            LoginResult with token, base_url, and account_id.
+
+        Raises:
+            ApiError: On unrecoverable API errors.
+            Exception: If all retries are exhausted.
+        """
+        last_qr_base64: Optional[str] = None
+
+        for attempt in range(max_retries):
+            qr_resp = await self.fetch_qrcode()
+            if not qr_resp.qrcode or not qr_resp.qrcode_img_content:
+                raise ApiError('Failed to get QR code from server', status=0)
+
+            # Convert QR image to base64 and notify caller
+            last_qr_base64 = await self._fetch_qr_image_base64(qr_resp.qrcode_img_content)
+            if on_qrcode:
+                try:
+                    result = on_qrcode(last_qr_base64, qr_resp.qrcode_img_content)
+                    if asyncio.iscoroutine(result) or asyncio.isfuture(result):
+                        await result
+                except Exception as e:
+                    logger.warning('on_qrcode callback error: %s', e)
+
+            # Poll until confirmed / expired / timeout
+            loop = asyncio.get_running_loop()
+            deadline = loop.time() + poll_timeout_ms / 1000.0
+
+            while loop.time() < deadline:
+                try:
+                    status_resp = await self.poll_qrcode_status(qr_resp.qrcode)
+                except Exception as e:
+                    logger.error('Error polling QR status: %s', e)
+                    await asyncio.sleep(2)
+                    continue
+
+                if on_status:
+                    try:
+                        cb_result = on_status(status_resp.status or 'unknown')
+                        if asyncio.iscoroutine(cb_result) or asyncio.isfuture(cb_result):
+                            await cb_result
+                    except Exception as e:
+                        logger.warning('on_status callback error: %s', e)
+
+                if status_resp.status == 'confirmed' and status_resp.bot_token:
+                    new_base_url = status_resp.baseurl or self.base_url
+                    # Update this client instance as well
+                    self.token = status_resp.bot_token
+                    self.base_url = new_base_url.rstrip('/')
+                    return LoginResult(
+                        token=status_resp.bot_token,
+                        base_url=new_base_url,
+                        account_id=status_resp.ilink_bot_id or '',
+                        qr_image_base64=last_qr_base64,
+                    )
+
+                if status_resp.status == 'expired':
+                    break  # retry with a new QR code
+
+                await asyncio.sleep(1)
+            else:
+                # While-loop ended without break → poll timeout, treat as expired
+                pass
+
+            remaining = max_retries - attempt - 1
+            if remaining > 0:
+                logger.info('QR code expired, refreshing... (%d retries left)', remaining)
+            else:
+                raise ApiError('QR code login failed: max retries exceeded', status=0)
+
+        # Should not reach here, but just in case
+        raise ApiError('QR code login failed', status=0)
+
+
+# ---------------------------------------------------------------------------
+# Parsing helpers
+# ---------------------------------------------------------------------------
+
+
+def _parse_cdn_media(data: Optional[dict]) -> Optional[CDNMedia]:
+    if not data:
+        return None
+    return CDNMedia(
+        encrypt_query_param=data.get('encrypt_query_param'),
+        aes_key=data.get('aes_key'),
+        encrypt_type=data.get('encrypt_type'),
+    )
+
+
+def _parse_message_item(data: dict) -> MessageItem:
+    item = MessageItem(
+        type=data.get('type'),
+        create_time_ms=data.get('create_time_ms'),
+        update_time_ms=data.get('update_time_ms'),
+        is_completed=data.get('is_completed'),
+        msg_id=data.get('msg_id'),
+    )
+
+    if data.get('text_item'):
+        item.text_item = TextItem(text=data['text_item'].get('text'))
+
+    if data.get('image_item'):
+        img = data['image_item']
+        item.image_item = ImageItem(
+            media=_parse_cdn_media(img.get('media')),
+            thumb_media=_parse_cdn_media(img.get('thumb_media')),
+            aeskey=img.get('aeskey'),
+            url=img.get('url'),
+            mid_size=img.get('mid_size'),
+        )
+
+    if data.get('voice_item'):
+        v = data['voice_item']
+        item.voice_item = VoiceItem(
+            media=_parse_cdn_media(v.get('media')),
+            encode_type=v.get('encode_type'),
+            playtime=v.get('playtime'),
+            text=v.get('text'),
+        )
+
+    if data.get('file_item'):
+        f = data['file_item']
+        item.file_item = FileItem(
+            media=_parse_cdn_media(f.get('media')),
+            file_name=f.get('file_name'),
+            md5=f.get('md5'),
+            len=f.get('len'),
+        )
+
+    if data.get('video_item'):
+        vid = data['video_item']
+        item.video_item = VideoItem(
+            media=_parse_cdn_media(vid.get('media')),
+            video_size=vid.get('video_size'),
+            play_length=vid.get('play_length'),
+            video_md5=vid.get('video_md5'),
+            thumb_media=_parse_cdn_media(vid.get('thumb_media')),
+        )
+
+    if data.get('ref_msg'):
+        ref = data['ref_msg']
+        item.ref_msg = RefMessage(
+            title=ref.get('title'),
+            message_item=_parse_message_item(ref['message_item']) if ref.get('message_item') else None,
+        )
+
+    return item
+
+
+def _parse_weixin_message(data: dict) -> WeixinMessage:
+    msg = WeixinMessage(
+        seq=data.get('seq'),
+        message_id=data.get('message_id'),
+        from_user_id=data.get('from_user_id'),
+        to_user_id=data.get('to_user_id'),
+        client_id=data.get('client_id'),
+        create_time_ms=data.get('create_time_ms'),
+        session_id=data.get('session_id'),
+        group_id=data.get('group_id'),
+        message_type=data.get('message_type'),
+        message_state=data.get('message_state'),
+        context_token=data.get('context_token'),
+    )
+    if data.get('item_list'):
+        msg.item_list = [_parse_message_item(item) for item in data['item_list']]
+    return msg
+
+
+def _parse_get_updates_response(data: dict) -> GetUpdatesResponse:
+    resp = GetUpdatesResponse(
+        ret=data.get('ret'),
+        errcode=data.get('errcode'),
+        errmsg=data.get('errmsg'),
+        get_updates_buf=data.get('get_updates_buf'),
+        longpolling_timeout_ms=data.get('longpolling_timeout_ms'),
+    )
+    if data.get('msgs'):
+        resp.msgs = [_parse_weixin_message(m) for m in data['msgs']]
+    return resp
+
+
+def _cdn_media_to_dict(media: Optional[CDNMedia]) -> Optional[dict]:
+    if not media:
+        return None
+    d: dict = {}
+    if media.encrypt_query_param is not None:
+        d['encrypt_query_param'] = media.encrypt_query_param
+    if media.aes_key is not None:
+        d['aes_key'] = media.aes_key
+    if media.encrypt_type is not None:
+        d['encrypt_type'] = media.encrypt_type
+    return d or None
+
+
+def _message_item_to_dict(item: MessageItem) -> dict:
+    d: dict = {'type': item.type}
+
+    if item.text_item:
+        d['text_item'] = {'text': item.text_item.text}
+
+    if item.image_item:
+        img_d: dict = {}
+        if item.image_item.media:
+            img_d['media'] = _cdn_media_to_dict(item.image_item.media)
+        if item.image_item.mid_size is not None:
+            img_d['mid_size'] = item.image_item.mid_size
+        d['image_item'] = img_d
+
+    if item.voice_item:
+        voice_d: dict = {}
+        if item.voice_item.media:
+            voice_d['media'] = _cdn_media_to_dict(item.voice_item.media)
+        if item.voice_item.playtime is not None:
+            voice_d['playtime'] = item.voice_item.playtime
+        d['voice_item'] = voice_d
+
+    if item.file_item:
+        file_d: dict = {}
+        if item.file_item.media:
+            file_d['media'] = _cdn_media_to_dict(item.file_item.media)
+        if item.file_item.file_name:
+            file_d['file_name'] = item.file_item.file_name
+        if item.file_item.len:
+            file_d['len'] = item.file_item.len
+        d['file_item'] = file_d
+
+    if item.video_item:
+        vid_d: dict = {}
+        if item.video_item.media:
+            vid_d['media'] = _cdn_media_to_dict(item.video_item.media)
+        if item.video_item.video_size is not None:
+            vid_d['video_size'] = item.video_item.video_size
+        d['video_item'] = vid_d
+
+    return d

src/langbot/libs/openclaw_weixin_api/types.py

@@ -0,0 +1,200 @@
+"""Type definitions for the OpenClaw WeChat API, mirroring the upstream protocol."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Optional
+
+SESSION_EXPIRED_ERRCODE = -14
+
+
+class ApiError(Exception):
+    """Structured error raised by the OpenClaw WeChat API."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        status: int = 0,
+        code: int | None = None,
+        payload: Any = None,
+    ):
+        super().__init__(message)
+        self.status = status
+        self.code = code
+        self.payload = payload
+
+    @property
+    def is_session_expired(self) -> bool:
+        return self.code == SESSION_EXPIRED_ERRCODE
+
+
+@dataclass
+class CDNMedia:
+    encrypt_query_param: Optional[str] = None
+    aes_key: Optional[str] = None
+    encrypt_type: Optional[int] = None
+
+
+@dataclass
+class TextItem:
+    text: Optional[str] = None
+
+
+@dataclass
+class ImageItem:
+    media: Optional[CDNMedia] = None
+    thumb_media: Optional[CDNMedia] = None
+    aeskey: Optional[str] = None
+    url: Optional[str] = None
+    mid_size: Optional[int] = None
+    thumb_size: Optional[int] = None
+    thumb_height: Optional[int] = None
+    thumb_width: Optional[int] = None
+    hd_size: Optional[int] = None
+    _downloaded_bytes: Optional[bytes] = field(default=None, repr=False)
+
+
+@dataclass
+class VoiceItem:
+    media: Optional[CDNMedia] = None
+    encode_type: Optional[int] = None
+    bits_per_sample: Optional[int] = None
+    sample_rate: Optional[int] = None
+    playtime: Optional[int] = None
+    text: Optional[str] = None
+    _downloaded_bytes: Optional[bytes] = field(default=None, repr=False)
+
+
+@dataclass
+class FileItem:
+    media: Optional[CDNMedia] = None
+    file_name: Optional[str] = None
+    md5: Optional[str] = None
+    len: Optional[str] = None
+    _downloaded_bytes: Optional[bytes] = field(default=None, repr=False)
+
+
+@dataclass
+class VideoItem:
+    media: Optional[CDNMedia] = None
+    video_size: Optional[int] = None
+    play_length: Optional[int] = None
+    video_md5: Optional[str] = None
+    thumb_media: Optional[CDNMedia] = None
+    thumb_size: Optional[int] = None
+    thumb_height: Optional[int] = None
+    thumb_width: Optional[int] = None
+    _downloaded_bytes: Optional[bytes] = field(default=None, repr=False)
+
+
+@dataclass
+class RefMessage:
+    message_item: Optional[MessageItem] = None
+    title: Optional[str] = None
+
+
+@dataclass
+class MessageItem:
+    """A single content item inside a WeixinMessage."""
+
+    # Item types
+    NONE = 0
+    TEXT = 1
+    IMAGE = 2
+    VOICE = 3
+    FILE = 4
+    VIDEO = 5
+
+    type: Optional[int] = None
+    create_time_ms: Optional[int] = None
+    update_time_ms: Optional[int] = None
+    is_completed: Optional[bool] = None
+    msg_id: Optional[str] = None
+    ref_msg: Optional[RefMessage] = None
+    text_item: Optional[TextItem] = None
+    image_item: Optional[ImageItem] = None
+    voice_item: Optional[VoiceItem] = None
+    file_item: Optional[FileItem] = None
+    video_item: Optional[VideoItem] = None
+
+
+@dataclass
+class WeixinMessage:
+    """Unified message from getUpdates or for sendMessage."""
+
+    # Message types
+    TYPE_USER = 1
+    TYPE_BOT = 2
+
+    # Message states
+    STATE_NEW = 0
+    STATE_GENERATING = 1
+    STATE_FINISH = 2
+
+    seq: Optional[int] = None
+    message_id: Optional[int] = None
+    from_user_id: Optional[str] = None
+    to_user_id: Optional[str] = None
+    client_id: Optional[str] = None
+    create_time_ms: Optional[int] = None
+    update_time_ms: Optional[int] = None
+    delete_time_ms: Optional[int] = None
+    session_id: Optional[str] = None
+    group_id: Optional[str] = None
+    message_type: Optional[int] = None
+    message_state: Optional[int] = None
+    item_list: Optional[list[MessageItem]] = None
+    context_token: Optional[str] = None
+
+
+@dataclass
+class GetUpdatesResponse:
+    ret: Optional[int] = None
+    errcode: Optional[int] = None
+    errmsg: Optional[str] = None
+    msgs: list[WeixinMessage] = field(default_factory=list)
+    get_updates_buf: Optional[str] = None
+    longpolling_timeout_ms: Optional[int] = None
+
+
+@dataclass
+class GetConfigResponse:
+    ret: Optional[int] = None
+    errmsg: Optional[str] = None
+    typing_ticket: Optional[str] = None
+
+
+@dataclass
+class GetUploadUrlResponse:
+    upload_param: Optional[str] = None
+    thumb_upload_param: Optional[str] = None
+
+
+@dataclass
+class QRCodeResponse:
+    """Response from get_bot_qrcode endpoint."""
+
+    qrcode: Optional[str] = None
+    qrcode_img_content: Optional[str] = None
+
+
+@dataclass
+class QRStatusResponse:
+    """Response from get_qrcode_status endpoint."""
+
+    status: Optional[str] = None  # "wait" | "scaned" | "confirmed" | "expired"
+    bot_token: Optional[str] = None
+    ilink_bot_id: Optional[str] = None
+    baseurl: Optional[str] = None
+    ilink_user_id: Optional[str] = None
+
+
+@dataclass
+class LoginResult:
+    """Result returned by the login flow."""
+
+    token: str
+    base_url: str
+    account_id: str
+    qr_image_base64: Optional[str] = None  # data URI of the last QR code shown

src/langbot/libs/wecom_ai_bot_api/api.py

@@ -199,6 +199,253 @@ class StreamSessionManager:
                 self._msg_index.pop(msg_id, None)
 
 
+async def download_encrypted_file(download_url: str, encoding_aes_key: str, logger: EventLogger) -> Optional[str]:
+    """Download an AES-encrypted file from WeChat Work and return as data URI.
+
+    Args:
+        download_url: The encrypted file download URL.
+        encoding_aes_key: The AES key used for decryption (base64-encoded, without trailing '=').
+        logger: Logger instance.
+
+    Returns:
+        A data URI string (e.g. 'data:image/jpeg;base64,...') or None on failure.
+    """
+    if not download_url:
+        return None
+    async with httpx.AsyncClient() as client:
+        response = await client.get(download_url)
+        if response.status_code != 200:
+            await logger.error(f'failed to get file: {response.text}')
+            return None
+        encrypted_bytes = response.content
+
+    aes_key = base64.b64decode(encoding_aes_key + '=')
+    iv = aes_key[:16]
+
+    cipher = AES.new(aes_key, AES.MODE_CBC, iv)
+    decrypted = cipher.decrypt(encrypted_bytes)
+
+    pad_len = decrypted[-1]
+    decrypted = decrypted[:-pad_len]
+
+    if decrypted.startswith(b'\xff\xd8'):
+        mime_type = 'image/jpeg'
+    elif decrypted.startswith(b'\x89PNG'):
+        mime_type = 'image/png'
+    elif decrypted.startswith((b'GIF87a', b'GIF89a')):
+        mime_type = 'image/gif'
+    elif decrypted.startswith(b'BM'):
+        mime_type = 'image/bmp'
+    elif decrypted.startswith(b'II*\x00') or decrypted.startswith(b'MM\x00*'):
+        mime_type = 'image/tiff'
+    else:
+        mime_type = 'application/octet-stream'
+
+    base64_str = base64.b64encode(decrypted).decode('utf-8')
+    return f'data:{mime_type};base64,{base64_str}'
+
+
+async def parse_wecom_bot_message(
+    msg_json: dict[str, Any], encoding_aes_key: str, logger: EventLogger
+) -> dict[str, Any]:
+    """Parse a decrypted WeChat Work AI Bot message JSON into a unified message dict.
+
+    This is the shared message parsing logic used by both webhook and WebSocket modes.
+
+    Args:
+        msg_json: The decrypted message JSON from WeChat Work.
+        encoding_aes_key: AES key for file decryption.
+        logger: Logger instance.
+
+    Returns:
+        A dict suitable for constructing a WecomBotEvent.
+    """
+    message_data: dict[str, Any] = {}
+
+    msg_type = msg_json.get('msgtype', '')
+    if msg_type:
+        message_data['msgtype'] = msg_type
+
+    if msg_json.get('chattype', '') == 'single':
+        message_data['type'] = 'single'
+    elif msg_json.get('chattype', '') == 'group':
+        message_data['type'] = 'group'
+
+    max_inline_file_size = 5 * 1024 * 1024
+
+    async def _safe_download(url: str):
+        if not url:
+            return None
+        return await download_encrypted_file(url, encoding_aes_key, logger)
+
+    if msg_type == 'text':
+        message_data['content'] = msg_json.get('text', {}).get('content')
+    elif msg_type == 'markdown':
+        message_data['content'] = msg_json.get('markdown', {}).get('content') or msg_json.get('text', {}).get(
+            'content', ''
+        )
+    elif msg_type == 'image':
+        picurl = msg_json.get('image', {}).get('url', '')
+        base64_data = await _safe_download(picurl)
+        if base64_data:
+            message_data['picurl'] = base64_data
+            message_data['images'] = [base64_data]
+    elif msg_type == 'voice':
+        voice_info = msg_json.get('voice', {}) or {}
+        download_url = voice_info.get('url')
+        message_data['voice'] = {
+            'url': download_url,
+            'md5sum': voice_info.get('md5sum') or voice_info.get('md5'),
+            'filesize': voice_info.get('filesize') or voice_info.get('size'),
+            'sdkfileid': voice_info.get('sdkfileid') or voice_info.get('fileid'),
+        }
+        if voice_info.get('content'):
+            message_data['content'] = voice_info.get('content')
+        if (message_data['voice'].get('filesize') or 0) <= max_inline_file_size:
+            voice_base64 = await _safe_download(download_url)
+            if voice_base64:
+                message_data['voice']['base64'] = voice_base64
+    elif msg_type == 'video':
+        video_info = msg_json.get('video', {}) or {}
+        download_url = video_info.get('url')
+        video_data = {
+            'url': download_url,
+            'filesize': video_info.get('filesize') or video_info.get('size'),
+            'sdkfileid': video_info.get('sdkfileid') or video_info.get('fileid'),
+            'md5sum': video_info.get('md5sum') or video_info.get('md5'),
+            'filename': video_info.get('filename') or video_info.get('name'),
+        }
+        if (video_data.get('filesize') or 0) <= max_inline_file_size:
+            video_base64 = await _safe_download(download_url)
+            if video_base64:
+                video_data['base64'] = video_base64
+        message_data['video'] = video_data
+    elif msg_type == 'file':
+        file_info = msg_json.get('file', {}) or {}
+        download_url = file_info.get('url') or file_info.get('fileurl')
+        file_data = {
+            'filename': file_info.get('filename') or file_info.get('name'),
+            'filesize': file_info.get('filesize') or file_info.get('size'),
+            'md5sum': file_info.get('md5sum') or file_info.get('md5'),
+            'sdkfileid': file_info.get('sdkfileid') or file_info.get('fileid'),
+            'download_url': download_url,
+            'extra': file_info,
+        }
+        if (file_data.get('filesize') or 0) <= max_inline_file_size:
+            file_base64 = await _safe_download(download_url)
+            if file_base64:
+                file_data['base64'] = file_base64
+        message_data['file'] = file_data
+    elif msg_type == 'link':
+        message_data['link'] = msg_json.get('link', {})
+        if not message_data.get('content'):
+            title = message_data['link'].get('title', '')
+            desc = message_data['link'].get('description') or message_data['link'].get('digest', '')
+            message_data['content'] = '\n'.join(filter(None, [title, desc]))
+    elif msg_type == 'mixed':
+        items = msg_json.get('mixed', {}).get('msg_item', [])
+        texts = []
+        images = []
+        files = []
+        voices = []
+        videos = []
+        links = []
+        for item in items:
+            item_type = item.get('msgtype')
+            if item_type == 'text':
+                texts.append(item.get('text', {}).get('content', ''))
+            elif item_type == 'image':
+                img_url = item.get('image', {}).get('url')
+                base64_data = await _safe_download(img_url)
+                if base64_data:
+                    images.append(base64_data)
+            elif item_type == 'file':
+                file_info = item.get('file', {}) or {}
+                download_url = file_info.get('url') or file_info.get('fileurl')
+                file_data = {
+                    'filename': file_info.get('filename') or file_info.get('name'),
+                    'filesize': file_info.get('filesize') or file_info.get('size'),
+                    'md5sum': file_info.get('md5sum') or file_info.get('md5'),
+                    'sdkfileid': file_info.get('sdkfileid') or file_info.get('fileid'),
+                    'download_url': download_url,
+                    'extra': file_info,
+                }
+                if (file_data.get('filesize') or 0) <= max_inline_file_size:
+                    file_base64 = await _safe_download(download_url)
+                    if file_base64:
+                        file_data['base64'] = file_base64
+                files.append(file_data)
+            elif item_type == 'voice':
+                voice_info = item.get('voice', {}) or {}
+                download_url = voice_info.get('url')
+                voice_data = {
+                    'url': download_url,
+                    'md5sum': voice_info.get('md5sum') or voice_info.get('md5'),
+                    'filesize': voice_info.get('filesize') or voice_info.get('size'),
+                    'sdkfileid': voice_info.get('sdkfileid') or voice_info.get('fileid'),
+                }
+                if voice_info.get('content'):
+                    texts.append(voice_info.get('content'))
+                if (voice_data.get('filesize') or 0) <= max_inline_file_size:
+                    voice_base64 = await _safe_download(download_url)
+                    if voice_base64:
+                        voice_data['base64'] = voice_base64
+                voices.append(voice_data)
+            elif item_type == 'video':
+                video_info = item.get('video', {}) or {}
+                download_url = video_info.get('url')
+                video_data = {
+                    'url': download_url,
+                    'filesize': video_info.get('filesize') or video_info.get('size'),
+                    'sdkfileid': video_info.get('sdkfileid') or video_info.get('fileid'),
+                    'md5sum': video_info.get('md5sum') or video_info.get('md5'),
+                    'filename': video_info.get('filename') or video_info.get('name'),
+                }
+                if (video_data.get('filesize') or 0) <= max_inline_file_size:
+                    video_base64 = await _safe_download(download_url)
+                    if video_base64:
+                        video_data['base64'] = video_base64
+                videos.append(video_data)
+            elif item_type == 'link':
+                links.append(item.get('link', {}))
+
+        if texts:
+            message_data['content'] = ' '.join(texts)
+        if images:
+            message_data['images'] = images
+            message_data['picurl'] = images[0]
+        if files:
+            message_data['files'] = files
+            message_data['file'] = files[0]
+        if voices:
+            message_data['voices'] = voices
+            message_data['voice'] = voices[0]
+        if videos:
+            message_data['videos'] = videos
+            message_data['video'] = videos[0]
+        if links:
+            message_data['link'] = links[0]
+        if items:
+            message_data['attachments'] = items
+    else:
+        message_data['raw_msg'] = msg_json
+
+    from_info = msg_json.get('from', {})
+    message_data['userid'] = from_info.get('userid', '')
+    message_data['username'] = from_info.get('alias', '') or from_info.get('name', '') or from_info.get('userid', '')
+
+    if msg_json.get('chattype', '') == 'group':
+        message_data['chatid'] = msg_json.get('chatid', '')
+        message_data['chatname'] = msg_json.get('chatname', '') or msg_json.get('chatid', '')
+
+    message_data['msgid'] = msg_json.get('msgid', '')
+
+    if msg_json.get('aibotid'):
+        message_data['aibotid'] = msg_json.get('aibotid', '')
+
+    return message_data
+
+
 class WecomBotClient:
     def __init__(self, Token: str, EnCodingAESKey: str, Corpid: str, logger: EventLogger, unified_mode: bool = False):
         """企业微信智能机器人客户端。
@@ -455,196 +702,7 @@ class WecomBotClient:
         return await self._handle_post_initial_response(msg_json, nonce)
 
     async def get_message(self, msg_json):
-        message_data = {}
+        return await parse_wecom_bot_message(msg_json, self.EnCodingAESKey, self.logger)
-
-        msg_type = msg_json.get('msgtype', '')
-        if msg_type:
-            message_data['msgtype'] = msg_type
-
-        if msg_json.get('chattype', '') == 'single':
-            message_data['type'] = 'single'
-        elif msg_json.get('chattype', '') == 'group':
-            message_data['type'] = 'group'
-
-        max_inline_file_size = 5 * 1024 * 1024  # avoid decoding very large payloads by default
-
-        async def _safe_download(url: str):
-            if not url:
-                return None
-            return await self.download_url_to_base64(url, self.EnCodingAESKey)
-
-        if msg_type == 'text':
-            message_data['content'] = msg_json.get('text', {}).get('content')
-        elif msg_type == 'markdown':
-            message_data['content'] = msg_json.get('markdown', {}).get('content') or msg_json.get('text', {}).get(
-                'content', ''
-            )
-        elif msg_type == 'image':
-            picurl = msg_json.get('image', {}).get('url', '')
-            base64_data = await _safe_download(picurl)
-            if base64_data:
-                message_data['picurl'] = base64_data
-                message_data['images'] = [base64_data]
-        elif msg_type == 'voice':
-            voice_info = msg_json.get('voice', {}) or {}
-            download_url = voice_info.get('url')
-            message_data['voice'] = {
-                'url': download_url,
-                'md5sum': voice_info.get('md5sum') or voice_info.get('md5'),
-                'filesize': voice_info.get('filesize') or voice_info.get('size'),
-                'sdkfileid': voice_info.get('sdkfileid') or voice_info.get('fileid'),
-            }
-            # 企业微信智能转写文本（如果已有）直接复用，避免重复转写
-            if voice_info.get('content'):
-                message_data['content'] = voice_info.get('content')
-            if (message_data['voice'].get('filesize') or 0) <= max_inline_file_size:
-                voice_base64 = await _safe_download(download_url)
-                if voice_base64:
-                    message_data['voice']['base64'] = voice_base64
-        elif msg_type == 'video':
-            video_info = msg_json.get('video', {}) or {}
-            download_url = video_info.get('url')
-            video_data = {
-                'url': download_url,
-                'filesize': video_info.get('filesize') or video_info.get('size'),
-                'sdkfileid': video_info.get('sdkfileid') or video_info.get('fileid'),
-                'md5sum': video_info.get('md5sum') or video_info.get('md5'),
-                'filename': video_info.get('filename') or video_info.get('name'),
-            }
-            if (video_data.get('filesize') or 0) <= max_inline_file_size:
-                video_base64 = await _safe_download(download_url)
-                if video_base64:
-                    video_data['base64'] = video_base64
-            message_data['video'] = video_data
-        elif msg_type == 'file':
-            file_info = msg_json.get('file', {}) or {}
-            download_url = file_info.get('url') or file_info.get('fileurl')
-            file_data = {
-                'filename': file_info.get('filename') or file_info.get('name'),
-                'filesize': file_info.get('filesize') or file_info.get('size'),
-                'md5sum': file_info.get('md5sum') or file_info.get('md5'),
-                'sdkfileid': file_info.get('sdkfileid') or file_info.get('fileid'),
-                'download_url': download_url,
-                'extra': file_info,
-            }
-            if (file_data.get('filesize') or 0) <= max_inline_file_size:
-                file_base64 = await _safe_download(download_url)
-                if file_base64:
-                    file_data['base64'] = file_base64
-            message_data['file'] = file_data
-        elif msg_type == 'link':
-            message_data['link'] = msg_json.get('link', {})
-            if not message_data.get('content'):
-                title = message_data['link'].get('title', '')
-                desc = message_data['link'].get('description') or message_data['link'].get('digest', '')
-                message_data['content'] = '\n'.join(filter(None, [title, desc]))
-        elif msg_type == 'mixed':
-            items = msg_json.get('mixed', {}).get('msg_item', [])
-            texts = []
-            images = []
-            files = []
-            voices = []
-            videos = []
-            links = []
-            for item in items:
-                item_type = item.get('msgtype')
-                if item_type == 'text':
-                    texts.append(item.get('text', {}).get('content', ''))
-                elif item_type == 'image':
-                    img_url = item.get('image', {}).get('url')
-                    base64_data = await _safe_download(img_url)
-                    if base64_data:
-                        images.append(base64_data)
-                elif item_type == 'file':
-                    file_info = item.get('file', {}) or {}
-                    download_url = file_info.get('url') or file_info.get('fileurl')
-                    file_data = {
-                        'filename': file_info.get('filename') or file_info.get('name'),
-                        'filesize': file_info.get('filesize') or file_info.get('size'),
-                        'md5sum': file_info.get('md5sum') or file_info.get('md5'),
-                        'sdkfileid': file_info.get('sdkfileid') or file_info.get('fileid'),
-                        'download_url': download_url,
-                        'extra': file_info,
-                    }
-                    if (file_data.get('filesize') or 0) <= max_inline_file_size:
-                        file_base64 = await _safe_download(download_url)
-                        if file_base64:
-                            file_data['base64'] = file_base64
-                    files.append(file_data)
-                elif item_type == 'voice':
-                    voice_info = item.get('voice', {}) or {}
-                    download_url = voice_info.get('url')
-                    voice_data = {
-                        'url': download_url,
-                        'md5sum': voice_info.get('md5sum') or voice_info.get('md5'),
-                        'filesize': voice_info.get('filesize') or voice_info.get('size'),
-                        'sdkfileid': voice_info.get('sdkfileid') or voice_info.get('fileid'),
-                    }
-                    if voice_info.get('content'):
-                        texts.append(voice_info.get('content'))
-                    if (voice_data.get('filesize') or 0) <= max_inline_file_size:
-                        voice_base64 = await _safe_download(download_url)
-                        if voice_base64:
-                            voice_data['base64'] = voice_base64
-                    voices.append(voice_data)
-                elif item_type == 'video':
-                    video_info = item.get('video', {}) or {}
-                    download_url = video_info.get('url')
-                    video_data = {
-                        'url': download_url,
-                        'filesize': video_info.get('filesize') or video_info.get('size'),
-                        'sdkfileid': video_info.get('sdkfileid') or video_info.get('fileid'),
-                        'md5sum': video_info.get('md5sum') or video_info.get('md5'),
-                        'filename': video_info.get('filename') or video_info.get('name'),
-                    }
-                    if (video_data.get('filesize') or 0) <= max_inline_file_size:
-                        video_base64 = await _safe_download(download_url)
-                        if video_base64:
-                            video_data['base64'] = video_base64
-                    videos.append(video_data)
-                elif item_type == 'link':
-                    links.append(item.get('link', {}))
-
-            if texts:
-                message_data['content'] = ' '.join(texts)  # 拼接所有 text
-            if images:
-                message_data['images'] = images
-                message_data['picurl'] = images[0]  # 只保留第一个 image
-            if files:
-                message_data['files'] = files
-                message_data['file'] = files[0]
-            if voices:
-                message_data['voices'] = voices
-                message_data['voice'] = voices[0]
-            if videos:
-                message_data['videos'] = videos
-                message_data['video'] = videos[0]
-            if links:
-                message_data['link'] = links[0]
-            if items:
-                message_data['attachments'] = items
-        else:
-            message_data['raw_msg'] = msg_json
-
-        # Extract user information
-        from_info = msg_json.get('from', {})
-        message_data['userid'] = from_info.get('userid', '')
-        message_data['username'] = (
-            from_info.get('alias', '') or from_info.get('name', '') or from_info.get('userid', '')
-        )
-
-        # Extract chat/group information
-        if msg_json.get('chattype', '') == 'group':
-            message_data['chatid'] = msg_json.get('chatid', '')
-            # Try to get group name if available
-            message_data['chatname'] = msg_json.get('chatname', '') or msg_json.get('chatid', '')
-
-        message_data['msgid'] = msg_json.get('msgid', '')
-
-        if msg_json.get('aibotid'):
-            message_data['aibotid'] = msg_json.get('aibotid', '')
-
-        return message_data
 
     async def _handle_message(self, event: wecombotevent.WecomBotEvent):
         """
@@ -712,39 +770,7 @@ class WecomBotClient:
         return decorator
 
     async def download_url_to_base64(self, download_url, encoding_aes_key):
-        async with httpx.AsyncClient() as client:
+        return await download_encrypted_file(download_url, encoding_aes_key, self.logger)
-            response = await client.get(download_url)
-            if response.status_code != 200:
-                await self.logger.error(f'failed to get file: {response.text}')
-                return None
-
-            encrypted_bytes = response.content
-
-        aes_key = base64.b64decode(encoding_aes_key + '=')  # base64 补齐
-        iv = aes_key[:16]
-
-        cipher = AES.new(aes_key, AES.MODE_CBC, iv)
-        decrypted = cipher.decrypt(encrypted_bytes)
-
-        pad_len = decrypted[-1]
-        decrypted = decrypted[:-pad_len]
-
-        if decrypted.startswith(b'\xff\xd8'):  # JPEG
-            mime_type = 'image/jpeg'
-        elif decrypted.startswith(b'\x89PNG'):  # PNG
-            mime_type = 'image/png'
-        elif decrypted.startswith((b'GIF87a', b'GIF89a')):  # GIF
-            mime_type = 'image/gif'
-        elif decrypted.startswith(b'BM'):  # BMP
-            mime_type = 'image/bmp'
-        elif decrypted.startswith(b'II*\x00') or decrypted.startswith(b'MM\x00*'):  # TIFF
-            mime_type = 'image/tiff'
-        else:
-            mime_type = 'application/octet-stream'
-
-        # 转 base64
-        base64_str = base64.b64encode(decrypted).decode('utf-8')
-        return f'data:{mime_type};base64,{base64_str}'
 
     async def run_task(self, host: str, port: int, *args, **kwargs):
         """

src/langbot/libs/wecom_ai_bot_api/ws_client.py

@@ -0,0 +1,596 @@
+"""WeChat Work AI Bot WebSocket long connection client.
+
+Implements the WebSocket protocol for receiving messages and sending replies
+via a persistent connection to wss://openws.work.weixin.qq.com, as an
+alternative to the HTTP callback (webhook) mode.
+
+Protocol reference: https://developer.work.weixin.qq.com/document/path/101463
+Official Node.js SDK: https://github.com/WecomTeam/aibot-node-sdk
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import secrets
+import time
+import traceback
+from typing import Any, Callable, Optional
+
+import aiohttp
+
+from langbot.libs.wecom_ai_bot_api import wecombotevent
+from langbot.libs.wecom_ai_bot_api.api import parse_wecom_bot_message
+from langbot.pkg.platform.logger import EventLogger
+
+DEFAULT_WS_URL = 'wss://openws.work.weixin.qq.com'
+
+# WebSocket frame command constants
+CMD_SUBSCRIBE = 'aibot_subscribe'
+CMD_HEARTBEAT = 'ping'
+CMD_MSG_CALLBACK = 'aibot_msg_callback'
+CMD_EVENT_CALLBACK = 'aibot_event_callback'
+CMD_RESPOND_MSG = 'aibot_respond_msg'
+CMD_RESPOND_WELCOME = 'aibot_respond_welcome_msg'
+CMD_RESPOND_UPDATE = 'aibot_respond_update_msg'
+CMD_SEND_MSG = 'aibot_send_msg'
+
+
+def _generate_req_id(prefix: str) -> str:
+    """Generate a unique request ID in the format: {prefix}_{timestamp}_{random}."""
+    ts = int(time.time() * 1000)
+    rand = secrets.token_hex(4)
+    return f'{prefix}_{ts}_{rand}'
+
+
+class WecomBotWsClient:
+    """WeChat Work AI Bot WebSocket long connection client.
+
+    Provides message receiving, streaming reply, proactive message sending,
+    and event callback handling over a persistent WebSocket connection.
+    """
+
+    def __init__(
+        self,
+        bot_id: str,
+        secret: str,
+        logger: EventLogger,
+        encoding_aes_key: str = '',
+        ws_url: str = DEFAULT_WS_URL,
+        heartbeat_interval: float = 30.0,
+        max_reconnect_attempts: int = -1,
+        reconnect_base_delay: float = 1.0,
+        reconnect_max_delay: float = 30.0,
+    ):
+        self.bot_id = bot_id
+        self.secret = secret
+        self.logger = logger
+        self.encoding_aes_key = encoding_aes_key
+        self.ws_url = ws_url
+        self.heartbeat_interval = heartbeat_interval
+        self.max_reconnect_attempts = max_reconnect_attempts
+        self.reconnect_base_delay = reconnect_base_delay
+        self.reconnect_max_delay = reconnect_max_delay
+
+        self._ws: Optional[aiohttp.ClientWebSocketResponse] = None
+        self._session: Optional[aiohttp.ClientSession] = None
+        self._running = False
+        self._heartbeat_task: Optional[asyncio.Task] = None
+        self._missed_pong_count = 0
+        self._max_missed_pong = 2
+        self._reconnect_attempts = 0
+
+        # Message handler registry (same pattern as WecomBotClient)
+        self._message_handlers: dict[str, list[Callable]] = {}
+        # Message deduplication
+        self._msg_id_map: dict[str, int] = {}
+
+        # Pending ACK futures: req_id -> Future[dict]
+        self._pending_acks: dict[str, asyncio.Future] = {}
+        # Per-req_id serial reply queues
+        self._reply_queues: dict[str, asyncio.Queue] = {}
+        self._reply_workers: dict[str, asyncio.Task] = {}
+        self._reply_ack_timeout = 5.0
+
+        # Stream ID tracking for WebSocket mode
+        self._stream_ids: dict[str, str] = {}  # msg_id -> req_id|stream_id
+        # Dedup: skip sending when content hasn't changed
+        self._stream_last_content: dict[str, str] = {}  # msg_id -> last content sent
+
+    # ── Public API ──────────────────────────────────────────────────
+
+    async def connect(self):
+        """Connect to WebSocket server with automatic reconnection.
+
+        This method blocks until disconnect() is called or max reconnect
+        attempts are exhausted.
+        """
+        self._running = True
+        self._reconnect_attempts = 0
+
+        while self._running:
+            try:
+                await self._connect_once()
+            except Exception:
+                if not self._running:
+                    break
+                await self.logger.error(f'WebSocket connection error: {traceback.format_exc()}')
+
+            if not self._running:
+                break
+
+            # Reconnect with exponential backoff
+            if self.max_reconnect_attempts != -1 and self._reconnect_attempts >= self.max_reconnect_attempts:
+                await self.logger.error(f'Max reconnect attempts reached ({self.max_reconnect_attempts}), giving up')
+                break
+
+            self._reconnect_attempts += 1
+            delay = min(
+                self.reconnect_base_delay * (2 ** (self._reconnect_attempts - 1)),
+                self.reconnect_max_delay,
+            )
+            await self.logger.info(f'Reconnecting in {delay:.1f}s (attempt {self._reconnect_attempts})...')
+            await asyncio.sleep(delay)
+
+    async def disconnect(self):
+        """Gracefully disconnect from the WebSocket server."""
+        self._running = False
+        if self._heartbeat_task and not self._heartbeat_task.done():
+            self._heartbeat_task.cancel()
+        for task in self._reply_workers.values():
+            if not task.done():
+                task.cancel()
+        if self._ws and not self._ws.closed:
+            await self._ws.close()
+        self._ws = None
+        if self._session and not self._session.closed:
+            await self._session.close()
+        self._session = None
+
+    def on_message(self, msg_type: str) -> Callable:
+        """Decorator to register a message handler.
+
+        Same interface as WecomBotClient.on_message for compatibility.
+
+        Args:
+            msg_type: 'single', 'group', or specific message type.
+        """
+
+        def decorator(func: Callable[[wecombotevent.WecomBotEvent], Any]):
+            if msg_type not in self._message_handlers:
+                self._message_handlers[msg_type] = []
+            self._message_handlers[msg_type].append(func)
+            return func
+
+        return decorator
+
+    async def reply_stream(
+        self,
+        req_id: str,
+        stream_id: str,
+        content: str,
+        finish: bool = False,
+    ) -> Optional[dict]:
+        """Send a streaming reply frame.
+
+        Args:
+            req_id: The req_id from the original message frame (must be passed through).
+            stream_id: The stream ID for this streaming session.
+            content: The content to send (supports Markdown).
+            finish: Whether this is the final chunk.
+
+        Returns:
+            The ACK frame dict, or None on failure.
+        """
+        body = {
+            'msgtype': 'stream',
+            'stream': {
+                'id': stream_id,
+                'finish': finish,
+                'content': content,
+            },
+        }
+        return await self._send_reply(req_id, body)
+
+    async def reply_text(self, req_id: str, content: str) -> Optional[dict]:
+        """Send a non-streaming text reply.
+
+        Args:
+            req_id: The req_id from the original message frame.
+            content: The text content to reply.
+
+        Returns:
+            The ACK frame dict, or None on failure.
+        """
+        body = {
+            'msgtype': 'markdown',
+            'markdown': {
+                'content': content,
+            },
+        }
+        return await self._send_reply(req_id, body)
+
+    async def send_message(self, chat_id: str, content: str, msgtype: str = 'markdown') -> Optional[dict]:
+        """Proactively send a message to a specified chat.
+
+        Args:
+            chat_id: The chat ID (userid for single chat, chatid for group chat).
+            content: The message content.
+            msgtype: Message type, 'markdown' by default.
+
+        Returns:
+            The ACK frame dict, or None on failure.
+        """
+        req_id = _generate_req_id(CMD_SEND_MSG)
+        body: dict[str, Any] = {
+            'chatid': chat_id,
+            'msgtype': msgtype,
+        }
+        if msgtype == 'markdown':
+            body['markdown'] = {'content': content}
+        elif msgtype == 'text':
+            body['text'] = {'content': content}
+        return await self._send_reply(req_id, body, cmd=CMD_SEND_MSG)
+
+    async def push_stream_chunk(self, msg_id: str, content: str, is_final: bool = False) -> bool:
+        """Push a streaming chunk for a given message ID.
+
+        Compatible interface with WecomBotClient.push_stream_chunk.
+
+        Args:
+            msg_id: The original message ID.
+            content: The cumulative content from the pipeline.
+            is_final: Whether this is the final chunk.
+
+        Returns:
+            True if the stream session exists and chunk was sent.
+        """
+        key = self._stream_ids.get(msg_id)
+        if not key:
+            return False
+        req_id, stream_id = key.split('|', 1)
+        try:
+            # Skip sending if content hasn't changed (e.g. during tool call argument streaming)
+            if not is_final and content == self._stream_last_content.get(msg_id):
+                return True
+            await self.reply_stream(req_id, stream_id, content, finish=is_final)
+            self._stream_last_content[msg_id] = content
+            if is_final:
+                self._stream_ids.pop(msg_id, None)
+                self._stream_last_content.pop(msg_id, None)
+            return True
+        except Exception:
+            await self.logger.error(f'Failed to push stream chunk: {traceback.format_exc()}')
+            return False
+
+    async def set_message(self, msg_id: str, content: str):
+        """Fallback: send content as a final stream chunk or direct reply.
+
+        Compatible interface with WecomBotClient.set_message.
+        """
+        handled = await self.push_stream_chunk(msg_id, content, is_final=True)
+        if not handled:
+            await self.logger.warning(f'No active stream for msg_id={msg_id}, message dropped')
+
+    # ── Connection lifecycle ────────────────────────────────────────
+
+    async def _connect_once(self):
+        """Establish a single WebSocket connection, authenticate, and listen."""
+        await self.logger.info(f'Connecting to {self.ws_url}...')
+
+        self._session = aiohttp.ClientSession()
+        try:
+            self._ws = await self._session.ws_connect(self.ws_url)
+            self._missed_pong_count = 0
+            self._reconnect_attempts = 0
+            await self.logger.info('WebSocket connected, sending auth...')
+
+            await self._send_auth()
+
+            # Wait for auth response
+            auth_ok = await self._wait_for_auth()
+            if not auth_ok:
+                await self.logger.error('Authentication failed')
+                return
+
+            await self.logger.info('Authenticated successfully')
+
+            # Start heartbeat
+            self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())
+
+            try:
+                await self._listen_loop()
+            finally:
+                if self._heartbeat_task and not self._heartbeat_task.done():
+                    self._heartbeat_task.cancel()
+                self._clear_pending_acks('Connection closed')
+        finally:
+            if self._ws and not self._ws.closed:
+                await self._ws.close()
+            self._ws = None
+            if self._session and not self._session.closed:
+                await self._session.close()
+            self._session = None
+
+    async def _send_auth(self):
+        """Send the authentication frame."""
+        frame = {
+            'cmd': CMD_SUBSCRIBE,
+            'headers': {'req_id': _generate_req_id(CMD_SUBSCRIBE)},
+            'body': {
+                'bot_id': self.bot_id,
+                'secret': self.secret,
+            },
+        }
+        await self._send_frame(frame)
+
+    async def _wait_for_auth(self) -> bool:
+        """Wait for and validate the authentication response."""
+        try:
+            msg = await asyncio.wait_for(self._ws.receive(), timeout=10.0)
+            if msg.type in (aiohttp.WSMsgType.TEXT,):
+                frame = json.loads(msg.data)
+                req_id = frame.get('headers', {}).get('req_id', '')
+                if req_id.startswith(CMD_SUBSCRIBE) and frame.get('errcode') == 0:
+                    return True
+                await self.logger.error(f'Auth response: errcode={frame.get("errcode")}, errmsg={frame.get("errmsg")}')
+                return False
+            elif msg.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.CLOSING):
+                await self.logger.error(f'WebSocket closed during auth: {msg.type}')
+                return False
+            await self.logger.error(f'Unexpected message type during auth: {msg.type}')
+            return False
+        except asyncio.TimeoutError:
+            await self.logger.error('Auth response timeout')
+            return False
+
+    async def _heartbeat_loop(self):
+        """Periodically send heartbeat pings."""
+        try:
+            while self._running and self._ws and not self._ws.closed:
+                await asyncio.sleep(self.heartbeat_interval)
+                if not self._running or not self._ws or self._ws.closed:
+                    break
+
+                if self._missed_pong_count >= self._max_missed_pong:
+                    await self.logger.warning(
+                        f'No heartbeat ack for {self._missed_pong_count} consecutive pings, connection considered dead'
+                    )
+                    await self._ws.close()
+                    break
+
+                self._missed_pong_count += 1
+                frame = {
+                    'cmd': CMD_HEARTBEAT,
+                    'headers': {'req_id': _generate_req_id(CMD_HEARTBEAT)},
+                }
+                try:
+                    await self._send_frame(frame)
+                except Exception:
+                    break
+        except asyncio.CancelledError:
+            pass
+
+    async def _listen_loop(self):
+        """Listen for incoming WebSocket frames and dispatch them."""
+        async for msg in self._ws:
+            if not self._running:
+                break
+            if msg.type == aiohttp.WSMsgType.TEXT:
+                try:
+                    frame = json.loads(msg.data)
+                    await self._handle_frame(frame)
+                except json.JSONDecodeError:
+                    await self.logger.error(f'Failed to parse WebSocket message: {str(msg.data)[:200]}')
+                except Exception:
+                    await self.logger.error(f'Error handling frame: {traceback.format_exc()}')
+            elif msg.type == aiohttp.WSMsgType.BINARY:
+                try:
+                    frame = json.loads(msg.data)
+                    await self._handle_frame(frame)
+                except Exception:
+                    await self.logger.error(f'Error handling binary frame: {traceback.format_exc()}')
+            elif msg.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.CLOSING):
+                await self.logger.warning(f'WebSocket connection closed: {msg.type}')
+                break
+
+    # ── Frame handling ──────────────────────────────────────────────
+
+    async def _handle_frame(self, frame: dict):
+        """Route an incoming frame to the appropriate handler."""
+        cmd = frame.get('cmd', '')
+
+        # Message push
+        if cmd == CMD_MSG_CALLBACK:
+            asyncio.create_task(self._handle_message_callback(frame))
+            return
+
+        # Event push
+        if cmd == CMD_EVENT_CALLBACK:
+            asyncio.create_task(self._handle_event_callback(frame))
+            return
+
+        # No cmd → response/ACK frame, dispatch by req_id prefix
+        req_id = frame.get('headers', {}).get('req_id', '')
+
+        # Check pending ACKs first
+        if req_id in self._pending_acks:
+            future = self._pending_acks.pop(req_id)
+            if not future.done():
+                future.set_result(frame)
+            return
+
+        # Heartbeat response
+        if req_id.startswith(CMD_HEARTBEAT):
+            if frame.get('errcode') == 0:
+                self._missed_pong_count = 0
+            return
+
+        # Unknown frame
+        await self.logger.warning(f'Unknown frame: {json.dumps(frame, ensure_ascii=False)[:200]}')
+
+    async def _handle_message_callback(self, frame: dict):
+        """Handle an incoming message callback frame."""
+        try:
+            body = frame.get('body', {})
+            req_id = frame.get('headers', {}).get('req_id', '')
+
+            # Parse message using shared logic
+            message_data = await parse_wecom_bot_message(body, self.encoding_aes_key, self.logger)
+            if not message_data:
+                return
+
+            # Generate stream_id for this message and store the mapping
+            stream_id = _generate_req_id('stream')
+            msg_id = message_data.get('msgid', '')
+            if msg_id:
+                self._stream_ids[msg_id] = f'{req_id}|{stream_id}'
+            message_data['stream_id'] = stream_id
+            message_data['req_id'] = req_id
+
+            event = wecombotevent.WecomBotEvent(message_data)
+            await self._dispatch_event(event)
+        except Exception:
+            await self.logger.error(f'Error in message callback: {traceback.format_exc()}')
+
+    async def _handle_event_callback(self, frame: dict):
+        """Handle an incoming event callback frame (enter_chat, template_card_event, etc.)."""
+        try:
+            body = frame.get('body', {})
+            req_id = frame.get('headers', {}).get('req_id', '')
+
+            event_info = body.get('event', {})
+            event_type = event_info.get('eventtype', '')
+
+            message_data = {
+                'msgtype': 'event',
+                'type': body.get('chattype', 'single'),
+                'event': event_info,
+                'eventtype': event_type,
+                'msgid': body.get('msgid', ''),
+                'aibotid': body.get('aibotid', ''),
+                'req_id': req_id,
+            }
+
+            from_info = body.get('from', {})
+            message_data['userid'] = from_info.get('userid', '')
+            message_data['username'] = from_info.get('alias', '') or from_info.get('userid', '')
+
+            if body.get('chatid'):
+                message_data['chatid'] = body.get('chatid', '')
+
+            event = wecombotevent.WecomBotEvent(message_data)
+
+            # Dispatch to event-specific handlers
+            if event_type in self._message_handlers:
+                for handler in self._message_handlers[event_type]:
+                    await handler(event)
+
+            # Also dispatch to generic 'event' handlers
+            if 'event' in self._message_handlers:
+                for handler in self._message_handlers['event']:
+                    await handler(event)
+
+        except Exception:
+            await self.logger.error(f'Error in event callback: {traceback.format_exc()}')
+
+    async def _dispatch_event(self, event: wecombotevent.WecomBotEvent):
+        """Dispatch a message event to registered handlers with deduplication."""
+        try:
+            message_id = event.message_id
+            if message_id in self._msg_id_map:
+                self._msg_id_map[message_id] += 1
+                return
+            self._msg_id_map[message_id] = 1
+
+            msg_type = event.type
+            if msg_type in self._message_handlers:
+                for handler in self._message_handlers[msg_type]:
+                    await handler(event)
+        except Exception:
+            await self.logger.error(f'Error dispatching event: {traceback.format_exc()}')
+
+    # ── Reply sending with serial queue ─────────────────────────────
+
+    async def _send_reply(
+        self,
+        req_id: str,
+        body: dict,
+        cmd: str = CMD_RESPOND_MSG,
+    ) -> Optional[dict]:
+        """Send a reply frame and wait for ACK.
+
+        Replies with the same req_id are serialized to maintain ordering.
+        """
+        if not self._ws or self._ws.closed:
+            return None
+
+        frame = {
+            'cmd': cmd,
+            'headers': {'req_id': req_id},
+            'body': body,
+        }
+
+        # Ensure serial delivery per req_id
+        if req_id not in self._reply_queues:
+            self._reply_queues[req_id] = asyncio.Queue()
+            self._reply_workers[req_id] = asyncio.create_task(self._reply_queue_worker(req_id))
+
+        future: asyncio.Future = asyncio.get_event_loop().create_future()
+        await self._reply_queues[req_id].put((frame, future))
+        return await future
+
+    async def _reply_queue_worker(self, req_id: str):
+        """Process reply queue items serially for a given req_id."""
+        queue = self._reply_queues[req_id]
+        try:
+            while self._running:
+                try:
+                    frame, future = await asyncio.wait_for(queue.get(), timeout=60.0)
+                except asyncio.TimeoutError:
+                    # Queue idle, clean up worker
+                    break
+
+                try:
+                    ack = await self._send_and_wait_ack(frame)
+                    if not future.done():
+                        future.set_result(ack)
+                except Exception as e:
+                    if not future.done():
+                        future.set_exception(e)
+        except asyncio.CancelledError:
+            pass
+        finally:
+            self._reply_queues.pop(req_id, None)
+            self._reply_workers.pop(req_id, None)
+
+    async def _send_and_wait_ack(self, frame: dict) -> Optional[dict]:
+        """Send a frame and wait for the corresponding ACK."""
+        req_id = frame['headers']['req_id']
+        ack_future: asyncio.Future = asyncio.get_event_loop().create_future()
+        self._pending_acks[req_id] = ack_future
+
+        try:
+            await self._send_frame(frame)
+            result = await asyncio.wait_for(ack_future, timeout=self._reply_ack_timeout)
+            if result.get('errcode', 0) != 0:
+                await self.logger.warning(
+                    f'Reply ACK error: errcode={result.get("errcode")}, errmsg={result.get("errmsg")}'
+                )
+            return result
+        except asyncio.TimeoutError:
+            self._pending_acks.pop(req_id, None)
+            await self.logger.warning(f'Reply ACK timeout ({self._reply_ack_timeout}s) for req_id={req_id}')
+            return None
+
+    async def _send_frame(self, frame: dict):
+        """Send a JSON frame over the WebSocket connection."""
+        if self._ws and not self._ws.closed:
+            await self._ws.send_str(json.dumps(frame, ensure_ascii=False))
+
+    def _clear_pending_acks(self, reason: str):
+        """Reject all pending ACK futures on disconnection."""
+        for req_id, future in self._pending_acks.items():
+            if not future.done():
+                future.set_exception(ConnectionError(reason))
+        self._pending_acks.clear()

src/langbot/pkg/api/http/controller/groups/files.py

@@ -13,9 +13,9 @@ from .. import group
 @group.group_class('files', '/api/v1/files')
 class FilesRouterGroup(group.RouterGroup):
     async def initialize(self) -> None:
-        @self.route('/image/<image_key>', methods=['GET'], auth_type=group.AuthType.NONE)
+        @self.route('/image/<path:image_key>', methods=['GET'], auth_type=group.AuthType.NONE)
         async def _(image_key: str) -> quart.Response:
-            if '/' in image_key or '\\' in image_key:
+            if '..' in image_key or '\\' in image_key:
                 return quart.Response(status=404)
 
             if not await self.ap.storage_mgr.storage_provider.exists(image_key):

src/langbot/pkg/api/http/service/bot.py

@@ -70,12 +70,17 @@ class BotService:
             'lark',
         ]:
             webhook_prefix = self.ap.instance_config.data['api'].get('webhook_prefix', 'http://127.0.0.1:5300')
+            extra_webhook_prefix = self.ap.instance_config.data['api'].get('extra_webhook_prefix', '')
             webhook_url = f'/bots/{bot_uuid}'
             adapter_runtime_values['webhook_url'] = webhook_url
             adapter_runtime_values['webhook_full_url'] = f'{webhook_prefix}{webhook_url}'
+            adapter_runtime_values['extra_webhook_full_url'] = (
+                f'{extra_webhook_prefix}{webhook_url}' if extra_webhook_prefix else ''
+            )
         else:
             adapter_runtime_values['webhook_url'] = None
             adapter_runtime_values['webhook_full_url'] = None
+            adapter_runtime_values['extra_webhook_full_url'] = None
 
         persistence_bot['adapter_runtime_values'] = adapter_runtime_values
 

src/langbot/pkg/api/http/service/model.py

@@ -105,11 +105,16 @@ class LLMModelsService:
                 )
             )
             pipeline = result.first()
-            if pipeline is not None and pipeline.config['ai']['local-agent']['model'] == '':
+            if pipeline is not None:
-                pipeline_config = pipeline.config
+                model_config = pipeline.config.get('ai', {}).get('local-agent', {}).get('model', {})
-                pipeline_config['ai']['local-agent']['model'] = model_data['uuid']
+                if not model_config.get('primary', ''):
-                pipeline_data = {'config': pipeline_config}
+                    pipeline_config = pipeline.config
-                await self.ap.pipeline_service.update_pipeline(pipeline.uuid, pipeline_data)
+                    pipeline_config['ai']['local-agent']['model'] = {
+                        'primary': model_data['uuid'],
+                        'fallbacks': [],
+                    }
+                    pipeline_data = {'config': pipeline_config}
+                    await self.ap.pipeline_service.update_pipeline(pipeline.uuid, pipeline_data)
 
         return model_data['uuid']
 

src/langbot/pkg/core/app.py

@@ -9,6 +9,7 @@ from ..platform import botmgr as im_mgr
 from ..platform.webhook_pusher import WebhookPusher
 from ..provider.session import sessionmgr as llm_session_mgr
 from ..provider.modelmgr import modelmgr as llm_model_mgr
+
 from langbot.pkg.provider.tools import toolmgr as llm_tool_mgr
 from ..config import manager as config_mgr
 from ..command import cmdmgr
@@ -30,6 +31,7 @@ from ..api.http.service import mcp as mcp_service
 from ..api.http.service import apikey as apikey_service
 from ..api.http.service import webhook as webhook_service
 from ..api.http.service import monitoring as monitoring_service
+
 from ..discover import engine as discover_engine
 from ..storage import mgr as storagemgr
 from ..utils import logcache

src/langbot/pkg/core/stages/load_config.py

@@ -74,20 +74,26 @@ def _apply_env_overrides_to_config(cfg: dict) -> dict:
         current = cfg
 
         for i, key in enumerate(keys):
-            if not isinstance(current, dict) or key not in current:
+            if not isinstance(current, dict):
                 break
 
             if i == len(keys) - 1:
-                # At the final key - check if it's a scalar value
+                # At the final key
-                if isinstance(current[key], (dict, list)):
+                if key in current:
-                    # Skip dict and list types
+                    if isinstance(current[key], (dict, list)):
-                    pass
+                        # Skip dict and list types
+                        pass
+                    else:
+                        # Valid scalar value - convert and set it
+                        converted_value = convert_value(env_value, current[key])
+                        current[key] = converted_value
                 else:
-                    # Valid scalar value - convert and set it
+                    # Key doesn't exist yet - create it as string
-                    converted_value = convert_value(env_value, current[key])
+                    current[key] = env_value
-                    current[key] = converted_value
             else:
-                # Navigate deeper
+                # Navigate deeper - create intermediate dict if needed
+                if key not in current:
+                    current[key] = {}
                 current = current[key]
 
     return cfg
@@ -146,16 +152,50 @@ class LoadConfigStage(stage.BootingStage):
         await ap.instance_config.dump_config()
 
         # load or generate instance id
-        ap.instance_id = await config.load_json_config(
+        # Priority:
-            'data/labels/instance_id.json',
+        # 1. system.instance_id from config.yaml (can be set via SYSTEM__INSTANCE_ID env var)
-            template_data={
+        # 2. data/labels/instance_id.json (if file exists)
-                'instance_id': f'instance_{str(uuid.uuid4())}',
+        # 3. Generate new and save to file
-                'instance_create_ts': int(time.time()),
+        config_instance_id = ap.instance_config.data.get('system', {}).get('instance_id', '')
-            },
-            completion=False,
-        )
 
-        constants.instance_id = ap.instance_id.data['instance_id']
+        if config_instance_id:
+            # Use the instance_id from config.yaml
+            constants.instance_id = config_instance_id
+            # Still load/create the file for backward compat, but don't use its value
+            ap.instance_id = await config.load_json_config(
+                'data/labels/instance_id.json',
+                template_data={
+                    'instance_id': f'instance_{str(uuid.uuid4())}',
+                    'instance_create_ts': int(time.time()),
+                },
+                completion=False,
+            )
+        else:
+            # Try loading file-based instance id
+            instance_id_path = os.path.join('data', 'labels', 'instance_id.json')
+            if os.path.exists(instance_id_path):
+                # File exists, read it
+                ap.instance_id = await config.load_json_config(
+                    'data/labels/instance_id.json',
+                    template_data={
+                        'instance_id': '',
+                        'instance_create_ts': 0,
+                    },
+                    completion=False,
+                )
+                constants.instance_id = ap.instance_id.data['instance_id']
+            else:
+                # Neither config nor file, generate new and save to file
+                new_id = f'instance_{str(uuid.uuid4())}'
+                ap.instance_id = await config.load_json_config(
+                    'data/labels/instance_id.json',
+                    template_data={
+                        'instance_id': new_id,
+                        'instance_create_ts': int(time.time()),
+                    },
+                    completion=False,
+                )
+                constants.instance_id = new_id
         constants.edition = ap.instance_config.data.get('system', {}).get('edition', 'community')
 
         print(f'LangBot instance id: {constants.instance_id}')

src/langbot/pkg/persistence/migrations/dbm023_model_fallback_config.py

@@ -0,0 +1,102 @@
+from .. import migration
+
+import sqlalchemy
+import json
+
+
+@migration.migration_class(23)
+class DBMigrateModelFallbackConfig(migration.DBMigration):
+    """Convert model field from plain UUID string to object with primary/fallbacks"""
+
+    async def upgrade(self):
+        """Upgrade"""
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
+
+        current_version = self.ap.ver_mgr.get_current_version()
+
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
+
+            if 'ai' not in config or 'local-agent' not in config['ai']:
+                continue
+
+            local_agent = config['ai']['local-agent']
+            changed = False
+
+            # Convert model from string to object
+            model_value = local_agent.get('model', '')
+            if isinstance(model_value, str):
+                local_agent['model'] = {
+                    'primary': model_value,
+                    'fallbacks': [],
+                }
+                changed = True
+
+            # Remove leftover fallback-models field if present
+            if 'fallback-models' in local_agent:
+                del local_agent['fallback-models']
+                changed = True
+
+            if not changed:
+                continue
+
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
+            if self.ap.persistence_mgr.db.name == 'postgresql':
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                )
+
+    async def downgrade(self):
+        """Downgrade"""
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
+
+        current_version = self.ap.ver_mgr.get_current_version()
+
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
+
+            if 'ai' not in config or 'local-agent' not in config['ai']:
+                continue
+
+            local_agent = config['ai']['local-agent']
+
+            # Convert model from object back to string
+            model_value = local_agent.get('model', '')
+            if isinstance(model_value, dict):
+                local_agent['model'] = model_value.get('primary', '')
+            else:
+                continue
+
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
+            if self.ap.persistence_mgr.db.name == 'postgresql':
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                )

src/langbot/pkg/persistence/migrations/dbm024_wecombot_websocket_mode.py

@@ -0,0 +1,49 @@
+from .. import migration
+
+import sqlalchemy
+import json
+
+
+@migration.migration_class(24)
+class DBMigrateWecomBotWebSocketMode(migration.DBMigration):
+    """Add enable-webhook field to existing wecombot adapter configs.
+
+    Existing wecombot bots were all using webhook mode, so we set
+    enable-webhook=true to preserve their behavior after the new
+    WebSocket long connection mode is introduced as default.
+    """
+
+    async def upgrade(self):
+        """Upgrade"""
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text("SELECT uuid, adapter_config FROM bots WHERE adapter = 'wecombot'")
+        )
+        bots = result.fetchall()
+
+        for bot_row in bots:
+            bot_uuid = bot_row[0]
+            adapter_config = json.loads(bot_row[1]) if isinstance(bot_row[1], str) else bot_row[1]
+
+            if 'enable-webhook' in adapter_config:
+                continue
+
+            # Determine mode based on existing config: if webhook fields are present, keep webhook mode
+            has_webhook_config = bool(
+                adapter_config.get('Token') and adapter_config.get('EncodingAESKey') and adapter_config.get('Corpid')
+            )
+            adapter_config['enable-webhook'] = has_webhook_config
+
+            if self.ap.persistence_mgr.db.name == 'postgresql':
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text('UPDATE bots SET adapter_config = :config::jsonb WHERE uuid = :uuid'),
+                    {'config': json.dumps(adapter_config), 'uuid': bot_uuid},
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text('UPDATE bots SET adapter_config = :config WHERE uuid = :uuid'),
+                    {'config': json.dumps(adapter_config), 'uuid': bot_uuid},
+                )
+
+    async def downgrade(self):
+        """Downgrade"""
+        pass

src/langbot/pkg/pipeline/preproc/preproc.py

@@ -36,17 +36,36 @@ class PreProcessor(stage.PipelineStage):
         session = await self.ap.sess_mgr.get_session(query)
 
         # When not local-agent, llm_model is None
-        try:
+        llm_model = None
-            llm_model = (
+        if selected_runner == 'local-agent':
-                await self.ap.model_mgr.get_model_by_uuid(query.pipeline_config['ai']['local-agent']['model'])
+            # Read model config — new format is { primary: str, fallbacks: [str] },
-                if selected_runner == 'local-agent'
+            # but handle legacy plain string for backward compatibility
-                else None
+            model_config = query.pipeline_config['ai']['local-agent'].get('model', {})
-            )
+            if isinstance(model_config, str):
-        except ValueError:
+                # Legacy format: plain UUID string
-            self.ap.logger.warning(
+                primary_uuid = model_config
-                f'LLM model {query.pipeline_config["ai"]["local-agent"]["model"] + " "}not found or not configured'
+                fallback_uuids = []
-            )
+            else:
-            llm_model = None
+                primary_uuid = model_config.get('primary', '')
+                fallback_uuids = model_config.get('fallbacks', [])
+
+            if primary_uuid:
+                try:
+                    llm_model = await self.ap.model_mgr.get_model_by_uuid(primary_uuid)
+                except ValueError:
+                    self.ap.logger.warning(f'LLM model {primary_uuid} not found or not configured')
+
+            # Resolve fallback model UUIDs
+            if fallback_uuids:
+                valid_fallbacks = []
+                for fb_uuid in fallback_uuids:
+                    try:
+                        await self.ap.model_mgr.get_model_by_uuid(fb_uuid)
+                        valid_fallbacks.append(fb_uuid)
+                    except ValueError:
+                        self.ap.logger.warning(f'Fallback model {fb_uuid} not found, skipping')
+                if valid_fallbacks:
+                    query.variables['_fallback_model_uuids'] = valid_fallbacks
 
         conversation = await self.ap.sess_mgr.get_conversation(
             query,
@@ -61,20 +80,28 @@ class PreProcessor(stage.PipelineStage):
         query.prompt = conversation.prompt.copy()
         query.messages = conversation.messages.copy()
 
-        if selected_runner == 'local-agent' and llm_model:
+        if selected_runner == 'local-agent':
             query.use_funcs = []
-            query.use_llm_model_uuid = llm_model.model_entity.uuid
+            if llm_model:
+                query.use_llm_model_uuid = llm_model.model_entity.uuid
 
-            if llm_model.model_entity.abilities.__contains__('func_call'):
+                if llm_model.model_entity.abilities.__contains__('func_call'):
-                # Get bound plugins and MCP servers for filtering tools
+                    # Get bound plugins and MCP servers for filtering tools
+                    bound_plugins = query.variables.get('_pipeline_bound_plugins', None)
+                    bound_mcp_servers = query.variables.get('_pipeline_bound_mcp_servers', None)
+                    query.use_funcs = await self.ap.tool_mgr.get_all_tools(bound_plugins, bound_mcp_servers)
+
+                    self.ap.logger.debug(f'Bound plugins: {bound_plugins}')
+                    self.ap.logger.debug(f'Bound MCP servers: {bound_mcp_servers}')
+                    self.ap.logger.debug(f'Use funcs: {query.use_funcs}')
+
+            # If primary model doesn't support func_call but fallback models exist,
+            # load tools anyway since fallback models may support them
+            if not query.use_funcs and query.variables.get('_fallback_model_uuids'):
                 bound_plugins = query.variables.get('_pipeline_bound_plugins', None)
                 bound_mcp_servers = query.variables.get('_pipeline_bound_mcp_servers', None)
                 query.use_funcs = await self.ap.tool_mgr.get_all_tools(bound_plugins, bound_mcp_servers)
 
-                self.ap.logger.debug(f'Bound plugins: {bound_plugins}')
-                self.ap.logger.debug(f'Bound MCP servers: {bound_mcp_servers}')
-                self.ap.logger.debug(f'Use funcs: {query.use_funcs}')
-
         sender_name = ''
 
         if isinstance(query.message_event, platform_events.GroupMessage):
@@ -149,6 +176,16 @@ class PreProcessor(stage.PipelineStage):
         query.variables['user_message_text'] = plain_text
 
         query.user_message = provider_message.Message(role='user', content=content_list)
+
+        # Extract knowledge base UUIDs into query variables so plugins can modify them
+        # during PromptPreProcessing before the runner performs retrieval.
+        kb_uuids = query.pipeline_config['ai']['local-agent'].get('knowledge-bases', [])
+        if not kb_uuids:
+            old_kb_uuid = query.pipeline_config['ai']['local-agent'].get('knowledge-base', '')
+            if old_kb_uuid and old_kb_uuid != '__none__':
+                kb_uuids = [old_kb_uuid]
+        query.variables['_knowledge_base_uuids'] = list(kb_uuids)
+
         # =========== 触发事件 PromptPreProcessing
 
         event = events.PromptPreProcessing(

src/langbot/pkg/platform/sources/lark.py

@@ -575,6 +575,127 @@ class LarkMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
 
 
 class LarkEventConverter(abstract_platform_adapter.AbstractEventConverter):
+    _processed_thread_quote_cache: typing.ClassVar[dict[str, float]] = {}
+    _processed_thread_quote_cache_max_size: typing.ClassVar[int] = 4096
+    _processed_thread_quote_cache_ttl_seconds: typing.ClassVar[int] = 86400
+
+    @classmethod
+    def _prune_processed_thread_quote_cache(cls, now: typing.Optional[float] = None) -> None:
+        if now is None:
+            now = time.time()
+
+        expire_before = now - cls._processed_thread_quote_cache_ttl_seconds
+        while cls._processed_thread_quote_cache:
+            oldest_key, oldest_ts = next(iter(cls._processed_thread_quote_cache.items()))
+            if oldest_ts >= expire_before:
+                break
+            cls._processed_thread_quote_cache.pop(oldest_key, None)
+
+        while len(cls._processed_thread_quote_cache) > cls._processed_thread_quote_cache_max_size:
+            oldest_key = next(iter(cls._processed_thread_quote_cache))
+            cls._processed_thread_quote_cache.pop(oldest_key, None)
+
+    @classmethod
+    def _mark_thread_quote_processed(cls, thread_id: str) -> None:
+        now = time.time()
+        cls._prune_processed_thread_quote_cache(now)
+        cls._processed_thread_quote_cache[thread_id] = now
+
+    @classmethod
+    def _extract_quote_message_id(cls, message: EventMessage) -> typing.Optional[str]:
+        """
+        Extract the message ID to quote from the given message.
+
+        Rules:
+        - First thread reply in a topic: return parent_id and mark topic as processed
+        - Follow-up thread replies in the same topic: return None
+        - Non-thread message: return parent_id if valid (non-empty, different from message_id)
+
+        Thread reply state is kept in a bounded TTL cache to avoid unbounded memory growth.
+        """
+        parent_id = getattr(message, 'parent_id', None)
+        if not parent_id:
+            return None
+
+        message_id = getattr(message, 'message_id', None)
+        if parent_id == message_id:
+            return None
+
+        thread_id = getattr(message, 'thread_id', None)
+        if thread_id:
+            cls._prune_processed_thread_quote_cache()
+            if thread_id in cls._processed_thread_quote_cache:
+                return None
+            cls._mark_thread_quote_processed(thread_id)
+
+        return parent_id
+
+    @staticmethod
+    def _build_event_message_from_message_item(message_item: Message) -> typing.Optional[EventMessage]:
+        """
+        Build EventMessage from SDK typed Message item.
+
+        Returns None if body or content is missing.
+        """
+        body = getattr(message_item, 'body', None)
+        if not body:
+            return None
+
+        content = getattr(body, 'content', None)
+        if not content:
+            return None
+
+        event_data = {
+            'message_id': message_item.message_id,
+            'message_type': message_item.msg_type,
+            'content': content,
+            'create_time': message_item.create_time,
+            'mentions': getattr(message_item, 'mentions', []) or [],
+        }
+
+        # Preserve thread-related fields
+        if hasattr(message_item, 'parent_id') and message_item.parent_id:
+            event_data['parent_id'] = message_item.parent_id
+        if hasattr(message_item, 'root_id') and message_item.root_id:
+            event_data['root_id'] = message_item.root_id
+        if hasattr(message_item, 'thread_id') and message_item.thread_id:
+            event_data['thread_id'] = message_item.thread_id
+        if hasattr(message_item, 'chat_id') and message_item.chat_id:
+            event_data['chat_id'] = message_item.chat_id
+
+        return EventMessage(event_data)
+
+    @staticmethod
+    async def _fetch_quoted_message(
+        quote_message_id: str,
+        api_client: lark_oapi.Client,
+    ) -> typing.Optional[platform_message.MessageChain]:
+        """
+        Fetch the quoted message and convert to MessageChain.
+
+        Returns None if:
+        - API call fails
+        - Response items is empty
+        - Message item normalization fails
+        """
+        request = GetMessageRequest.builder().message_id(quote_message_id).build()
+        response = await api_client.im.v1.message.aget(request)
+
+        if not response.success():
+            return None
+
+        items = getattr(response.data, 'items', None)
+        if not items:
+            return None
+
+        message_item = items[0]
+        event_message = LarkEventConverter._build_event_message_from_message_item(message_item)
+        if event_message is None:
+            return None
+
+        quote_chain = await LarkMessageConverter.target2yiri(event_message, api_client)
+        return quote_chain
+
     @staticmethod
     async def yiri2target(
         event: platform_events.MessageEvent,
@@ -587,6 +708,23 @@ class LarkEventConverter(abstract_platform_adapter.AbstractEventConverter):
     ) -> platform_events.Event:
         message_chain = await LarkMessageConverter.target2yiri(event.event.message, api_client)
 
+        # Check for quote/reply message
+        quote_message_id = LarkEventConverter._extract_quote_message_id(event.event.message)
+        if quote_message_id:
+            quote_chain = await LarkEventConverter._fetch_quoted_message(quote_message_id, api_client)
+            if quote_chain:
+                # Filter out Source component from quoted chain, keep only content
+                quote_origin = platform_message.MessageChain(
+                    [comp for comp in quote_chain if not isinstance(comp, platform_message.Source)]
+                )
+                if quote_origin:
+                    message_chain.append(
+                        platform_message.Quote(
+                            message_id=quote_message_id,
+                            origin=quote_origin,
+                        )
+                    )
+
         if event.event.message.chat_type == 'p2p':
             return platform_events.FriendMessage(
                 sender=platform_entities.Friend(
@@ -770,6 +908,32 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             self.request_tenant_access_token(tenant_key)
         return self.tenant_access_tokens.get(tenant_key)['token'] if self.tenant_access_tokens.get(tenant_key) else None
 
+    def get_launcher_id(self, event: platform_events.MessageEvent) -> str | None:
+        """
+        Get topic-scoped launcher_id for thread-aware session isolation.
+
+        For group thread messages, returns "{group_id}_{thread_id}"
+        to ensure conversation context stays stable per topic.
+
+        Returns None for non-thread messages or P2P messages.
+        """
+        source_event = getattr(event.source_platform_object, 'event', None)
+        if not source_event:
+            return None
+
+        message = getattr(source_event, 'message', None)
+        if not message:
+            return None
+
+        thread_id = getattr(message, 'thread_id', None)
+        if not thread_id:
+            return None
+
+        if isinstance(event, platform_events.GroupMessage):
+            return f'{event.group.id}_{thread_id}'
+
+        return None
+
     def build_api_client(self, config):
         app_id = config['app_id']
         app_secret = config['app_secret']

src/langbot/pkg/platform/sources/openclaw_weixin.py

@@ -0,0 +1,577 @@
+"""OpenClaw WeChat adapter for LangBot.
+
+Uses the OpenClaw WeChat HTTP JSON API (long-poll getUpdates + sendMessage)
+to integrate personal WeChat accounts with LangBot.
+
+Reference: https://github.com/epiral/weixin-bot
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import traceback
+import typing
+
+import pydantic
+import sqlalchemy
+
+from langbot.libs.openclaw_weixin_api.client import (
+    DEFAULT_BASE_URL,
+    SESSION_EXPIRED_ERRCODE,
+    OpenClawWeixinClient,
+)
+from langbot.libs.openclaw_weixin_api.types import (
+    MessageItem,
+    WeixinMessage,
+)
+from langbot.pkg.entity.persistence import bot as persistence_bot
+
+import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
+import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
+import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
+import langbot_plugin.api.entities.builtin.platform.events as platform_events
+import langbot_plugin.api.entities.builtin.platform.message as platform_message
+
+
+class OpenClawWeixinMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
+    """Converts between LangBot MessageChain and OpenClaw WeChat message items."""
+
+    @staticmethod
+    async def yiri2target(message_chain: platform_message.MessageChain) -> list[dict]:
+        """Convert LangBot MessageChain to a list of OpenClaw message item dicts."""
+        items = []
+        for component in message_chain:
+            if isinstance(component, platform_message.Plain):
+                items.append({'type': MessageItem.TEXT, 'text_item': {'text': component.text}})
+            elif isinstance(component, platform_message.Image):
+                # OpenClaw WeChat only supports text messages without CDN upload.
+                # For images, we send a placeholder text with the URL if available.
+                if component.url:
+                    items.append(
+                        {
+                            'type': MessageItem.TEXT,
+                            'text_item': {'text': f'[Image: {component.url}]'},
+                        }
+                    )
+                elif component.base64:
+                    items.append(
+                        {
+                            'type': MessageItem.TEXT,
+                            'text_item': {'text': '[Image]'},
+                        }
+                    )
+            elif isinstance(component, platform_message.File):
+                if component.name:
+                    items.append(
+                        {
+                            'type': MessageItem.TEXT,
+                            'text_item': {'text': f'[File: {component.name}]'},
+                        }
+                    )
+            elif isinstance(component, platform_message.Forward):
+                for node in component.node_list:
+                    if node.message_chain:
+                        items.extend(await OpenClawWeixinMessageConverter.yiri2target(node.message_chain))
+        return items
+
+    @staticmethod
+    async def target2yiri(
+        msg: WeixinMessage,
+    ) -> platform_message.MessageChain:
+        """Convert an OpenClaw WeixinMessage to LangBot MessageChain."""
+        components: list[platform_message.MessageComponent] = []
+
+        if not msg.item_list:
+            return platform_message.MessageChain(components)
+
+        for item in msg.item_list:
+            if item.type == MessageItem.TEXT and item.text_item and item.text_item.text:
+                text = item.text_item.text
+
+                # Handle quoted messages
+                if item.ref_msg:
+                    ref_parts = []
+                    if item.ref_msg.title:
+                        ref_parts.append(item.ref_msg.title)
+                    if item.ref_msg.message_item:
+                        ref_item = item.ref_msg.message_item
+                        if ref_item.text_item and ref_item.text_item.text:
+                            ref_parts.append(ref_item.text_item.text)
+                    if ref_parts:
+                        components.append(
+                            platform_message.Quote(
+                                sender_id='',
+                                origin=platform_message.MessageChain(
+                                    [platform_message.Plain(text=' | '.join(ref_parts))]
+                                ),
+                            )
+                        )
+
+                components.append(platform_message.Plain(text=text))
+
+            elif item.type == MessageItem.IMAGE and item.image_item:
+                if hasattr(item.image_item, '_downloaded_bytes') and item.image_item._downloaded_bytes:
+                    b64 = base64.b64encode(item.image_item._downloaded_bytes).decode('utf-8')
+                    components.append(platform_message.Image(base64=f'data:image/jpeg;base64,{b64}'))
+                else:
+                    components.append(platform_message.Unknown(text='[Image]'))
+
+            elif item.type == MessageItem.VOICE and item.voice_item:
+                # Voice with speech-to-text: use the transcribed text
+                if item.voice_item.text:
+                    components.append(platform_message.Plain(text=item.voice_item.text))
+                else:
+                    components.append(platform_message.Unknown(text='[Voice]'))
+
+            # TODO: enable after full testing
+            # elif item.type == MessageItem.VOICE and item.voice_item:
+            #     if item.voice_item.text:
+            #         components.append(platform_message.Plain(text=item.voice_item.text))
+            #     elif hasattr(item.voice_item, '_downloaded_bytes') and item.voice_item._downloaded_bytes:
+            #         b64 = base64.b64encode(item.voice_item._downloaded_bytes).decode('utf-8')
+            #         components.append(
+            #             platform_message.Voice(
+            #                 base64=b64,
+            #                 length=item.voice_item.playtime or 0,
+            #             )
+            #         )
+            #     else:
+            #         components.append(
+            #             platform_message.Voice(
+            #                 length=item.voice_item.playtime or 0,
+            #             )
+            #         )
+
+            elif item.type == MessageItem.FILE and item.file_item:
+                components.append(platform_message.Unknown(text=f'[File: {item.file_item.file_name or ""}]'))
+
+            # TODO: enable after full testing
+            # elif item.type == MessageItem.FILE and item.file_item:
+            #     file_name = item.file_item.file_name or ''
+            #     file_size = int(item.file_item.len) if item.file_item.len else 0
+            #     if hasattr(item.file_item, '_downloaded_bytes') and item.file_item._downloaded_bytes:
+            #         b64 = base64.b64encode(item.file_item._downloaded_bytes).decode('utf-8')
+            #         components.append(
+            #             platform_message.File(
+            #                 name=file_name,
+            #                 size=file_size,
+            #                 base64=b64,
+            #             )
+            #         )
+            #     else:
+            #         components.append(
+            #             platform_message.File(
+            #                 name=file_name,
+            #                 size=file_size,
+            #             )
+            #         )
+
+            elif item.type == MessageItem.VIDEO and item.video_item:
+                components.append(platform_message.Unknown(text='[Video]'))
+
+            # TODO: enable after full testing
+            # elif item.type == MessageItem.VIDEO and item.video_item:
+            #     if hasattr(item.video_item, '_downloaded_bytes') and item.video_item._downloaded_bytes:
+            #         b64 = base64.b64encode(item.video_item._downloaded_bytes).decode('utf-8')
+            #         components.append(
+            #             platform_message.File(
+            #                 name='video.mp4',
+            #                 size=item.video_item.video_size or 0,
+            #                 base64=b64,
+            #             )
+            #         )
+            #     else:
+            #         components.append(
+            #             platform_message.File(
+            #                 name='video.mp4',
+            #                 size=item.video_item.video_size or 0,
+            #             )
+            #         )
+
+            else:
+                components.append(platform_message.Unknown(text='[Unknown message type]'))
+
+        return platform_message.MessageChain(components)
+
+
+class OpenClawWeixinEventConverter(abstract_platform_adapter.AbstractEventConverter):
+    """Converts OpenClaw WeChat messages to LangBot events."""
+
+    @staticmethod
+    async def yiri2target(event: platform_events.MessageEvent) -> dict:
+        return event.source_platform_object
+
+    @staticmethod
+    async def target2yiri(msg: WeixinMessage) -> typing.Optional[platform_events.MessageEvent]:
+        """Convert an inbound WeixinMessage to a LangBot event."""
+        if msg.message_type != WeixinMessage.TYPE_USER:
+            return None
+
+        from_user_id = msg.from_user_id or ''
+        if not from_user_id:
+            return None
+
+        message_chain = await OpenClawWeixinMessageConverter.target2yiri(msg)
+        if not message_chain:
+            return None
+
+        timestamp = (msg.create_time_ms or 0) / 1000.0
+
+        return platform_events.FriendMessage(
+            sender=platform_entities.Friend(
+                id=from_user_id,
+                nickname=from_user_id,
+                remark='',
+            ),
+            message_chain=message_chain,
+            time=timestamp,
+            source_platform_object=msg,
+        )
+
+
+class OpenClawWeixinAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
+    """LangBot adapter for OpenClaw WeChat (long-poll based)."""
+
+    name: str = 'openclaw-weixin'
+
+    client: OpenClawWeixinClient = pydantic.Field(exclude=True)
+
+    config: dict
+
+    message_converter: OpenClawWeixinMessageConverter = OpenClawWeixinMessageConverter()
+    event_converter: OpenClawWeixinEventConverter = OpenClawWeixinEventConverter()
+
+    # context_token cache: from_user_id -> context_token
+    _context_tokens: dict[str, str] = pydantic.PrivateAttr(default_factory=dict)
+
+    _polling: bool = pydantic.PrivateAttr(default=False)
+    _poll_task: typing.Optional[asyncio.Task] = pydantic.PrivateAttr(default=None)
+    _bot_uuid: typing.Optional[str] = pydantic.PrivateAttr(default=None)
+
+    listeners: typing.Dict[
+        typing.Type[platform_events.Event],
+        typing.Callable[[platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None],
+    ] = {}
+
+    def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger):
+        client = OpenClawWeixinClient(
+            base_url=config.get('base_url', DEFAULT_BASE_URL),
+            token=config.get('token', ''),
+        )
+        super().__init__(
+            config=config,
+            logger=logger,
+            client=client,
+            bot_account_id='',
+            listeners={},
+            name='openclaw-weixin',
+        )
+
+    def set_bot_uuid(self, bot_uuid: str):
+        """Called by BotManager to provide the bot's UUID for config persistence."""
+        self._bot_uuid = bot_uuid
+
+    async def _persist_config(self) -> None:
+        """Persist current self.config to the database so token survives restart."""
+        if not self._bot_uuid:
+            return
+        try:
+            ap = self.logger.ap
+            await ap.persistence_mgr.execute_async(
+                sqlalchemy.update(persistence_bot.Bot)
+                .where(persistence_bot.Bot.uuid == self._bot_uuid)
+                .values(adapter_config=self.config)
+            )
+        except Exception as e:
+            await self.logger.warning(f'Failed to persist adapter config: {e}')
+
+    async def _do_login(self) -> None:
+        """Run the QR code login flow via client.login() and update config."""
+        adapter_logger = self.logger
+
+        async def _on_qrcode(qr_base64: str, _qr_url: str):
+            await adapter_logger.info(
+                f'Please scan the QR code to login WeChat: {_qr_url}',
+                images=[platform_message.Image(base64=qr_base64)],
+            )
+
+        login_result = await self.client.login(
+            on_qrcode=_on_qrcode,
+        )
+
+        # client.login() already updates client.token and client.base_url
+        self.config['token'] = login_result.token
+        self.config['base_url'] = login_result.base_url
+        if login_result.account_id:
+            self.config['account_id'] = login_result.account_id
+
+        await self.logger.info(f'WeChat login successful! account_id={login_result.account_id}')
+
+        # Persist token to database so it survives restart
+        await self._persist_config()
+
+    async def send_message(
+        self,
+        target_type: str,
+        target_id: str,
+        message: platform_message.MessageChain,
+    ):
+        """Send a message to a user."""
+        context_token = self._context_tokens.get(target_id, '')
+
+        for component in message:
+            try:
+                if isinstance(component, platform_message.Plain):
+                    if component.text:
+                        await self.client.send_text(target_id, component.text, context_token)
+
+                elif isinstance(component, platform_message.Image):
+                    img_bytes, _ = await component.get_bytes()
+                    await self.client.send_image(target_id, img_bytes, context_token)
+
+                elif isinstance(component, platform_message.File):
+                    file_bytes = await self._get_component_bytes(component)
+                    if file_bytes:
+                        await self.client.send_file(target_id, file_bytes, component.name or 'file', context_token)
+
+                elif isinstance(component, platform_message.Voice):
+                    voice_bytes = await self._get_component_bytes(component)
+                    if voice_bytes:
+                        await self.client.send_voice(target_id, voice_bytes, component.length or 0, context_token)
+
+                elif isinstance(component, platform_message.Forward):
+                    for node in component.node_list:
+                        if node.message_chain:
+                            await self.send_message(target_type, target_id, node.message_chain)
+
+            except Exception:
+                await self.logger.error(
+                    f'Failed to send component {type(component).__name__}: {traceback.format_exc()}'
+                )
+
+    async def reply_message(
+        self,
+        message_source: platform_events.MessageEvent,
+        message: platform_message.MessageChain,
+        quote_origin: bool = False,
+    ):
+        """Reply to a received message."""
+        source_msg = message_source.source_platform_object
+        if isinstance(source_msg, WeixinMessage):
+            target_id = source_msg.from_user_id or ''
+            if target_id:
+                await self.send_message('friend', target_id, message)
+
+    async def is_muted(self, group_id: int) -> bool:
+        return False
+
+    @staticmethod
+    async def _get_component_bytes(component: platform_message.MessageComponent) -> typing.Optional[bytes]:
+        """Extract raw bytes from a File or Voice component."""
+        b64_val = getattr(component, 'base64', None)
+        url_val = getattr(component, 'url', None)
+        path_val = getattr(component, 'path', None)
+
+        if b64_val:
+            return base64.b64decode(b64_val)
+        elif url_val and url_val.startswith(('http://', 'https://')):
+            import aiohttp
+
+            async with aiohttp.ClientSession() as session:
+                async with session.get(url_val) as resp:
+                    if resp.status == 200:
+                        return await resp.read()
+        elif path_val:
+            import asyncio
+
+            with open(path_val, 'rb') as f:
+                return await asyncio.to_thread(f.read)
+        return None
+
+    def register_listener(
+        self,
+        event_type: typing.Type[platform_events.Event],
+        callback: typing.Callable[
+            [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter],
+            None,
+        ],
+    ):
+        self.listeners[event_type] = callback
+
+    def unregister_listener(
+        self,
+        event_type: typing.Type[platform_events.Event],
+        callback: typing.Callable[
+            [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter],
+            None,
+        ],
+    ):
+        self.listeners.pop(event_type, None)
+
+    async def run_async(self):
+        """Start the adapter. If no token is configured, trigger QR code login first."""
+        base_url = self.config.get('base_url', DEFAULT_BASE_URL)
+        token = self.config.get('token', '')
+
+        await self.logger.info('OpenClaw WeChat adapter starting...')
+
+        # QR code login flow when no token is provided
+        if not token:
+            await self.logger.info('No token configured, starting QR code login...')
+            try:
+                await self._do_login()
+            except Exception as e:
+                await self.logger.error(f'QR code login failed: {e}')
+                raise
+
+        # Rebuild client with the (possibly updated) config
+        self.client = OpenClawWeixinClient(
+            base_url=self.config.get('base_url', base_url),
+            token=self.config.get('token', token),
+        )
+        self.bot_account_id = self.config.get('account_id', 'openclaw-weixin')
+        self._polling = True
+
+        # Start the long-poll loop
+        self._poll_task = asyncio.create_task(self._poll_loop())
+        await self.logger.info('OpenClaw WeChat adapter running')
+
+        try:
+            await self._poll_task
+        except asyncio.CancelledError:
+            pass
+
+    async def _poll_loop(self):
+        """Long-poll loop: call getUpdates continuously.
+
+        Error handling follows the weixin-bot SDK pattern:
+        - Exponential backoff (1s -> 10s max) on failures
+        - Session expired (errcode -14) triggers automatic re-login
+        """
+        get_updates_buf = ''
+        poll_timeout = float(self.config.get('poll_timeout', 35))
+
+        backoff_delay = 1.0
+        max_backoff = 10.0
+
+        while self._polling:
+            try:
+                resp = await self.client.get_updates(
+                    get_updates_buf=get_updates_buf,
+                    timeout=poll_timeout + 5,
+                )
+
+                if resp.longpolling_timeout_ms and resp.longpolling_timeout_ms > 0:
+                    poll_timeout = resp.longpolling_timeout_ms / 1000.0
+
+                is_api_error = (resp.ret is not None and resp.ret != 0) or (
+                    resp.errcode is not None and resp.errcode != 0
+                )
+                if is_api_error:
+                    is_session_expired = resp.errcode == SESSION_EXPIRED_ERRCODE or resp.ret == SESSION_EXPIRED_ERRCODE
+
+                    if is_session_expired:
+                        await self.logger.error('OpenClaw WeChat session expired, attempting re-login...')
+                        try:
+                            await self._do_login()
+                            # Rebuild client with new credentials
+                            self.client = OpenClawWeixinClient(
+                                base_url=self.config.get('base_url', DEFAULT_BASE_URL),
+                                token=self.config.get('token', ''),
+                            )
+                            self._context_tokens.clear()
+                            get_updates_buf = ''
+                            backoff_delay = 1.0
+                            continue
+                        except Exception:
+                            await self.logger.error(f'Re-login failed: {traceback.format_exc()}')
+                            break
+
+                    await self.logger.error(
+                        f'OpenClaw getUpdates failed: ret={resp.ret} errcode={resp.errcode} errmsg={resp.errmsg}'
+                    )
+                    await asyncio.sleep(backoff_delay)
+                    backoff_delay = min(backoff_delay * 2, max_backoff)
+                    continue
+
+                backoff_delay = 1.0
+
+                if resp.get_updates_buf:
+                    get_updates_buf = resp.get_updates_buf
+
+                for msg in resp.msgs:
+                    try:
+                        await self._handle_inbound_message(msg)
+                    except Exception:
+                        await self.logger.error(f'Error handling message: {traceback.format_exc()}')
+
+            except asyncio.CancelledError:
+                break
+            except Exception:
+                await self.logger.error(f'OpenClaw poll error: {traceback.format_exc()}')
+                await asyncio.sleep(backoff_delay)
+                backoff_delay = min(backoff_delay * 2, max_backoff)
+
+    async def _handle_inbound_message(self, msg: WeixinMessage):
+        """Process a single inbound message from getUpdates."""
+        if msg.context_token and msg.from_user_id:
+            self._context_tokens[msg.from_user_id] = msg.context_token
+
+        # Download CDN media (files, images) before converting to LangBot events
+        await self._download_media_items(msg)
+
+        event = await OpenClawWeixinEventConverter.target2yiri(msg)
+        if event is None:
+            return
+
+        if type(event) in self.listeners:
+            await self.listeners[type(event)](event, self)
+
+    async def _download_media_items(self, msg: WeixinMessage):
+        """Download CDN media for image items in the message."""
+        if not msg.item_list:
+            return
+
+        for item in msg.item_list:
+            try:
+                if item.type == MessageItem.IMAGE and item.image_item:
+                    if (
+                        item.image_item.media
+                        and item.image_item.media.encrypt_query_param
+                        and item.image_item.media.aes_key
+                    ):
+                        img_bytes = await self.client.download_media(item.image_item.media)
+                        item.image_item._downloaded_bytes = img_bytes
+
+                # TODO: enable after full testing
+                # elif item.type == MessageItem.FILE and item.file_item and item.file_item.media:
+                #     if item.file_item.media.encrypt_query_param and item.file_item.media.aes_key:
+                #         file_bytes = await self.client.download_media(item.file_item.media)
+                #         item.file_item._downloaded_bytes = file_bytes
+                #
+                # elif item.type == MessageItem.VOICE and item.voice_item and item.voice_item.media:
+                #     if item.voice_item.media.encrypt_query_param and item.voice_item.media.aes_key:
+                #         voice_bytes = await self.client.download_media(item.voice_item.media)
+                #         item.voice_item._downloaded_bytes = voice_bytes
+                #
+                # elif item.type == MessageItem.VIDEO and item.video_item and item.video_item.media:
+                #     if item.video_item.media.encrypt_query_param and item.video_item.media.aes_key:
+                #         video_bytes = await self.client.download_media(item.video_item.media)
+                #         item.video_item._downloaded_bytes = video_bytes
+
+            except Exception:
+                await self.logger.warning(f'Failed to download CDN media: {traceback.format_exc()}')
+
+    async def kill(self) -> bool:
+        """Stop the adapter."""
+        self._polling = False
+        if self._poll_task and not self._poll_task.done():
+            self._poll_task.cancel()
+            try:
+                await self._poll_task
+            except asyncio.CancelledError:
+                pass
+        await self.client.close()
+        await self.logger.info('OpenClaw WeChat adapter stopped')
+        return True

src/langbot/pkg/platform/sources/openclaw_weixin.yaml

@@ -0,0 +1,57 @@
+apiVersion: v1
+kind: MessagePlatformAdapter
+metadata:
+  name: openclaw-weixin
+  label:
+    en_US: OpenClaw WeChat
+    zh_Hans: OpenClaw 微信
+  description:
+    en_US: OpenClaw WeChat adapter, supports personal WeChat via QR code login
+    zh_Hans: OpenClaw 微信适配器，通过扫码登录支持个人微信
+  icon: wechat.png
+spec:
+  config:
+    - name: base_url
+      label:
+        en_US: API Base URL
+        zh_Hans: API 基础地址
+      description:
+        en_US: The base URL of the OpenClaw WeChat backend API
+        zh_Hans: OpenClaw 微信后端 API 的基础地址
+      type: string
+      required: true
+      default: "https://ilinkai.weixin.qq.com"
+    - name: token
+      label:
+        en_US: Token
+        zh_Hans: 令牌
+      description:
+        en_US: Bearer token obtained after QR code login authorization. Leave empty to trigger QR code login on startup.
+        zh_Hans: 扫码登录授权后获取的 Bearer 令牌。留空则启动时自动触发扫码登录。
+      type: string
+      required: false
+      default: ""
+    - name: account_id
+      label:
+        en_US: Account ID
+        zh_Hans: 账号标识
+      description:
+        en_US: A label for this WeChat account (used for display purposes)
+        zh_Hans: 此微信账号的标识（用于显示）
+      type: string
+      required: false
+      default: "openclaw-weixin"
+    - name: poll_timeout
+      label:
+        en_US: Poll Timeout (seconds)
+        zh_Hans: 轮询超时（秒）
+      description:
+        en_US: Long-poll timeout for getUpdates, the server may hold the request up to this duration
+        zh_Hans: getUpdates 长轮询超时时间，服务端最多持有请求的时长
+      type: integer
+      required: false
+      default: 35
+execution:
+  python:
+    path: ./openclaw_weixin.py
+    attr: OpenClawWeixinAdapter

src/langbot/pkg/platform/sources/telegram.py

@@ -42,6 +42,25 @@ class TelegramMessageConverter(abstract_platform_adapter.AbstractMessageConverte
                         photo_bytes = f.read()
 
                 components.append({'type': 'photo', 'photo': photo_bytes})
+            elif isinstance(component, platform_message.File):
+                file_bytes = None
+
+                if component.base64:
+                    # Strip data URI prefix if present (e.g. "data:application/pdf;base64,...")
+                    b64_data = component.base64
+                    if ';base64,' in b64_data:
+                        b64_data = b64_data.split(';base64,', 1)[1]
+                    file_bytes = base64.b64decode(b64_data)
+                elif component.url:
+                    session = httpclient.get_session()
+                    async with session.get(component.url) as response:
+                        file_bytes = await response.read()
+                elif component.path:
+                    with open(component.path, 'rb') as f:
+                        file_bytes = f.read()
+
+                file_name = getattr(component, 'name', None) or 'file'
+                components.append({'type': 'document', 'document': file_bytes, 'filename': file_name})
             elif isinstance(component, platform_message.Forward):
                 for node in component.node_list:
                     components.extend(await TelegramMessageConverter.yiri2target(node.message_chain, bot))
@@ -104,6 +123,27 @@ class TelegramMessageConverter(abstract_platform_adapter.AbstractMessageConverte
                 )
             )
 
+        if message.document:
+            if message.caption:
+                message_components.extend(parse_message_text(message.caption))
+
+            file = await message.document.get_file()
+            file_name = message.document.file_name or 'document'
+            file_size = message.document.file_size or 0
+            file_format = message.document.mime_type or 'application/octet-stream'
+
+            file_bytes = None
+            async with httpclient.get_session(trust_env=True).get(file.file_path) as response:
+                file_bytes = await response.read()
+
+            message_components.append(
+                platform_message.File(
+                    name=file_name,
+                    size=file_size,
+                    base64=f'data:{file_format};base64,{base64.b64encode(file_bytes).decode("utf-8")}',
+                )
+            )
+
         return platform_message.MessageChain(message_components)
 
 
@@ -179,7 +219,10 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         application = ApplicationBuilder().token(config['token']).build()
         bot = application.bot
         application.add_handler(
-            MessageHandler(filters.TEXT | (filters.COMMAND) | filters.PHOTO | filters.VOICE, telegram_callback)
+            MessageHandler(
+                filters.TEXT | (filters.COMMAND) | filters.PHOTO | filters.VOICE | filters.Document.ALL,
+                telegram_callback,
+            )
         )
         super().__init__(
             config=config,
@@ -218,6 +261,13 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                     continue
                 args['photo'] = telegram.InputFile(photo)
                 await self.bot.send_photo(**args)
+            elif component_type == 'document':
+                doc = component.get('document')
+                if doc is None:
+                    continue
+                filename = component.get('filename', 'file')
+                args['document'] = telegram.InputFile(doc, filename=filename)
+                await self.bot.send_document(**args)
 
     async def reply_message(
         self,

src/langbot/pkg/platform/sources/wecombot.py

@@ -11,6 +11,7 @@ import langbot_plugin.api.entities.builtin.platform.entities as platform_entitie
 from ..logger import EventLogger
 from langbot.libs.wecom_ai_bot_api.wecombotevent import WecomBotEvent
 from langbot.libs.wecom_ai_bot_api.api import WecomBotClient
+from langbot.libs.wecom_ai_bot_api.ws_client import WecomBotWsClient
 
 
 class WecomBotMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
@@ -23,14 +24,18 @@ class WecomBotMessageConverter(abstract_platform_adapter.AbstractMessageConverte
         return content
 
     @staticmethod
-    async def target2yiri(event: WecomBotEvent):
+    async def target2yiri(event: WecomBotEvent, bot_name: str = ''):
         yiri_msg_list = []
         if event.type == 'group':
             yiri_msg_list.append(platform_message.At(target=event.ai_bot_id))
+
         yiri_msg_list.append(platform_message.Source(id=event.message_id, time=datetime.datetime.now()))
 
         if event.content:
-            yiri_msg_list.append(platform_message.Plain(text=event.content))
+            content = event.content
+            if bot_name:
+                content = content.replace(f'@{bot_name}', '').strip()
+            yiri_msg_list.append(platform_message.Plain(text=content))
 
         images = []
         if event.images:
@@ -133,13 +138,15 @@ class WecomBotMessageConverter(abstract_platform_adapter.AbstractMessageConverte
 
 
 class WecomBotEventConverter(abstract_platform_adapter.AbstractEventConverter):
+    def __init__(self, bot_name: str = ''):
+        self.bot_name = bot_name
+
     @staticmethod
     async def yiri2target(event: platform_events.MessageEvent):
         return event.source_platform_object
 
-    @staticmethod
+    async def target2yiri(self, event: WecomBotEvent):
-    async def target2yiri(event: WecomBotEvent):
+        message_chain = await WecomBotMessageConverter.target2yiri(event, bot_name=self.bot_name)
-        message_chain = await WecomBotMessageConverter.target2yiri(event)
         if event.type == 'single':
             return platform_events.FriendMessage(
                 sender=platform_entities.Friend(
@@ -176,34 +183,53 @@ class WecomBotEventConverter(abstract_platform_adapter.AbstractEventConverter):
 
 
 class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
-    bot: WecomBotClient
+    bot: typing.Union[WecomBotClient, WecomBotWsClient]
     bot_account_id: str
     message_converter: WecomBotMessageConverter = WecomBotMessageConverter()
-    event_converter: WecomBotEventConverter = WecomBotEventConverter()
+    event_converter: WecomBotEventConverter
     config: dict
     bot_uuid: str = None
+    _ws_mode: bool = False
+    bot_name: str = ''
+    listeners: dict = {}
 
     def __init__(self, config: dict, logger: EventLogger):
-        required_keys = ['Token', 'EncodingAESKey', 'Corpid', 'BotId']
+        enable_webhook = config.get('enable-webhook', False)
-        missing_keys = [key for key in required_keys if key not in config]
+        bot_name = config.get('robot_name', '')
-        if missing_keys:
-            raise Exception(f'WecomBot 缺少配置项: {missing_keys}')
 
-        bot = WecomBotClient(
+        if not enable_webhook:
-            Token=config['Token'],
+            bot = WecomBotWsClient(
-            EnCodingAESKey=config['EncodingAESKey'],
+                bot_id=config['BotId'],
-            Corpid=config['Corpid'],
+                secret=config['Secret'],
-            logger=logger,
+                logger=logger,
-            unified_mode=True,
+                encoding_aes_key=config.get('EncodingAESKey', ''),
-        )
+            )
-        bot_account_id = config['BotId']
+        else:
+            # Webhook callback mode
+            required_keys = ['Token', 'EncodingAESKey', 'Corpid']
+            missing_keys = [key for key in required_keys if key not in config or not config[key]]
+            if missing_keys:
+                raise Exception(f'WecomBot webhook mode missing config: {missing_keys}')
 
+            bot = WecomBotClient(
+                Token=config['Token'],
+                EnCodingAESKey=config['EncodingAESKey'],
+                Corpid=config['Corpid'],
+                logger=logger,
+                unified_mode=True,
+            )
+
+        bot_account_id = config.get('BotId', '')
+        event_converter = WecomBotEventConverter(bot_name=bot_name)
         super().__init__(
             config=config,
             logger=logger,
             bot=bot,
             bot_account_id=bot_account_id,
+            bot_name=bot_name,
+            event_converter=event_converter,
         )
+        self.listeners = {}
 
     async def reply_message(
         self,
@@ -212,7 +238,17 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         quote_origin: bool = False,
     ):
         content = await self.message_converter.yiri2target(message)
-        await self.bot.set_message(message_source.source_platform_object.message_id, content)
+        _ws_mode = not self.config.get('enable-webhook', False)
+
+        if _ws_mode:
+            event = message_source.source_platform_object
+            req_id = event.get('req_id', '')
+            if req_id:
+                await self.bot.reply_text(req_id, content)
+            else:
+                await self.bot.set_message(event.message_id, content)
+        else:
+            await self.bot.set_message(message_source.source_platform_object.message_id, content)
 
     async def reply_message_chunk(
         self,
@@ -222,31 +258,23 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         quote_origin: bool = False,
         is_final: bool = False,
     ):
-        """将流水线增量输出写入企业微信 stream 会话。
-
-        Args:
-            message_source: 流水线提供的原始消息事件。
-            bot_message: 当前片段对应的模型元信息（未使用）。
-            message: 需要回复的消息链。
-            quote_origin: 是否引用原消息（企业微信暂不支持）。
-            is_final: 标记当前片段是否为最终回复。
-
-        Returns:
-            dict: 包含 `stream` 键，标识写入是否成功。
-
-        Example:
-            在流水线 `reply_message_chunk` 调用中自动触发，无需手动调用。
-        """
-        # 转换为纯文本（智能机器人当前协议仅支持文本流）
         content = await self.message_converter.yiri2target(message)
         msg_id = message_source.source_platform_object.message_id
+        _ws_mode = not self.config.get('enable-webhook', False)
 
-        # 将片段推送到 WecomBotClient 中的队列，返回值用于判断是否走降级逻辑
+        if _ws_mode:
-        success = await self.bot.push_stream_chunk(msg_id, content, is_final=is_final)
+            success = await self.bot.push_stream_chunk(msg_id, content, is_final=is_final)
-        if not success and is_final:
+            if not success and is_final:
-            # 未命中流式队列时使用旧有 set_message 兜底
+                event = message_source.source_platform_object
-            await self.bot.set_message(msg_id, content)
+                req_id = event.get('req_id', '')
-        return {'stream': success}
+                if req_id:
+                    await self.bot.reply_text(req_id, content)
+            return {'stream': success}
+        else:
+            success = await self.bot.push_stream_chunk(msg_id, content, is_final=is_final)
+            if not success and is_final:
+                await self.bot.set_message(msg_id, content)
+            return {'stream': success}
 
     async def is_stream_output_supported(self) -> bool:
         """智能机器人侧默认开启流式能力。
@@ -259,7 +287,21 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         return True
 
     async def send_message(self, target_type, target_id, message):
-        pass
+        _ws_mode = not self.config.get('enable-webhook', False)
+        if _ws_mode:
+            content = await self.message_converter.yiri2target(message)
+            await self.bot.send_message(target_id, content)
+        else:
+            pass
+
+    async def on_message(self, event: WecomBotEvent):
+        try:
+            lb_event = await self.event_converter.target2yiri(event)
+            if lb_event:
+                await self.listeners[type(lb_event)](lb_event, self)
+        except Exception:
+            await self.logger.error(f'Error in wecombot callback: {traceback.format_exc()}')
+            print(traceback.format_exc())
 
     def register_listener(
         self,
@@ -268,18 +310,13 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None
         ],
     ):
-        async def on_message(event: WecomBotEvent):
+        self.listeners[event_type] = callback
-            try:
-                return await callback(await self.event_converter.target2yiri(event), self)
-            except Exception:
-                await self.logger.error(f'Error in wecombot callback: {traceback.format_exc()}')
-                print(traceback.format_exc())
 
         try:
             if event_type == platform_events.FriendMessage:
-                self.bot.on_message('single')(on_message)
+                self.bot.on_message('single')(self.on_message)
             elif event_type == platform_events.GroupMessage:
-                self.bot.on_message('group')(on_message)
+                self.bot.on_message('group')(self.on_message)
         except Exception:
             print(traceback.format_exc())
 
@@ -288,29 +325,28 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         self.bot_uuid = bot_uuid
 
     async def handle_unified_webhook(self, bot_uuid: str, path: str, request):
-        """处理统一 webhook 请求。
+        _ws_mode = not self.config.get('enable-webhook', False)
-
+        if _ws_mode:
-        Args:
+            return None
-            bot_uuid: Bot 的 UUID
-            path: 子路径（如果有的话）
-            request: Quart Request 对象
-
-        Returns:
-            响应数据
-        """
         return await self.bot.handle_unified_webhook(request)
 
     async def run_async(self):
-        # 统一 webhook 模式下，不启动独立的 Quart 应用
+        _ws_mode = not self.config.get('enable-webhook', False)
-        # 保持运行但不启动独立端口
+        if _ws_mode:
+            await self.bot.connect()
+        else:
 
-        async def keep_alive():
+            async def keep_alive():
-            while True:
+                while True:
-                await asyncio.sleep(1)
+                    await asyncio.sleep(1)
 
-        await keep_alive()
+            await keep_alive()
 
     async def kill(self) -> bool:
+        _ws_mode = not self.config.get('enable-webhook', False)
+        if _ws_mode:
+            await self.bot.disconnect()
+            return True
         return False
 
     async def unregister_listener(

src/langbot/pkg/platform/sources/wecombot.yaml

@@ -11,35 +11,71 @@ metadata:
   icon: wecombot.png
 spec:
   config:
+    - name: BotId
+      label:
+        en_US: BotId
+        zh_Hans: 机器人ID (BotId)
+      type: string
+      required: true
+      default: ""
+    - name: robot_name
+      label:
+        en_US: Robot Name
+        zh_Hans: 机器人名称
+      type: string
+      required: true
+      default: ""
+    - name: enable-webhook
+      label:
+        en_US: Enable Webhook Mode
+        zh_Hans: 启用Webhook模式
+      description:
+        en_US: If enabled, the bot will use webhook mode to receive messages. Otherwise, it will use WS long connection mode
+        zh_Hans: 如果启用，机器人将使用 Webhook 模式接收消息。否则，将使用 WS 长连接模式
+      type: boolean
+      required: true
+      default: false
+    - name: Secret
+      label:
+        en_US: Secret
+        zh_Hans: 机器人密钥 (Secret)
+      description:
+        en_US: Required for WebSocket long connection mode
+        zh_Hans: 使用 WS 长连接模式时必填
+      type: string
+      required: false
+      default: ""
     - name: Corpid
       label:
         en_US: Corpid
         zh_Hans: 企业ID
+      description:
+        en_US: Required for Webhook mode
+        zh_Hans: 使用 Webhook 模式时必填
       type: string
-      required: true
+      required: false
       default: ""
     - name: Token
       label:
         en_US: Token
         zh_Hans: 令牌 (Token)
+      description:
+        en_US: Required for Webhook mode
+        zh_Hans: 使用 Webhook 模式时必填
       type: string
-      required: true
+      required: false
       default: ""
     - name: EncodingAESKey
       label:
         en_US: EncodingAESKey
         zh_Hans: 消息加解密密钥 (EncodingAESKey)
-      type: string
+      description:
-      required: true
+        en_US: Required for Webhook mode. Optional for WebSocket mode (used for file decryption)
-      default: ""
+        zh_Hans: 使用 Webhook 模式时必填。WebSocket 模式下可选（用于文件解密）
-    - name: BotId
-      label:
-        en_US: BotId
-        zh_Hans: 机器人ID
       type: string
       required: false
       default: ""
 execution:
   python:
     path: ./wecombot.py
-    attr: WecomBotAdapter
+    attr: WecomBotAdapter

src/langbot/pkg/plugin/handler.py

@@ -314,11 +314,11 @@ class RuntimeConnectionHandler(handler.Handler):
 
         @self.action(PluginToRuntimeAction.GET_LLM_MODELS)
         async def get_llm_models(data: dict[str, Any]) -> handler.ActionResponse:
-            """Get llm models"""
+            """Get llm models, returns list of UUID strings"""
             llm_models = await self.ap.llm_model_service.get_llm_models(include_secret=False)
             return handler.ActionResponse.success(
                 data={
-                    'llm_models': llm_models,
+                    'llm_models': [m['uuid'] for m in llm_models],
                 },
             )
 
@@ -531,6 +531,7 @@ class RuntimeConnectionHandler(handler.Handler):
             filters = data.get('filters')
             search_type = data.get('search_type', 'vector')
             query_text = data.get('query_text', '')
+            vector_weight = data.get('vector_weight')
             try:
                 results = await self.ap.rag_runtime_service.vector_search(
                     collection_id,
@@ -539,6 +540,7 @@ class RuntimeConnectionHandler(handler.Handler):
                     filters,
                     search_type,
                     query_text,
+                    vector_weight=vector_weight,
                 )
                 return handler.ActionResponse.success(data={'results': results})
             except Exception as e:
@@ -555,6 +557,18 @@ class RuntimeConnectionHandler(handler.Handler):
             except Exception as e:
                 return _make_rag_error_response(e, 'VectorStoreError', collection_id=collection_id)
 
+        @self.action(PluginToRuntimeAction.VECTOR_LIST)
+        async def vector_list(data: dict[str, Any]) -> handler.ActionResponse:
+            collection_id = data['collection_id']
+            filters = data.get('filters')
+            limit = data.get('limit', 20)
+            offset = data.get('offset', 0)
+            try:
+                items, total = await self.ap.rag_runtime_service.vector_list(collection_id, filters, limit, offset)
+                return handler.ActionResponse.success(data={'items': items, 'total': total})
+            except Exception as e:
+                return _make_rag_error_response(e, 'VectorStoreError', collection_id=collection_id)
+
         @self.action(PluginToRuntimeAction.GET_KNOWLEDEGE_FILE_STREAM)
         async def get_knowledge_file_stream(data: dict[str, Any]) -> handler.ActionResponse:
             storage_path = data['storage_path']
@@ -565,6 +579,16 @@ class RuntimeConnectionHandler(handler.Handler):
             except Exception as e:
                 return _make_rag_error_response(e, 'FileServiceError', storage_path=storage_path)
 
+        @self.action(PluginToRuntimeAction.LIST_PARSERS)
+        async def list_parsers(data: dict[str, Any]) -> handler.ActionResponse:
+            """Plugin requests host to list available parser plugins."""
+            mime_type = data.get('mime_type')
+            try:
+                parsers = await self.ap.knowledge_service.list_parsers(mime_type)
+                return handler.ActionResponse.success(data={'parsers': parsers})
+            except Exception as e:
+                return _make_rag_error_response(e, 'ParserDiscoveryError', mime_type=mime_type)
+
         @self.action(PluginToRuntimeAction.INVOKE_PARSER)
         async def invoke_parser(data: dict[str, Any]) -> handler.ActionResponse:
             """Plugin requests host to invoke a parser plugin."""
@@ -589,6 +613,139 @@ class RuntimeConnectionHandler(handler.Handler):
             except Exception as e:
                 return _make_rag_error_response(e, 'ParserError')
 
+        # ================= Knowledge Base Query APIs =================
+
+        @self.action(PluginToRuntimeAction.LIST_KNOWLEDGE_BASES)
+        async def list_knowledge_bases(data: dict[str, Any]) -> handler.ActionResponse:
+            """List all knowledge bases available in the LangBot instance (unrestricted)."""
+            knowledge_bases = []
+            for kb_uuid, kb in self.ap.rag_mgr.knowledge_bases.items():
+                knowledge_bases.append(
+                    {
+                        'uuid': kb.get_uuid(),
+                        'name': kb.get_name(),
+                        'description': kb.knowledge_base_entity.description or '',
+                    }
+                )
+            return handler.ActionResponse.success(data={'knowledge_bases': knowledge_bases})
+
+        @self.action(PluginToRuntimeAction.RETRIEVE_KNOWLEDGE)
+        async def retrieve_knowledge(data: dict[str, Any]) -> handler.ActionResponse:
+            """Retrieve documents from any knowledge base (unrestricted)."""
+            kb_id = data['kb_id']
+            query_text = data['query_text']
+            top_k = data.get('top_k', 5)
+            filters = data.get('filters', {})
+
+            kb = await self.ap.rag_mgr.get_knowledge_base_by_uuid(kb_id)
+            if not kb:
+                return handler.ActionResponse.error(
+                    message=f'Knowledge base {kb_id} not found',
+                )
+
+            try:
+                entries = await kb.retrieve(
+                    query_text,
+                    settings={
+                        'top_k': top_k,
+                        'filters': filters,
+                    },
+                )
+                results = [entry.model_dump(mode='json') for entry in entries]
+                return handler.ActionResponse.success(data={'results': results})
+            except Exception as e:
+                return _make_rag_error_response(e, 'RetrievalError', kb_id=kb_id)
+
+        @self.action(PluginToRuntimeAction.LIST_PIPELINE_KNOWLEDGE_BASES)
+        async def list_pipeline_knowledge_bases(data: dict[str, Any]) -> handler.ActionResponse:
+            """List knowledge bases configured for the current query's pipeline."""
+            query_id = data['query_id']
+
+            if query_id not in self.ap.query_pool.cached_queries:
+                return handler.ActionResponse.error(
+                    message=f'Query with query_id {query_id} not found',
+                )
+
+            query = self.ap.query_pool.cached_queries[query_id]
+
+            kb_uuids = []
+            if query.pipeline_config:
+                local_agent_config = query.pipeline_config.get('ai', {}).get('local-agent', {})
+                kb_uuids = local_agent_config.get('knowledge-bases', [])
+                # Backward compatibility
+                if not kb_uuids:
+                    old_kb_uuid = local_agent_config.get('knowledge-base', '')
+                    if old_kb_uuid and old_kb_uuid != '__none__':
+                        kb_uuids = [old_kb_uuid]
+
+            knowledge_bases = []
+            for kb_uuid in kb_uuids:
+                kb = await self.ap.rag_mgr.get_knowledge_base_by_uuid(kb_uuid)
+                if kb:
+                    knowledge_bases.append(
+                        {
+                            'uuid': kb.get_uuid(),
+                            'name': kb.get_name(),
+                            'description': kb.knowledge_base_entity.description or '',
+                        }
+                    )
+
+            return handler.ActionResponse.success(data={'knowledge_bases': knowledge_bases})
+
+        @self.action(PluginToRuntimeAction.RETRIEVE_KNOWLEDGE_BASE)
+        async def retrieve_knowledge_base(data: dict[str, Any]) -> handler.ActionResponse:
+            """Retrieve documents from a knowledge base within the pipeline's scope."""
+            query_id = data['query_id']
+            kb_id = data['kb_id']
+            query_text = data['query_text']
+            top_k = data.get('top_k', 5)
+            filters = data.get('filters', {})
+
+            if query_id not in self.ap.query_pool.cached_queries:
+                return handler.ActionResponse.error(
+                    message=f'Query with query_id {query_id} not found',
+                )
+
+            query = self.ap.query_pool.cached_queries[query_id]
+
+            # Validate kb_id is in pipeline's allowed list
+            allowed_kb_uuids = []
+            if query.pipeline_config:
+                local_agent_config = query.pipeline_config.get('ai', {}).get('local-agent', {})
+                allowed_kb_uuids = local_agent_config.get('knowledge-bases', [])
+                if not allowed_kb_uuids:
+                    old_kb_uuid = local_agent_config.get('knowledge-base', '')
+                    if old_kb_uuid and old_kb_uuid != '__none__':
+                        allowed_kb_uuids = [old_kb_uuid]
+
+            if kb_id not in allowed_kb_uuids:
+                return handler.ActionResponse.error(
+                    message=f'Knowledge base {kb_id} is not configured for this pipeline',
+                )
+
+            kb = await self.ap.rag_mgr.get_knowledge_base_by_uuid(kb_id)
+            if not kb:
+                return handler.ActionResponse.error(
+                    message=f'Knowledge base {kb_id} not found',
+                )
+
+            try:
+                session_name = f'{query.session.launcher_type.value}_{query.session.launcher_id}'
+                entries = await kb.retrieve(
+                    query_text,
+                    settings={
+                        'top_k': top_k,
+                        'filters': filters,
+                        'session_name': session_name,
+                        'bot_uuid': query.bot_uuid or '',
+                        'sender_id': str(query.sender_id),
+                    },
+                )
+                results = [entry.model_dump(mode='json') for entry in entries]
+                return handler.ActionResponse.success(data={'results': results})
+            except Exception as e:
+                return _make_rag_error_response(e, 'RetrievalError', kb_id=kb_id)
+
         @self.action(CommonAction.PING)
         async def ping(data: dict[str, Any]) -> handler.ActionResponse:
             """Ping"""
@@ -895,7 +1052,7 @@ class RuntimeConnectionHandler(handler.Handler):
         result = await self.call_action(
             LangBotToRuntimeAction.RAG_INGEST_DOCUMENT,
             {'plugin_author': plugin_author, 'plugin_name': plugin_name, 'context': context_data},
-            timeout=300,  # Ingestion can be slow
+            timeout=1200,  # Ingestion can be slow for large documents
         )
         return result
 

src/langbot/pkg/provider/modelmgr/requesters/anthropicmsgs.py

@@ -288,10 +288,10 @@ class AnthropicMessages(requester.ProviderAPIRequester):
             think_started = False
             think_ended = False
             finish_reason = False
-            content = ''
             tool_name = ''
             tool_id = ''
             async for chunk in await self.client.messages.create(**args):
+                content = ''
                 tool_call = {'id': None, 'function': {'name': None, 'arguments': None}, 'type': 'function'}
                 if isinstance(
                     chunk, anthropic.types.raw_content_block_start_event.RawContentBlockStartEvent

src/langbot/pkg/provider/runners/localagent.py

@@ -4,6 +4,7 @@ import json
 import copy
 import typing
 from .. import runner
+from ..modelmgr import requester as modelmgr_requester
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
 import langbot_plugin.api.entities.builtin.provider.message as provider_message
 import langbot_plugin.api.entities.builtin.rag.context as rag_context
@@ -26,29 +27,114 @@ Respond in the same language as the user's input.
 
 @runner.runner_class('local-agent')
 class LocalAgentRunner(runner.RequestRunner):
-    """本地Agent请求运行器"""
+    """Local agent request runner"""
 
-    class ToolCallTracker:
+    async def _get_model_candidates(
-        """工具调用追踪器"""
+        self,
+        query: pipeline_query.Query,
+    ) -> list[modelmgr_requester.RuntimeLLMModel]:
+        """Build ordered list of models to try: primary model + fallback models."""
+        candidates = []
 
-        def __init__(self):
+        # Primary model
-            self.active_calls: dict[str, dict] = {}
+        if query.use_llm_model_uuid:
-            self.completed_calls: list[provider_message.ToolCall] = []
+            try:
+                primary = await self.ap.model_mgr.get_model_by_uuid(query.use_llm_model_uuid)
+                candidates.append(primary)
+            except ValueError:
+                self.ap.logger.warning(f'Primary model {query.use_llm_model_uuid} not found')
+
+        # Fallback models
+        fallback_uuids = (query.variables or {}).get('_fallback_model_uuids', [])
+        for fb_uuid in fallback_uuids:
+            try:
+                fb_model = await self.ap.model_mgr.get_model_by_uuid(fb_uuid)
+                candidates.append(fb_model)
+            except ValueError:
+                self.ap.logger.warning(f'Fallback model {fb_uuid} not found, skipping')
+
+        return candidates
+
+    async def _invoke_with_fallback(
+        self,
+        query: pipeline_query.Query,
+        candidates: list[modelmgr_requester.RuntimeLLMModel],
+        messages: list,
+        funcs: list,
+        remove_think: bool,
+    ) -> tuple[provider_message.Message, modelmgr_requester.RuntimeLLMModel]:
+        """Try non-streaming invocation with sequential fallback. Returns (message, model_used)."""
+        last_error = None
+        for model in candidates:
+            try:
+                msg = await model.provider.invoke_llm(
+                    query,
+                    model,
+                    messages,
+                    funcs if model.model_entity.abilities.__contains__('func_call') else [],
+                    extra_args=model.model_entity.extra_args,
+                    remove_think=remove_think,
+                )
+                return msg, model
+            except Exception as e:
+                last_error = e
+                self.ap.logger.warning(f'Model {model.model_entity.name} failed: {e}, trying next fallback...')
+        raise last_error or RuntimeError('No model candidates available')
+
+    async def _invoke_stream_with_fallback(
+        self,
+        query: pipeline_query.Query,
+        candidates: list[modelmgr_requester.RuntimeLLMModel],
+        messages: list,
+        funcs: list,
+        remove_think: bool,
+    ) -> tuple[typing.AsyncGenerator, modelmgr_requester.RuntimeLLMModel]:
+        """Try streaming invocation with sequential fallback. Returns (stream_generator, model_used).
+
+        Fallback is only possible before any chunks have been yielded to the client.
+        Once streaming starts, the model is committed.
+        """
+        last_error = None
+        for model in candidates:
+            try:
+                stream = model.provider.invoke_llm_stream(
+                    query,
+                    model,
+                    messages,
+                    funcs if model.model_entity.abilities.__contains__('func_call') else [],
+                    extra_args=model.model_entity.extra_args,
+                    remove_think=remove_think,
+                )
+                # Attempt to get the first chunk to verify the stream works
+                first_chunk = await stream.__anext__()
+
+                async def _chain_stream(first, rest):
+                    yield first
+                    async for chunk in rest:
+                        yield chunk
+
+                return _chain_stream(first_chunk, stream), model
+            except StopAsyncIteration:
+                # Empty stream — treat as success (model returned nothing)
+                async def _empty_stream():
+                    return
+                    yield  # make it a generator
+
+                return _empty_stream(), model
+            except Exception as e:
+                last_error = e
+                self.ap.logger.warning(f'Model {model.model_entity.name} stream failed: {e}, trying next fallback...')
+        raise last_error or RuntimeError('No model candidates available')
 
     async def run(
         self, query: pipeline_query.Query
     ) -> typing.AsyncGenerator[provider_message.Message | provider_message.MessageChunk, None]:
-        """运行请求"""
+        """Run request"""
         pending_tool_calls = []
 
-        # Get knowledge bases list (new field)
+        # Get knowledge bases list from query variables (set by PreProcessor,
-        kb_uuids = query.pipeline_config['ai']['local-agent'].get('knowledge-bases', [])
+        # may have been modified by plugins during PromptPreProcessing)
-
+        kb_uuids = query.variables.get('_knowledge_base_uuids', [])
-        # Fallback to old field for backward compatibility
-        if not kb_uuids:
-            old_kb_uuid = query.pipeline_config['ai']['local-agent'].get('knowledge-base', '')
-            if old_kb_uuid and old_kb_uuid != '__none__':
-                kb_uuids = [old_kb_uuid]
 
         user_message = copy.deepcopy(query.user_message)
 
@@ -77,6 +163,7 @@ class LocalAgentRunner(runner.RequestRunner):
                 result = await kb.retrieve(
                     user_message_text,
                     settings={
+                        'bot_uuid': query.bot_uuid or '',
                         'sender_id': str(query.sender_id),
                         'session_name': f'{query.session.launcher_type.value}_{query.session.launcher_id}',
                     },
@@ -119,51 +206,51 @@ class LocalAgentRunner(runner.RequestRunner):
 
         remove_think = query.pipeline_config['output'].get('misc', '').get('remove-think')
 
-        use_llm_model = await self.ap.model_mgr.get_model_by_uuid(query.use_llm_model_uuid)
+        # Build ordered candidate list (primary + fallbacks)
+        candidates = await self._get_model_candidates(query)
+        if not candidates:
+            raise RuntimeError('No LLM model configured for local-agent runner')
 
         self.ap.logger.debug(
-            f'localagent req: query={query.query_id} req_messages={req_messages} use_llm_model={query.use_llm_model_uuid}'
+            f'localagent req: query={query.query_id} req_messages={req_messages} '
+            f'candidates={[m.model_entity.name for m in candidates]}'
         )
 
         if not is_stream:
-            # 非流式输出，直接请求
+            # Non-streaming: invoke with fallback
-
+            msg, use_llm_model = await self._invoke_with_fallback(
-            msg = await use_llm_model.provider.invoke_llm(
                 query,
-                use_llm_model,
+                candidates,
                 req_messages,
                 query.use_funcs,
-                extra_args=use_llm_model.model_entity.extra_args,
+                remove_think,
-                remove_think=remove_think,
             )
             yield msg
             final_msg = msg
         else:
-            # 流式输出，需要处理工具调用
+            # Streaming: invoke with fallback
             tool_calls_map: dict[str, provider_message.ToolCall] = {}
             msg_idx = 0
-            accumulated_content = ''  # 从开始累积的所有内容
+            accumulated_content = ''
             last_role = 'assistant'
             msg_sequence = 1
-            async for msg in use_llm_model.provider.invoke_llm_stream(
+
+            stream_src, use_llm_model = await self._invoke_stream_with_fallback(
                 query,
-                use_llm_model,
+                candidates,
                 req_messages,
                 query.use_funcs,
-                extra_args=use_llm_model.model_entity.extra_args,
+                remove_think,
-                remove_think=remove_think,
+            )
-            ):
+            async for msg in stream_src:
                 msg_idx = msg_idx + 1
 
-                # 记录角色
                 if msg.role:
                     last_role = msg.role
 
-                # 累积内容
                 if msg.content:
                     accumulated_content += msg.content
 
-                # 处理工具调用
                 if msg.tool_calls:
                     for tool_call in msg.tool_calls:
                         if tool_call.id not in tool_calls_map:
@@ -175,21 +262,18 @@ class LocalAgentRunner(runner.RequestRunner):
                                 ),
                             )
                         if tool_call.function and tool_call.function.arguments:
-                            # 流式处理中，工具调用参数可能分多个chunk返回，需要追加而不是覆盖
                             tool_calls_map[tool_call.id].function.arguments += tool_call.function.arguments
-                # continue
+
-                # 每8个chunk或最后一个chunk时，输出所有累积的内容
                 if msg_idx % 8 == 0 or msg.is_final:
                     msg_sequence += 1
                     yield provider_message.MessageChunk(
                         role=last_role,
-                        content=accumulated_content,  # 输出所有累积内容
+                        content=accumulated_content,
                         tool_calls=list(tool_calls_map.values()) if (tool_calls_map and msg.is_final) else None,
                         is_final=msg.is_final,
                         msg_sequence=msg_sequence,
                     )
 
-            # 创建最终消息用于后续处理
             final_msg = provider_message.MessageChunk(
                 role=last_role,
                 content=accumulated_content,
@@ -204,7 +288,8 @@ class LocalAgentRunner(runner.RequestRunner):
 
         req_messages.append(final_msg)
 
-        # 持续请求，只要还有待处理的工具调用就继续处理调用
+        # Once a model succeeds, commit to it for the tool call loop
+        # (no fallback mid-conversation — different models may interpret tool results differently)
         while pending_tool_calls:
             for tool_call in pending_tool_calls:
                 try:
@@ -245,7 +330,6 @@ class LocalAgentRunner(runner.RequestRunner):
 
                     req_messages.append(msg)
                 except Exception as e:
-                    # 工具调用出错，添加一个报错信息到 req_messages
                     err_msg = provider_message.Message(role='tool', content=f'err: {e}', tool_call_id=tool_call.id)
 
                     yield err_msg
@@ -253,39 +337,38 @@ class LocalAgentRunner(runner.RequestRunner):
                     req_messages.append(err_msg)
 
             self.ap.logger.debug(
-                f'localagent req: query={query.query_id} req_messages={req_messages} use_llm_model={query.use_llm_model_uuid}'
+                f'localagent req: query={query.query_id} req_messages={req_messages} '
+                f'use_llm_model={use_llm_model.model_entity.name}'
             )
 
             if is_stream:
                 tool_calls_map = {}
                 msg_idx = 0
-                accumulated_content = ''  # 从开始累积的所有内容
+                accumulated_content = ''
                 last_role = 'assistant'
                 msg_sequence = first_end_sequence
 
-                async for msg in use_llm_model.provider.invoke_llm_stream(
+                tool_stream_src = use_llm_model.provider.invoke_llm_stream(
                     query,
                     use_llm_model,
                     req_messages,
-                    query.use_funcs,
+                    query.use_funcs if use_llm_model.model_entity.abilities.__contains__('func_call') else [],
                     extra_args=use_llm_model.model_entity.extra_args,
                     remove_think=remove_think,
-                ):
+                )
+                async for msg in tool_stream_src:
                     msg_idx += 1
 
-                    # 记录角色
                     if msg.role:
                         last_role = msg.role
 
-                    # 第一次请求工具调用时的内容
+                    # Prepend first-round content on first chunk of tool-call round
                     if msg_idx == 1:
                         accumulated_content = first_content if first_content is not None else accumulated_content
 
-                    # 累积内容
                     if msg.content:
                         accumulated_content += msg.content
 
-                    # 处理工具调用
                     if msg.tool_calls:
                         for tool_call in msg.tool_calls:
                             if tool_call.id not in tool_calls_map:
@@ -297,15 +380,13 @@ class LocalAgentRunner(runner.RequestRunner):
                                     ),
                                 )
                             if tool_call.function and tool_call.function.arguments:
-                                # 流式处理中，工具调用参数可能分多个chunk返回，需要追加而不是覆盖
                                 tool_calls_map[tool_call.id].function.arguments += tool_call.function.arguments
 
-                    # 每8个chunk或最后一个chunk时，输出所有累积的内容
                     if msg_idx % 8 == 0 or msg.is_final:
                         msg_sequence += 1
                         yield provider_message.MessageChunk(
                             role=last_role,
-                            content=accumulated_content,  # 输出所有累积内容
+                            content=accumulated_content,
                             tool_calls=list(tool_calls_map.values()) if (tool_calls_map and msg.is_final) else None,
                             is_final=msg.is_final,
                             msg_sequence=msg_sequence,
@@ -318,12 +399,12 @@ class LocalAgentRunner(runner.RequestRunner):
                     msg_sequence=msg_sequence,
                 )
             else:
-                # 处理完所有调用，再次请求
+                # Non-streaming: use committed model directly (no fallback in tool loop)
                 msg = await use_llm_model.provider.invoke_llm(
                     query,
                     use_llm_model,
                     req_messages,
-                    query.use_funcs,
+                    query.use_funcs if use_llm_model.model_entity.abilities.__contains__('func_call') else [],
                     extra_args=use_llm_model.model_entity.extra_args,
                     remove_think=remove_think,
                 )

src/langbot/pkg/rag/service/runtime.py

@@ -41,6 +41,7 @@ class RAGRuntimeService:
         filters: dict[str, Any] | None = None,
         search_type: str = 'vector',
         query_text: str = '',
+        vector_weight: float | None = None,
     ) -> list[dict[str, Any]]:
         """Handle VECTOR_SEARCH action."""
         return await self.ap.vector_db_mgr.search(
@@ -50,6 +51,7 @@ class RAGRuntimeService:
             filter=filters,
             search_type=search_type,
             query_text=query_text,
+            vector_weight=vector_weight,
         )
 
     async def vector_delete(
@@ -75,6 +77,31 @@ class RAGRuntimeService:
             count = await self.ap.vector_db_mgr.delete_by_filter(collection_name=collection_id, filter=filters)
         return count
 
+    async def vector_list(
+        self,
+        collection_id: str,
+        filters: dict[str, Any] | None = None,
+        limit: int = 20,
+        offset: int = 0,
+    ) -> tuple[list[dict[str, Any]], int]:
+        """Handle VECTOR_LIST action.
+
+        Args:
+            collection_id: The collection to list from.
+            filters: Optional metadata filters.
+            limit: Maximum number of items to return.
+            offset: Number of items to skip.
+
+        Returns:
+            Tuple of (items, total).
+        """
+        return await self.ap.vector_db_mgr.list_by_filter(
+            collection_name=collection_id,
+            filter=filters,
+            limit=limit,
+            offset=offset,
+        )
+
     async def get_file_stream(self, storage_path: str) -> bytes:
         """Handle GET_KNOWLEDEGE_FILE_STREAM action.
 

src/langbot/pkg/utils/constants.py

@@ -2,7 +2,7 @@ import langbot
 
 semantic_version = f'v{langbot.__version__}'
 
-required_database_version = 22
+required_database_version = 24
 """Tag the version of the database schema, used to check if the database needs to be migrated"""
 
 debug_mode = False

src/langbot/pkg/vector/filter_utils.py

@@ -49,17 +49,25 @@ def normalize_filter(
 def strip_unsupported_fields(
     triples: list[tuple[str, str, Any]],
     supported_fields: set[str],
+    field_aliases: dict[str, str] | None = None,
 ) -> list[tuple[str, str, Any]]:
     """Return only triples whose field is in *supported_fields*.
 
+    If *field_aliases* is provided, aliased field names are mapped to the
+    canonical backend name before the support check.  For example,
+    ``{'uuid': 'chunk_uuid'}`` allows callers to use ``uuid`` which is
+    transparently rewritten to ``chunk_uuid``.
+
     Dropped fields are logged at WARNING level so the caller knows they were
     silently ignored (useful for Milvus / pgvector which only store a fixed
     schema).
     """
+    aliases = field_aliases or {}
     kept: list[tuple[str, str, Any]] = []
     for field, op, value in triples:
-        if field in supported_fields:
+        resolved = aliases.get(field, field)
-            kept.append((field, op, value))
+        if resolved in supported_fields:
+            kept.append((resolved, op, value))
         else:
             logger.warning(
                 'Filter field %r is not supported by this backend and will be ignored (supported: %s)',

src/langbot/pkg/vector/mgr.py

@@ -97,10 +97,11 @@ class VectorDBManager:
         filter: dict | None = None,
         search_type: str = 'vector',
         query_text: str = '',
+        vector_weight: float | None = None,
     ) -> list[dict]:
         """Proxy: Search vectors.
 
-        Returns a list of dicts with keys: 'id', 'score', 'metadata'.
+        Returns a list of dicts with keys: 'id', 'distance', 'metadata'.
         The underlying VectorDatabase.search returns Chroma-style format:
         { 'ids': [['id1']], 'distances': [[0.1]], 'metadatas': [[{}]] }
         """
@@ -111,6 +112,7 @@ class VectorDBManager:
             search_type=search_type,
             query_text=query_text,
             filter=filter,
+            vector_weight=vector_weight,
         )
 
         if not results or 'ids' not in results or not results['ids']:
@@ -130,7 +132,7 @@ class VectorDBManager:
             parsed_results.append(
                 {
                     'id': id_val,
-                    'score': r_dists[i] if r_dists and i < len(r_dists) else 0.0,
+                    'distance': r_dists[i] if r_dists and i < len(r_dists) else 0.0,
                     'metadata': r_metas[i] if r_metas and i < len(r_metas) else {},
                 }
             )
@@ -157,3 +159,17 @@ class VectorDBManager:
             Number of deleted vectors (best-effort; some backends return 0).
         """
         return await self.vector_db.delete_by_filter(collection_name, filter)
+
+    async def list_by_filter(
+        self,
+        collection_name: str,
+        filter: dict | None = None,
+        limit: int = 20,
+        offset: int = 0,
+    ) -> tuple[list[dict], int]:
+        """Proxy: List vectors by metadata filter with pagination.
+
+        Returns:
+            Tuple of (items, total).
+        """
+        return await self.vector_db.list_by_filter(collection_name, filter, limit, offset)

src/langbot/pkg/vector/vdb.py

@@ -53,6 +53,7 @@ class VectorDatabase(abc.ABC):
         search_type: str = 'vector',
         query_text: str = '',
         filter: dict[str, Any] | None = None,
+        vector_weight: float | None = None,
     ) -> Dict[str, Any]:
         """Search for the most similar vectors in the specified collection.
 
@@ -70,6 +71,8 @@ class VectorDatabase(abc.ABC):
                     {"file_id": "abc"}
                     {"created_at": {"$gte": 1700000000}}
                     {"file_type": {"$in": ["pdf", "docx"]}}
+            vector_weight: Weight for vector search in hybrid mode (0.0–1.0).
+                ``None`` means use equal weights (backward compatible).
         """
         pass
 
@@ -92,6 +95,28 @@ class VectorDatabase(abc.ABC):
         """
         pass
 
+    async def list_by_filter(
+        self,
+        collection: str,
+        filter: dict[str, Any] | None = None,
+        limit: int = 20,
+        offset: int = 0,
+    ) -> tuple[list[dict[str, Any]], int]:
+        """List vectors matching the given metadata filter with pagination.
+
+        Args:
+            collection: Collection name.
+            filter: Optional metadata filter dict in canonical format.
+            limit: Maximum number of items to return.
+            offset: Number of items to skip.
+
+        Returns:
+            Tuple of (items, total) where items is a list of dicts with
+            keys 'id', 'document', 'metadata', and total is the best-effort
+            count of all matching vectors (-1 if unknown).
+        """
+        return [], -1
+
     @abc.abstractmethod
     async def get_or_create_collection(self, collection: str):
         """Get or create collection."""

src/langbot/pkg/vector/vdbs/chroma.py

@@ -52,13 +52,16 @@ class ChromaVectorDatabase(VectorDatabase):
         search_type: str = 'vector',
         query_text: str = '',
         filter: dict[str, Any] | None = None,
+        vector_weight: float | None = None,
     ) -> dict[str, Any]:
         col = await self.get_or_create_collection(collection)
 
         if search_type == SearchType.FULL_TEXT:
             return await self._full_text_search(col, collection, k, query_text, filter)
         elif search_type == SearchType.HYBRID:
-            return await self._hybrid_search(col, collection, query_embedding, k, query_text, filter)
+            return await self._hybrid_search(
+                col, collection, query_embedding, k, query_text, filter, vector_weight=vector_weight
+            )
 
         # Default: vector search
         return await self._vector_search(col, collection, query_embedding, k, filter)
@@ -127,6 +130,7 @@ class ChromaVectorDatabase(VectorDatabase):
         k: int,
         query_text: str,
         filter: dict[str, Any] | None,
+        vector_weight: float | None = None,
     ) -> dict[str, Any]:
         # Fall back to pure vector search when no text is provided
         if not query_text:
@@ -144,7 +148,15 @@ class ChromaVectorDatabase(VectorDatabase):
             return {'ids': [[]], 'metadatas': [[]], 'distances': [[]], 'documents': [[]]}
 
         # RRF fusion
-        fused = self._rrf_fuse([vector_ids, text_ids], k)
+        weights = None
+        if vector_weight is not None:
+            weights = [vector_weight, 1.0 - vector_weight]
+        self.ap.logger.info(
+            f"Chroma hybrid fusion config in '{collection}': "
+            f'vector_weight={vector_weight}, weights={weights or [1.0, 1.0]}, '
+            f'vector_hits={len(vector_ids)}, text_hits={len(text_ids)}'
+        )
+        fused = self._rrf_fuse([vector_ids, text_ids], k, weights=weights)
         if not fused:
             return {'ids': [[]], 'metadatas': [[]], 'distances': [[]], 'documents': [[]]}
 
@@ -197,16 +209,24 @@ class ChromaVectorDatabase(VectorDatabase):
         }
 
     @staticmethod
-    def _rrf_fuse(result_lists: list[list[str]], k: int) -> list[tuple[str, float]]:
+    def _rrf_fuse(result_lists: list[list[str]], k: int, weights: list[float] | None = None) -> list[tuple[str, float]]:
         """Reciprocal Rank Fusion over multiple ranked ID lists.
 
         Returns a list of (doc_id, rrf_score) sorted by descending score,
         truncated to *k* entries.
+
+        Args:
+            result_lists: Ranked ID lists from different search methods.
+            k: Number of results to return.
+            weights: Per-list weights.  ``None`` means equal weight (1.0 each).
         """
+        if weights is None:
+            weights = [1.0] * len(result_lists)
         scores: dict[str, float] = {}
-        for ranked_ids in result_lists:
+        for list_idx, ranked_ids in enumerate(result_lists):
+            w = weights[list_idx]
             for rank, doc_id in enumerate(ranked_ids):
-                scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (_RRF_K + rank + 1)
+                scores[doc_id] = scores.get(doc_id, 0.0) + w / (_RRF_K + rank + 1)
         sorted_results = sorted(scores.items(), key=lambda x: x[1], reverse=True)
         return sorted_results[:k]
 
@@ -221,6 +241,41 @@ class ChromaVectorDatabase(VectorDatabase):
         self.ap.logger.info(f"Deleted embeddings from Chroma collection '{collection}' by filter")
         return 0  # Chroma delete does not return a count
 
+    async def list_by_filter(
+        self,
+        collection: str,
+        filter: dict[str, Any] | None = None,
+        limit: int = 20,
+        offset: int = 0,
+    ) -> tuple[list[dict[str, Any]], int]:
+        col = await self.get_or_create_collection(collection)
+        get_kwargs: dict[str, Any] = dict(
+            include=['metadatas', 'documents'],
+            limit=limit,
+            offset=offset,
+        )
+        if filter:
+            get_kwargs['where'] = filter
+        results = await asyncio.to_thread(col.get, **get_kwargs)
+
+        ids = results.get('ids', [])
+        metadatas = results.get('metadatas', []) or [None] * len(ids)
+        documents = results.get('documents', []) or [None] * len(ids)
+
+        items = []
+        for i, vid in enumerate(ids):
+            items.append(
+                {
+                    'id': vid,
+                    'document': documents[i] if i < len(documents) else None,
+                    'metadata': metadatas[i] if i < len(metadatas) else {},
+                }
+            )
+
+        # Chroma col.count() gives total in collection; filtered count not available
+        total = await asyncio.to_thread(col.count) if not filter else -1
+        return items, total
+
     async def delete_collection(self, collection: str):
         if collection in self._collections:
             del self._collections[collection]

src/langbot/pkg/vector/vdbs/milvus.py

@@ -11,11 +11,14 @@ from langbot.pkg.core import app
 # silently dropped with a warning.
 _MILVUS_SUPPORTED_FIELDS = {'text', 'file_id', 'chunk_uuid'}
 
+# Callers use canonical metadata key 'uuid' but Milvus stores it as 'chunk_uuid'.
+_MILVUS_FIELD_ALIASES = {'uuid': 'chunk_uuid'}
+
 
 def _build_milvus_expr(filter_dict: dict[str, Any]) -> str:
     """Translate canonical filter dict into a Milvus boolean expression string."""
     triples = normalize_filter(filter_dict)
-    triples = strip_unsupported_fields(triples, _MILVUS_SUPPORTED_FIELDS)
+    triples = strip_unsupported_fields(triples, _MILVUS_SUPPORTED_FIELDS, _MILVUS_FIELD_ALIASES)
     if not triples:
         return ''
 
@@ -252,6 +255,7 @@ class MilvusVectorDatabase(VectorDatabase):
         search_type: str = 'vector',
         query_text: str = '',
         filter: dict[str, Any] | None = None,
+        vector_weight: float | None = None,
     ) -> Dict[str, Any]:
         """Search for similar vectors in Milvus collection
 
@@ -340,6 +344,62 @@ class MilvusVectorDatabase(VectorDatabase):
         self.ap.logger.info(f"Deleted embeddings from Milvus collection '{collection}' by filter")
         return 0  # Milvus delete does not return a count
 
+    async def list_by_filter(
+        self,
+        collection: str,
+        filter: dict[str, Any] | None = None,
+        limit: int = 20,
+        offset: int = 0,
+    ) -> tuple[list[dict[str, Any]], int]:
+        collection = self._normalize_collection_name(collection)
+        await self.get_or_create_collection(collection)
+
+        query_kwargs: dict[str, Any] = dict(
+            collection_name=collection,
+            output_fields=['text', 'file_id', 'chunk_uuid'],
+            limit=limit,
+            offset=offset,
+        )
+        if filter:
+            expr = _build_milvus_expr(filter)
+            if expr:
+                query_kwargs['filter'] = expr
+
+        results = await asyncio.to_thread(self.client.query, **query_kwargs)
+
+        items = []
+        for row in results:
+            items.append(
+                {
+                    'id': row.get('id', ''),
+                    'document': row.get('text'),
+                    'metadata': {
+                        'text': row.get('text', ''),
+                        'file_id': row.get('file_id', ''),
+                        'uuid': row.get('chunk_uuid', ''),
+                    },
+                }
+            )
+
+        # Milvus query with count(*)
+        total = -1
+        try:
+            count_kwargs: dict[str, Any] = dict(
+                collection_name=collection,
+                output_fields=['count(*)'],
+            )
+            if filter:
+                expr = _build_milvus_expr(filter)
+                if expr:
+                    count_kwargs['filter'] = expr
+            count_result = await asyncio.to_thread(self.client.query, **count_kwargs)
+            if count_result:
+                total = count_result[0].get('count(*)', -1)
+        except Exception:
+            pass
+
+        return items, total
+
     async def delete_collection(self, collection: str):
         """Delete a Milvus collection
 

src/langbot/pkg/vector/vdbs/pgvector_db.py

@@ -13,6 +13,9 @@ Base = declarative_base()
 # pgvector schema only stores these metadata fields.
 _PG_SUPPORTED_FIELDS = {'text', 'file_id', 'chunk_uuid'}
 
+# Callers use canonical metadata key 'uuid' but pgvector stores it as 'chunk_uuid'.
+_PG_FIELD_ALIASES = {'uuid': 'chunk_uuid'}
+
 # Map schema field names to SQLAlchemy columns (resolved lazily from PgVectorEntry).
 _PG_COLUMN_MAP = {
     'text': 'text',
@@ -37,7 +40,7 @@ class PgVectorEntry(Base):
 def _build_pg_conditions(filter_dict: dict[str, Any]) -> list:
     """Translate canonical filter dict into a list of SQLAlchemy conditions."""
     triples = normalize_filter(filter_dict)
-    triples = strip_unsupported_fields(triples, _PG_SUPPORTED_FIELDS)
+    triples = strip_unsupported_fields(triples, _PG_SUPPORTED_FIELDS, _PG_FIELD_ALIASES)
 
     conditions = []
     for field, op, value in triples:
@@ -189,6 +192,7 @@ class PgVectorDatabase(VectorDatabase):
         search_type: str = 'vector',
         query_text: str = '',
         filter: dict[str, Any] | None = None,
+        vector_weight: float | None = None,
     ) -> Dict[str, Any]:
         """Search for similar vectors using cosine distance
 
@@ -309,6 +313,65 @@ class PgVectorDatabase(VectorDatabase):
                 self.ap.logger.error(f'Error deleting from pgvector by filter: {e}')
                 raise
 
+    async def list_by_filter(
+        self,
+        collection: str,
+        filter: dict[str, Any] | None = None,
+        limit: int = 20,
+        offset: int = 0,
+    ) -> tuple[list[dict[str, Any]], int]:
+        await self.get_or_create_collection(collection)
+
+        async with self.AsyncSessionLocal() as session:
+            try:
+                from sqlalchemy import select, func
+
+                stmt = (
+                    select(
+                        PgVectorEntry.id,
+                        PgVectorEntry.text,
+                        PgVectorEntry.file_id,
+                        PgVectorEntry.chunk_uuid,
+                    )
+                    .filter(PgVectorEntry.collection == collection)
+                    .offset(offset)
+                    .limit(limit)
+                )
+
+                count_stmt = (
+                    select(func.count()).select_from(PgVectorEntry).filter(PgVectorEntry.collection == collection)
+                )
+
+                if filter:
+                    for cond in _build_pg_conditions(filter):
+                        stmt = stmt.filter(cond)
+                        count_stmt = count_stmt.filter(cond)
+
+                result = await session.execute(stmt)
+                rows = result.fetchall()
+
+                count_result = await session.execute(count_stmt)
+                total = count_result.scalar() or 0
+
+                items = []
+                for row in rows:
+                    items.append(
+                        {
+                            'id': row.id,
+                            'document': row.text or '',
+                            'metadata': {
+                                'text': row.text or '',
+                                'file_id': row.file_id or '',
+                                'uuid': row.chunk_uuid or '',
+                            },
+                        }
+                    )
+
+                return items, total
+            except Exception as e:
+                self.ap.logger.error(f'Error listing from pgvector: {e}')
+                raise
+
     async def delete_collection(self, collection: str):
         """Delete all vectors in a collection
 

src/langbot/pkg/vector/vdbs/qdrant.py

@@ -100,6 +100,7 @@ class QdrantVectorDatabase(VectorDatabase):
         search_type: str = 'vector',
         query_text: str = '',
         filter: dict[str, Any] | None = None,
+        vector_weight: float | None = None,
     ) -> dict[str, Any]:
         exists = await self.client.collection_exists(collection)
         if not exists:
@@ -150,6 +151,97 @@ class QdrantVectorDatabase(VectorDatabase):
         self.ap.logger.info(f"Deleted embeddings from Qdrant collection '{collection}' by filter")
         return 0  # Qdrant delete does not return a count
 
+    async def list_by_filter(
+        self,
+        collection: str,
+        filter: dict[str, Any] | None = None,
+        limit: int = 20,
+        offset: int = 0,
+    ) -> tuple[list[dict[str, Any]], int]:
+        exists = await self.client.collection_exists(collection)
+        if not exists:
+            return [], 0
+
+        qdrant_filter = _build_qdrant_filter(filter) if filter else None
+
+        # Qdrant scroll uses cursor-based pagination (offset = point ID),
+        # not numeric skip.  To support numeric offset we scroll through
+        # `offset + limit` items and discard the first `offset`.
+        remaining_to_skip = offset
+        remaining_to_collect = limit
+        cursor: int | str | None = None
+        collected: list[dict[str, Any]] = []
+
+        while remaining_to_skip > 0 or remaining_to_collect > 0:
+            batch_size = remaining_to_skip + remaining_to_collect if remaining_to_skip > 0 else remaining_to_collect
+            scroll_kwargs: dict[str, Any] = dict(
+                collection_name=collection,
+                limit=min(batch_size, 256),
+                with_payload=True if remaining_to_skip == 0 else False,
+                with_vectors=False,
+            )
+            if qdrant_filter:
+                scroll_kwargs['scroll_filter'] = qdrant_filter
+            if cursor is not None:
+                scroll_kwargs['offset'] = cursor
+
+            points, next_cursor = await self.client.scroll(**scroll_kwargs)
+            if not points:
+                break
+
+            for point in points:
+                if remaining_to_skip > 0:
+                    remaining_to_skip -= 1
+                    continue
+                if remaining_to_collect <= 0:
+                    break
+                # Re-fetch payload if we skipped it during the skip phase
+                payload = point.payload or {}
+                collected.append(
+                    {
+                        'id': str(point.id),
+                        'document': payload.get('text') or payload.get('document'),
+                        'metadata': payload,
+                    }
+                )
+                remaining_to_collect -= 1
+
+            if next_cursor is None:
+                break
+            cursor = next_cursor
+
+        # If we skipped without payload, re-fetch the collected items' payloads
+        # (only needed when offset > 0 and items were collected in a skip batch)
+        if offset > 0 and collected:
+            refetch_ids = [item['id'] for item in collected if not item.get('metadata')]
+            if refetch_ids:
+                fetched_points = await self.client.retrieve(
+                    collection_name=collection,
+                    ids=refetch_ids,
+                    with_payload=True,
+                    with_vectors=False,
+                )
+                payload_map = {str(p.id): p.payload or {} for p in fetched_points}
+                for item in collected:
+                    if item['id'] in payload_map:
+                        payload = payload_map[item['id']]
+                        item['metadata'] = payload
+                        item['document'] = payload.get('text') or payload.get('document')
+
+        # Use count() for accurate total (supports filter)
+        total = -1
+        try:
+            count_result = await self.client.count(
+                collection_name=collection,
+                count_filter=qdrant_filter,
+                exact=True,
+            )
+            total = count_result.count
+        except Exception:
+            pass
+
+        return collected, total
+
     async def delete_collection(self, collection: str):
         try:
             await self.client.delete_collection(collection)

src/langbot/pkg/vector/vdbs/seekdb.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import asyncio
+from decimal import Decimal
+import re
 from typing import Any, Dict, List
 
 
@@ -101,8 +103,28 @@ class SeekDBVectorDatabase(VectorDatabase):
             }
         )
 
+    def _normalize_collection_name(self, collection: str) -> str:
+        """SeekDB only accepts [a-zA-Z0-9_], while LangBot uses UUID-like KB IDs."""
+        normalized = re.sub(r'[^A-Za-z0-9_]', '_', collection)
+        if normalized != collection:
+            self.ap.logger.info(f"Normalized SeekDB collection name: '{collection}' -> '{normalized}'")
+        return normalized
+
+    def _json_safe(self, value: Any) -> Any:
+        """Convert SeekDB result values into JSON-serializable Python primitives."""
+        if isinstance(value, Decimal):
+            return float(value)
+        if isinstance(value, dict):
+            return {k: self._json_safe(v) for k, v in value.items()}
+        if isinstance(value, list):
+            return [self._json_safe(v) for v in value]
+        if isinstance(value, tuple):
+            return [self._json_safe(v) for v in value]
+        return value
+
     async def _get_or_create_collection_internal(self, collection: str, vector_size: int = None) -> Any:
         """Internal method to get or create a collection with proper configuration."""
+        collection = self._normalize_collection_name(collection)
         if collection in self._collections:
             return self._collections[collection]
 
@@ -173,6 +195,7 @@ class SeekDBVectorDatabase(VectorDatabase):
         if not embeddings_list:
             return
 
+        collection = self._normalize_collection_name(collection)
         # Ensure collection exists with correct dimension
         vector_size = len(embeddings_list[0])
         coll = await self._get_or_create_collection_internal(collection, vector_size)
@@ -194,6 +217,7 @@ class SeekDBVectorDatabase(VectorDatabase):
         search_type: str = 'vector',
         query_text: str = '',
         filter: Dict[str, Any] | None = None,
+        vector_weight: float | None = None,
     ) -> Dict[str, Any]:
         """Search for the most similar vectors in the specified collection.
 
@@ -210,6 +234,7 @@ class SeekDBVectorDatabase(VectorDatabase):
         Returns:
             Dictionary with 'ids', 'metadatas', 'distances' keys
         """
+        collection = self._normalize_collection_name(collection)
         # Check if collection exists
         exists = await asyncio.to_thread(self.client.has_collection, collection)
         if not exists:
@@ -271,6 +296,17 @@ class SeekDBVectorDatabase(VectorDatabase):
                     query_cfg['where'] = filter
                     knn_cfg['where'] = filter
 
+                # Apply vector_weight via pyseekdb's native boost parameter
+                if vector_weight is not None:
+                    knn_cfg['boost'] = vector_weight
+                    query_cfg['boost'] = 1.0 - vector_weight
+                self.ap.logger.info(
+                    f"SeekDB hybrid fusion config in '{collection}': "
+                    f'vector_weight={vector_weight}, '
+                    f'knn_boost={knn_cfg.get("boost", 1.0)}, '
+                    f'query_boost={query_cfg.get("boost", 1.0)}'
+                )
+
                 results = await asyncio.to_thread(
                     coll.hybrid_search,
                     query=query_cfg,
@@ -279,6 +315,9 @@ class SeekDBVectorDatabase(VectorDatabase):
                     n_results=k,
                     include=['documents', 'metadatas'],
                 )
+                self.ap.logger.info(
+                    f"SeekDB hybrid search in '{collection}' returned {len(results.get('ids', [[]])[0])} results."
+                )
         else:
             # Default: vector search via query()
             query_kwargs = {'n_results': k, 'query_embeddings': query_embedding}
@@ -286,6 +325,7 @@ class SeekDBVectorDatabase(VectorDatabase):
                 query_kwargs['where'] = filter
             results = await asyncio.to_thread(coll.query, **query_kwargs)
 
+        results = self._json_safe(results)
         self.ap.logger.info(
             f"SeekDB {search_type} search in '{collection}' returned {len(results.get('ids', [[]])[0])} results"
         )
@@ -299,6 +339,7 @@ class SeekDBVectorDatabase(VectorDatabase):
             collection: Collection name
             file_id: File ID to delete
         """
+        collection = self._normalize_collection_name(collection)
         # Check if collection exists
         exists = await asyncio.to_thread(self.client.has_collection, collection)
         if not exists:
@@ -325,6 +366,7 @@ class SeekDBVectorDatabase(VectorDatabase):
             collection: Collection name
             filter: Chroma-style ``where`` filter dict
         """
+        collection = self._normalize_collection_name(collection)
         exists = await asyncio.to_thread(self.client.has_collection, collection)
         if not exists:
             self.ap.logger.warning(f"SeekDB collection '{collection}' not found for deletion")
@@ -340,12 +382,59 @@ class SeekDBVectorDatabase(VectorDatabase):
         self.ap.logger.info(f"Deleted embeddings from SeekDB collection '{collection}' by filter")
         return 0  # SeekDB delete does not return a count
 
+    async def list_by_filter(
+        self,
+        collection: str,
+        filter: Dict[str, Any] | None = None,
+        limit: int = 20,
+        offset: int = 0,
+    ) -> tuple[list[Dict[str, Any]], int]:
+        collection = self._normalize_collection_name(collection)
+        exists = await asyncio.to_thread(self.client.has_collection, collection)
+        if not exists:
+            return [], 0
+
+        if collection not in self._collections:
+            coll = await asyncio.to_thread(self.client.get_collection, collection, embedding_function=None)
+            self._collections[collection] = coll
+        else:
+            coll = self._collections[collection]
+
+        get_kwargs: Dict[str, Any] = dict(
+            include=['metadatas', 'documents'],
+            limit=limit,
+            offset=offset,
+        )
+        if filter:
+            get_kwargs['where'] = filter
+
+        results = await asyncio.to_thread(coll.get, **get_kwargs)
+
+        results = self._json_safe(results)
+        ids = results.get('ids', [])
+        metadatas = results.get('metadatas', []) or [None] * len(ids)
+        documents = results.get('documents', []) or [None] * len(ids)
+
+        items = []
+        for i, vid in enumerate(ids):
+            items.append(
+                {
+                    'id': vid,
+                    'document': documents[i] if i < len(documents) else None,
+                    'metadata': metadatas[i] if i < len(metadatas) else {},
+                }
+            )
+
+        total = await asyncio.to_thread(coll.count) if not filter else -1
+        return items, total
+
     async def delete_collection(self, collection: str):
         """Delete the entire collection.
 
         Args:
             collection: Collection name
         """
+        collection = self._normalize_collection_name(collection)
         # Remove from cache
         if collection in self._collections:
             del self._collections[collection]

src/langbot/templates/config.yaml

@@ -2,6 +2,7 @@ admins: []
 api:
     port: 5300
     webhook_prefix: 'http://127.0.0.1:5300'
+    extra_webhook_prefix: ''
 command:
     enable: true
     prefix:
@@ -15,6 +16,7 @@ proxy:
     http: ''
     https: ''
 system:
+    instance_id: ''
     edition: community
     recovery_key: ''
     allow_modify_login_info: true

src/langbot/templates/default-pipeline-config.json

@@ -41,7 +41,10 @@
             "runner": "local-agent"
         },
         "local-agent": {
-            "model": "",
+            "model": {
+                "primary": "",
+                "fallbacks": []
+            },
             "max-round": 10,
             "prompt": [
                 {

src/langbot/templates/metadata/pipeline/ai.yaml

@@ -59,8 +59,11 @@ stages:
         label:
           en_US: Model
           zh_Hans: 模型
-        type: llm-model-selector
+        type: model-fallback-selector
         required: true
+        default:
+          primary: ''
+          fallbacks: []
       - name: max-round
         label:
           en_US: Max Round

src/langbot/templates/metadata/sensitive-words.json

@@ -2,77 +2,5 @@
     "说明": "mask将替换敏感词中的每一个字，若mask_word值不为空，则将敏感词整个替换为mask_word的值",
     "mask": "*",
     "mask_word": "",
-    "words": [
+    "words": []
-        "习近平",
-        "胡锦涛",
-        "江泽民",
-        "温家宝",
-        "李克强",
-        "李长春",
-        "毛泽东",
-        "邓小平",
-        "周恩来",
-        "马克思",
-        "社会主义",
-        "共产党",
-        "共产主义",
-        "大陆官方",
-        "北京政权",
-        "中华帝国",
-        "中国政府",
-        "共狗",
-        "六四事件",
-        "天安门",
-        "六四",
-        "政治局常委",
-        "两会",
-        "共青团",
-        "学潮",
-        "八九",
-        "二十大",
-        "民进党",
-        "台独",
-        "台湾独立",
-        "台湾国",
-        "国民党",
-        "台湾民国",
-        "中华民国",
-        "pornhub",
-        "Pornhub",
-        "[Yy]ou[Pp]orn",
-        "porn",
-        "Porn",
-        "[Xx][Vv]ideos",
-        "[Rr]ed[Tt]ube",
-        "[Xx][Hh]amster",
-        "[Ss]pank[Ww]ire",
-        "[Ss]pank[Bb]ang",
-        "[Tt]ube8",
-        "[Yy]ou[Jj]izz",
-        "[Bb]razzers",
-        "[Nn]aughty[ ]?[Aa]merica",
-        "作爱",
-        "做爱",
-        "性交",
-        "性爱",
-        "自慰",
-        "阴茎",
-        "淫妇",
-        "肛交",
-        "交配",
-        "性关系",
-        "性活动",
-        "色情",
-        "色图",
-        "涩图",
-        "裸体",
-        "小穴",
-        "淫荡",
-        "性爱",
-        "翻墙",
-        "VPN",
-        "科学上网",
-        "挂梯子",
-        "GFW"
-    ]
 }

tests/unit_tests/config/test_webhook_display_prefix.py

@@ -91,14 +91,15 @@ class TestWebhookDisplayPrefix:
 
     def test_default_webhook_prefix(self):
         """Test that the default webhook display prefix is correctly set"""
-        cfg = {'api': {'port': 5300, 'webhook_prefix': 'http://127.0.0.1:5300'}}
+        cfg = {'api': {'port': 5300, 'webhook_prefix': 'http://127.0.0.1:5300', 'extra_webhook_prefix': ''}}
 
         # Should have the default value
         assert cfg['api']['webhook_prefix'] == 'http://127.0.0.1:5300'
+        assert cfg['api']['extra_webhook_prefix'] == ''
 
     def test_webhook_prefix_env_override(self):
         """Test overriding webhook_prefix via environment variable"""
-        cfg = {'api': {'port': 5300, 'webhook_prefix': 'http://127.0.0.1:5300'}}
+        cfg = {'api': {'port': 5300, 'webhook_prefix': 'http://127.0.0.1:5300', 'extra_webhook_prefix': ''}}
 
         # Set environment variable
         os.environ['API__WEBHOOK_PREFIX'] = 'https://example.com:8080'
@@ -112,7 +113,7 @@ class TestWebhookDisplayPrefix:
 
     def test_webhook_prefix_with_custom_domain(self):
         """Test webhook_prefix with custom domain"""
-        cfg = {'api': {'port': 5300, 'webhook_prefix': 'http://127.0.0.1:5300'}}
+        cfg = {'api': {'port': 5300, 'webhook_prefix': 'http://127.0.0.1:5300', 'extra_webhook_prefix': ''}}
 
         # Set to a custom domain
         os.environ['API__WEBHOOK_PREFIX'] = 'https://bot.mycompany.com'
@@ -126,7 +127,7 @@ class TestWebhookDisplayPrefix:
 
     def test_webhook_prefix_with_subdirectory(self):
         """Test webhook_prefix with subdirectory path"""
-        cfg = {'api': {'port': 5300, 'webhook_prefix': 'http://127.0.0.1:5300'}}
+        cfg = {'api': {'port': 5300, 'webhook_prefix': 'http://127.0.0.1:5300', 'extra_webhook_prefix': ''}}
 
         # Set to a URL with subdirectory
         os.environ['API__WEBHOOK_PREFIX'] = 'https://example.com/langbot'
@@ -138,6 +139,37 @@ class TestWebhookDisplayPrefix:
         # Cleanup
         del os.environ['API__WEBHOOK_PREFIX']
 
+    def test_extra_webhook_prefix_default_empty(self):
+        """Test that extra_webhook_prefix defaults to empty string"""
+        cfg = {'api': {'port': 5300, 'webhook_prefix': 'http://127.0.0.1:5300', 'extra_webhook_prefix': ''}}
+
+        bot_uuid = 'test-bot-uuid'
+        webhook_prefix = cfg['api'].get('webhook_prefix', 'http://127.0.0.1:5300')
+        extra_webhook_prefix = cfg['api'].get('extra_webhook_prefix', '')
+        webhook_url = f'/bots/{bot_uuid}'
+
+        assert f'{webhook_prefix}{webhook_url}' == 'http://127.0.0.1:5300/bots/test-bot-uuid'
+        # extra should be empty when not configured
+        assert extra_webhook_prefix == ''
+
+    def test_extra_webhook_prefix_env_override(self):
+        """Test overriding extra_webhook_prefix via environment variable"""
+        cfg = {'api': {'port': 5300, 'webhook_prefix': 'http://127.0.0.1:5300', 'extra_webhook_prefix': ''}}
+
+        os.environ['API__EXTRA_WEBHOOK_PREFIX'] = 'https://extra.example.com'
+
+        result = _apply_env_overrides_to_config(cfg)
+
+        assert result['api']['extra_webhook_prefix'] == 'https://extra.example.com'
+
+        bot_uuid = 'test-bot-uuid'
+        extra_prefix = result['api']['extra_webhook_prefix']
+        webhook_url = f'/bots/{bot_uuid}'
+        assert f'{extra_prefix}{webhook_url}' == 'https://extra.example.com/bots/test-bot-uuid'
+
+        # Cleanup
+        del os.environ['API__EXTRA_WEBHOOK_PREFIX']
+
 
 if __name__ == '__main__':
     pytest.main([__file__, '-v'])

tests/unit_tests/pipeline/conftest.py

@@ -194,7 +194,7 @@ def sample_query(sample_message_chain, sample_message_event, mock_adapter):
         pipeline_config={
             'ai': {
                 'runner': {'runner': 'local-agent'},
-                'local-agent': {'model': 'test-model-uuid', 'prompt': 'test-prompt'},
+                'local-agent': {'model': {'primary': 'test-model-uuid', 'fallbacks': []}, 'prompt': 'test-prompt'},
             },
             'output': {'misc': {'at-sender': False, 'quote-origin': False}},
             'trigger': {'misc': {'combine-quote-message': False}},
@@ -219,7 +219,7 @@ def sample_pipeline_config():
     return {
         'ai': {
             'runner': {'runner': 'local-agent'},
-            'local-agent': {'model': 'test-model-uuid', 'prompt': 'test-prompt'},
+            'local-agent': {'model': {'primary': 'test-model-uuid', 'fallbacks': []}, 'prompt': 'test-prompt'},
         },
         'output': {'misc': {'at-sender': False, 'quote-origin': False}},
         'trigger': {'misc': {'combine-quote-message': False}},

web/package.json

@@ -102,5 +102,10 @@
     "typescript": "^5.8.3",
     "typescript-eslint": "^8.31.1"
   },
-  "packageManager": "pnpm@8.9.2+sha512.b9d35fe91b2a5854dadc43034a3e7b2e675fa4b56e20e8e09ef078fa553c18f8aed44051e7b36e8b8dd435f97eb0c44c4ff3b44fc7c6fa7d21e1fac17bbe661e"
+  "packageManager": "pnpm@8.9.2+sha512.b9d35fe91b2a5854dadc43034a3e7b2e675fa4b56e20e8e09ef078fa553c18f8aed44051e7b36e8b8dd435f97eb0c44c4ff3b44fc7c6fa7d21e1fac17bbe661e",
-}
+  "pnpm": {
+    "overrides": {
+      "minimatch": "3.1.3"
+    }
+  }
+}

web/pnpm-lock.yaml

@@ -4,6 +4,9 @@ settings:
   autoInstallPeers: true
   excludeLinksFromLockfile: false
 
+overrides:
+  minimatch: 3.1.3
+
 dependencies:
   '@dnd-kit/core':
     specifier: ^6.3.1
@@ -345,7 +348,7 @@ packages:
     dependencies:
       '@eslint/object-schema': 2.1.7
       debug: 4.4.3
-      minimatch: 3.1.2
+      minimatch: 3.1.3
     transitivePeerDependencies:
       - supports-color
     dev: true
@@ -375,7 +378,7 @@ packages:
       ignore: 5.3.2
       import-fresh: 3.3.1
       js-yaml: 4.1.1
-      minimatch: 3.1.2
+      minimatch: 3.1.3
       strip-json-comments: 3.1.1
     transitivePeerDependencies:
       - supports-color
@@ -2260,7 +2263,7 @@ packages:
       '@typescript-eslint/types': 8.54.0
       '@typescript-eslint/visitor-keys': 8.54.0
       debug: 4.4.3
-      minimatch: 9.0.5
+      minimatch: 3.1.3
       semver: 7.7.3
       tinyglobby: 0.2.15
       ts-api-utils: 2.4.0(typescript@5.9.3)
@@ -2678,12 +2681,6 @@ packages:
       concat-map: 0.0.1
     dev: true
 
-  /brace-expansion@2.0.2:
-    resolution: {integrity: sha512-Jt0vHyM+jmUBqojB7E1NIYadt0vI0Qxjxd2TErW94wDz+E2LAm5vKMXXwg6ZZBTHPuUlDgQHKXvjGBdfcF1ZDQ==}
-    dependencies:
-      balanced-match: 1.0.2
-    dev: true
-
   /braces@3.0.3:
     resolution: {integrity: sha512-yQbXgO/OSZVD2IsiLlro+7Hf6Q18EJrKSEsdoMzKePKXct3gvD8oLcOQdIzGupr5Fj+EDe8gO/lxc1BzfMpxvA==}
     engines: {node: '>=8'}
@@ -3345,7 +3342,7 @@ packages:
       hasown: 2.0.2
       is-core-module: 2.16.1
       is-glob: 4.0.3
-      minimatch: 3.1.2
+      minimatch: 3.1.3
       object.fromentries: 2.0.8
       object.groupby: 1.0.3
       object.values: 1.2.1
@@ -3376,7 +3373,7 @@ packages:
       hasown: 2.0.2
       jsx-ast-utils: 3.3.5
       language-tags: 1.0.9
-      minimatch: 3.1.2
+      minimatch: 3.1.3
       object.fromentries: 2.0.8
       safe-regex-test: 1.1.0
       string.prototype.includes: 2.0.1
@@ -3428,7 +3425,7 @@ packages:
       estraverse: 5.3.0
       hasown: 2.0.2
       jsx-ast-utils: 3.3.5
-      minimatch: 3.1.2
+      minimatch: 3.1.3
       object.entries: 1.1.9
       object.fromentries: 2.0.8
       object.values: 1.2.1
@@ -3498,7 +3495,7 @@ packages:
       is-glob: 4.0.3
       json-stable-stringify-without-jsonify: 1.0.1
       lodash.merge: 4.6.2
-      minimatch: 3.1.2
+      minimatch: 3.1.3
       natural-compare: 1.4.0
       optionator: 0.9.4
     transitivePeerDependencies:
@@ -5113,19 +5110,12 @@ packages:
     engines: {node: '>=18'}
     dev: true
 
-  /minimatch@3.1.2:
+  /minimatch@3.1.3:
-    resolution: {integrity: sha512-J7p63hRiAjw1NDEww1W7i37+ByIrOWO5XQQAzZ3VOcL0PNybwpfmV/N05zFAzwQ9USyEcX6t3UO+K5aqBQOIHw==}
+    resolution: {integrity: sha512-M2GCs7Vk83NxkUyQV1bkABc4yxgz9kILhHImZiBPAZ9ybuvCb0/H7lEl5XvIg3g+9d4eNotkZA5IWwYl0tibaA==}
     dependencies:
       brace-expansion: 1.1.12
     dev: true
 
-  /minimatch@9.0.5:
-    resolution: {integrity: sha512-G6T0ZX48xgozx7587koeX9Ys2NYy6Gmv//P89sEte9V9whIapMNF4idKxnW2QtCcLiTWlb/wfCabAtAFWhhBow==}
-    engines: {node: '>=16 || 14 >=14.17'}
-    dependencies:
-      brace-expansion: 2.0.2
-    dev: true
-
   /minimist@1.2.8:
     resolution: {integrity: sha512-2yyAR8qBkN3YuheJanUpWC5U3bb5osDywNB8RzDVlDwDHbocAJveqqj1u8+SVD7jkWT4yvsHCpWqqWqAxb0zCA==}
     dev: true

web/src/app/home/bots/components/bot-form/BotForm.tsx

@@ -1,4 +1,4 @@
-import React, { useEffect, useState } from 'react';
+import React, { useEffect, useMemo, useState } from 'react';
 import {
   IChooseAdapterEntity,
   IPipelineEntity,
@@ -113,115 +113,73 @@ export default function BotForm({
   const [dynamicFormConfigList, setDynamicFormConfigList] = useState<
     IDynamicFormItemSchema[]
   >([]);
-  const [filteredDynamicFormConfigList, setFilteredDynamicFormConfigList] =
-    useState<IDynamicFormItemSchema[]>([]);
   const [, setIsLoading] = useState<boolean>(false);
   const [webhookUrl, setWebhookUrl] = useState<string>('');
-  const webhookInputRef = React.useRef<HTMLInputElement>(null);
+  const [extraWebhookUrl, setExtraWebhookUrl] = useState<string>('');
   const [copied, setCopied] = useState<boolean>(false);
+  const [extraCopied, setExtraCopied] = useState<boolean>(false);
 
   // Watch adapter and adapter_config for filtering
   const currentAdapter = form.watch('adapter');
   const currentAdapterConfig = form.watch('adapter_config');
 
-  // Serialize adapter_config to a stable string so it can be used as a
+  // Derive the filtered config list via useMemo instead of useEffect+setState
-  // useEffect dependency without triggering on every render.  form.watch()
+  // to avoid creating new array references that would cause DynamicFormComponent
-  // returns a new object reference each time, which would otherwise cause
+  // to re-subscribe its form.watch, re-emit values, and trigger an infinite loop.
-  // the filtering effect below to loop indefinitely.
+  // Only depend on the specific field we care about (enable-webhook) rather than
-  const adapterConfigJson = JSON.stringify(currentAdapterConfig);
+  // the entire currentAdapterConfig object, which changes on every emission.
+  const enableWebhook = currentAdapterConfig?.['enable-webhook'];
+  const filteredDynamicFormConfigList = useMemo(() => {
+    if (currentAdapter === 'lark' && enableWebhook === false) {
+      // Hide encrypt-key field when webhook is disabled
+      return dynamicFormConfigList.filter(
+        (config) => config.name !== 'encrypt-key',
+      );
+    }
+    // For non-Lark adapters or when webhook is enabled/undefined, show all fields
+    return dynamicFormConfigList;
+  }, [currentAdapter, enableWebhook, dynamicFormConfigList]);
 
   useEffect(() => {
     setBotFormValues();
   }, []);
 
-  // Filter dynamic form config list based on enable-webhook status for Lark adapter
+  // 复制到剪贴板的辅助函数
-  useEffect(() => {
+  const copyToClipboard = (
-    if (currentAdapter === 'lark') {
+    text: string,
-      const enableWebhook = currentAdapterConfig?.['enable-webhook'];
+    setStatus: React.Dispatch<React.SetStateAction<boolean>>,
-      if (enableWebhook === false) {
+  ) => {
-        // Hide encrypt-key field when webhook is disabled
+    if (navigator.clipboard && navigator.clipboard.writeText) {
-        setFilteredDynamicFormConfigList(
+      navigator.clipboard
-          dynamicFormConfigList.filter(
+        .writeText(text)
-            (config) => config.name !== 'encrypt-key',
+        .then(() => {
-          ),
+          setStatus(true);
-        );
+          setTimeout(() => setStatus(false), 2000);
-      } else {
+        })
-        // Show all fields when webhook is enabled or undefined
+        .catch(() => {
-        setFilteredDynamicFormConfigList(dynamicFormConfigList);
+          // 降级：创建临时textarea复制
-      }
+          fallbackCopy(text, setStatus);
+        });
     } else {
-      // For non-Lark adapters, show all fields
+      fallbackCopy(text, setStatus);
-      setFilteredDynamicFormConfigList(dynamicFormConfigList);
     }
-  }, [currentAdapter, adapterConfigJson, dynamicFormConfigList]);
+  };
 
-  // 复制到剪贴板的辅助函数 - 使用页面上的真实input元素
+  const fallbackCopy = (
-  const copyToClipboard = () => {
+    text: string,
-    console.log('[Copy] Attempting to copy from input element');
+    setStatus: React.Dispatch<React.SetStateAction<boolean>>,
-
+  ) => {
-    const inputElement = webhookInputRef.current;
+    const textarea = document.createElement('textarea');
-    if (!inputElement) {
+    textarea.value = text;
-      console.error('[Copy] Input element not found');
+    textarea.style.position = 'fixed';
-      return;
+    textarea.style.opacity = '0';
-    }
+    document.body.appendChild(textarea);
-
+    textarea.select();
-    try {
+    const successful = document.execCommand('copy');
-      // 确保input元素可见且未被禁用
+    document.body.removeChild(textarea);
-      inputElement.disabled = false;
+    if (successful) {
-      inputElement.readOnly = false;
+      setStatus(true);
-
+      setTimeout(() => setStatus(false), 2000);
-      // 聚焦并选中所有文本
-      inputElement.focus();
-      inputElement.select();
-
-      // 尝试使用现代API
-      if (navigator.clipboard && navigator.clipboard.writeText) {
-        console.log(
-          '[Copy] Using Clipboard API with input value:',
-          inputElement.value,
-        );
-        navigator.clipboard
-          .writeText(inputElement.value)
-          .then(() => {
-            console.log('[Copy] Clipboard API success');
-            inputElement.blur(); // 取消选中
-            inputElement.readOnly = true;
-            setCopied(true);
-            setTimeout(() => setCopied(false), 2000);
-          })
-          .catch((err) => {
-            console.error(
-              '[Copy] Clipboard API failed, trying execCommand:',
-              err,
-            );
-            // 降级到execCommand
-            const successful = document.execCommand('copy');
-            console.log('[Copy] execCommand result:', successful);
-            inputElement.blur();
-            inputElement.readOnly = true;
-            if (successful) {
-              setCopied(true);
-              setTimeout(() => setCopied(false), 2000);
-            }
-          });
-      } else {
-        // 直接使用execCommand
-        console.log(
-          '[Copy] Using execCommand with input value:',
-          inputElement.value,
-        );
-        const successful = document.execCommand('copy');
-        console.log('[Copy] execCommand result:', successful);
-        inputElement.blur();
-        inputElement.readOnly = true;
-        if (successful) {
-          setCopied(true);
-          setTimeout(() => setCopied(false), 2000);
-        }
-      }
-    } catch (err) {
-      console.error('[Copy] Copy failed:', err);
-      inputElement.readOnly = true;
     }
   };
 
@@ -246,6 +204,7 @@ export default function BotForm({
             } else {
               setWebhookUrl('');
             }
+            setExtraWebhookUrl(val.extra_webhook_full_url || '');
           })
           .catch((err) => {
             toast.error(
@@ -255,6 +214,7 @@ export default function BotForm({
       } else {
         form.reset();
         setWebhookUrl('');
+        setExtraWebhookUrl('');
       }
     });
   }
@@ -327,14 +287,20 @@ export default function BotForm({
     setAdapterNameToDynamicConfigMap(adapterNameToDynamicConfigMap);
   }
 
-  async function getBotConfig(
+  async function getBotConfig(botId: string): Promise<
-    botId: string,
+    z.infer<typeof formSchema> & {
-  ): Promise<z.infer<typeof formSchema> & { webhook_full_url?: string }> {
+      webhook_full_url?: string;
+      extra_webhook_full_url?: string;
+    }
+  > {
     return new Promise((resolve, reject) => {
       httpClient
         .getBot(botId)
         .then((res) => {
           const bot = res.bot;
+          const runtimeValues = bot.adapter_runtime_values as
+            | Record<string, unknown>
+            | undefined;
           resolve({
             adapter: bot.adapter,
             description: bot.description,
@@ -342,10 +308,12 @@ export default function BotForm({
             adapter_config: bot.adapter_config,
             enable: bot.enable ?? true,
             use_pipeline_uuid: bot.use_pipeline_uuid ?? '',
-            webhook_full_url: bot.adapter_runtime_values
+            webhook_full_url: runtimeValues?.webhook_full_url as
-              ? ((bot.adapter_runtime_values as Record<string, unknown>)
+              | string
-                  .webhook_full_url as string)
+              | undefined,
-              : undefined,
+            extra_webhook_full_url: runtimeValues?.extra_webhook_full_url as
+              | string
+              | undefined,
           });
         })
         .catch((err) => {
@@ -536,13 +504,11 @@ export default function BotForm({
 
                 {/* Webhook 地址显示（统一 Webhook 模式） */}
                 {webhookUrl &&
-                  (currentAdapter !== 'lark' ||
+                  (currentAdapter !== 'lark' || enableWebhook !== false) && (
-                    currentAdapterConfig?.['enable-webhook'] !== false) && (
                     <FormItem>
                       <FormLabel>{t('bots.webhookUrl')}</FormLabel>
                       <div className="flex items-center gap-2">
                         <Input
-                          ref={webhookInputRef}
                           value={webhookUrl}
                           readOnly
                           className="flex-1 bg-gray-50 dark:bg-gray-900"
@@ -555,7 +521,7 @@ export default function BotForm({
                           type="button"
                           variant="outline"
                           size="sm"
-                          onClick={copyToClipboard}
+                          onClick={() => copyToClipboard(webhookUrl, setCopied)}
                         >
                           {copied ? (
                             <Check className="h-4 w-4 text-green-600 mr-2" />
@@ -565,8 +531,37 @@ export default function BotForm({
                           {t('common.copy')}
                         </Button>
                       </div>
+                      {extraWebhookUrl && (
+                        <div className="flex items-center gap-2 mt-2">
+                          <Input
+                            value={extraWebhookUrl}
+                            readOnly
+                            className="flex-1 bg-gray-50 dark:bg-gray-900"
+                            onClick={(e) => {
+                              (e.target as HTMLInputElement).select();
+                            }}
+                          />
+                          <Button
+                            type="button"
+                            variant="outline"
+                            size="sm"
+                            onClick={() =>
+                              copyToClipboard(extraWebhookUrl, setExtraCopied)
+                            }
+                          >
+                            {extraCopied ? (
+                              <Check className="h-4 w-4 text-green-600 mr-2" />
+                            ) : (
+                              <Copy className="h-4 w-4 mr-2" />
+                            )}
+                            {t('common.copy')}
+                          </Button>
+                        </div>
+                      )}
                       <p className="text-sm text-gray-500 mt-1">
-                        {t('bots.webhookUrlHint')}
+                        {extraWebhookUrl
+                          ? t('bots.webhookUrlHintEither')
+                          : t('bots.webhookUrlHint')}
                       </p>
                     </FormItem>
                   )}
@@ -673,7 +668,7 @@ export default function BotForm({
                 </div>
                 <DynamicFormComponent
                   itemConfigList={filteredDynamicFormConfigList}
-                  initialValues={form.watch('adapter_config')}
+                  initialValues={currentAdapterConfig}
                   onSubmit={(values) => {
                     form.setValue('adapter_config', values);
                   }}

web/src/app/home/components/dynamic-form/DynamicFormComponent.tsx

@@ -11,7 +11,7 @@ import {
   FormMessage,
 } from '@/components/ui/form';
 import DynamicFormItemComponent from '@/app/home/components/dynamic-form/DynamicFormItemComponent';
-import { useCallback, useEffect, useRef } from 'react';
+import { useEffect, useRef } from 'react';
 import { extractI18nObject } from '@/i18n/I18nProvider';
 import { useTranslation } from 'react-i18next';
 
@@ -34,6 +34,35 @@ export default function DynamicFormComponent({
   const previousInitialValues = useRef(initialValues);
   const { t } = useTranslation();
 
+  // Normalize a form value according to its field type.
+  // This ensures legacy/malformed data (e.g. a plain string for
+  // model-fallback-selector) is coerced to the expected shape
+  // so that downstream components never crash.
+  const normalizeFieldValue = (
+    item: IDynamicFormItemSchema,
+    value: unknown,
+  ): unknown => {
+    if (item.type === 'model-fallback-selector') {
+      if (value != null && typeof value === 'object' && !Array.isArray(value)) {
+        const obj = value as Record<string, unknown>;
+        return {
+          primary: typeof obj.primary === 'string' ? obj.primary : '',
+          fallbacks: Array.isArray(obj.fallbacks)
+            ? (obj.fallbacks as unknown[]).filter(
+                (v): v is string => typeof v === 'string',
+              )
+            : [],
+        };
+      }
+      // Legacy string format or any other unexpected type
+      return {
+        primary: typeof value === 'string' ? value : '',
+        fallbacks: [],
+      };
+    }
+    return value;
+  };
+
   // 根据 itemConfigList 动态生成 zod schema
   const formSchema = z.object(
     itemConfigList.reduce(
@@ -73,6 +102,12 @@ export default function DynamicFormComponent({
           case 'bot-selector':
             fieldSchema = z.string();
             break;
+          case 'model-fallback-selector':
+            fieldSchema = z.object({
+              primary: z.string(),
+              fallbacks: z.array(z.string()),
+            });
+            break;
           case 'prompt-editor':
             fieldSchema = z.array(
               z.object({
@@ -110,10 +145,10 @@ export default function DynamicFormComponent({
     resolver: zodResolver(formSchema),
     defaultValues: itemConfigList.reduce((acc, item) => {
       // 优先使用 initialValues，如果没有则使用默认值
-      const value = initialValues?.[item.name] ?? item.default;
+      const rawValue = initialValues?.[item.name] ?? item.default;
       return {
         ...acc,
-        [item.name]: value,
+        [item.name]: normalizeFieldValue(item, rawValue),
       };
     }, {} as FormValues),
   });
@@ -138,7 +173,8 @@ export default function DynamicFormComponent({
       // 合并默认值和初始值
       const mergedValues = itemConfigList.reduce(
         (acc, item) => {
-          acc[item.name] = initialValues[item.name] ?? item.default;
+          const rawValue = initialValues[item.name] ?? item.default;
+          acc[item.name] = normalizeFieldValue(item, rawValue) as object;
           return acc;
         },
         {} as Record<string, object>,
@@ -160,39 +196,44 @@ export default function DynamicFormComponent({
   const onSubmitRef = useRef(onSubmit);
   onSubmitRef.current = onSubmit;
 
-  // Track the last emitted values to avoid emitting identical snapshots,
+  // 监听表单值变化
-  // which would cause the parent to call setValue with an equivalent object,
+  useEffect(() => {
-  // triggering a re-render loop.
+    // Emit initial form values immediately so the parent always has a valid snapshot,
-  const lastEmittedRef = useRef<string>('');
+    // even if the user saves without modifying any field.
-
+    // form.watch(callback) only fires on subsequent changes, not on mount.
-  const emitValues = useCallback(() => {
     const formValues = form.getValues();
-    const finalValues = itemConfigList.reduce(
+    const initialFinalValues = itemConfigList.reduce(
       (acc, item) => {
         acc[item.name] = formValues[item.name] ?? item.default;
         return acc;
       },
       {} as Record<string, object>,
     );
-    const serialized = JSON.stringify(finalValues);
+    onSubmitRef.current?.(initialFinalValues);
-    if (serialized !== lastEmittedRef.current) {
-      lastEmittedRef.current = serialized;
-      onSubmitRef.current?.(finalValues);
-    }
-  }, [form, itemConfigList]);
 
-  // 监听表单值变化
+    // Update previousInitialValues to the emitted snapshot so that if the
-  useEffect(() => {
+    // parent writes these values back as new initialValues, the deep
-    // Emit initial form values immediately so the parent always has a valid snapshot,
+    // comparison in the initialValues-sync useEffect won't detect a change
-    // even if the user saves without modifying any field.
+    // and won't trigger an infinite update loop.
-    // form.watch(callback) only fires on subsequent changes, not on mount.
+    previousInitialValues.current = initialFinalValues as Record<
-    emitValues();
+      string,
+      object
+    >;
 
     const subscription = form.watch(() => {
-      emitValues();
+      const formValues = form.getValues();
+      const finalValues = itemConfigList.reduce(
+        (acc, item) => {
+          acc[item.name] = formValues[item.name] ?? item.default;
+          return acc;
+        },
+        {} as Record<string, object>,
+      );
+      onSubmitRef.current?.(finalValues);
+      previousInitialValues.current = finalValues as Record<string, object>;
     });
     return () => subscription.unsubscribe();
-  }, [form, itemConfigList, emitValues]);
+  }, [form, itemConfigList]);
 
   return (
     <Form {...form}>
@@ -231,6 +272,7 @@ export default function DynamicFormComponent({
 
           // All fields are disabled when editing (creation_settings are immutable)
           const isFieldDisabled = !!isEditing;
+
           return (
             <FormField
               key={config.id}

web/src/app/home/components/dynamic-form/DynamicFormItemComponent.tsx

@@ -124,6 +124,28 @@ export default function DynamicFormItemComponent({
     }
   }, [config.type]);
 
+  useEffect(() => {
+    if (config.type === DynamicFormItemType.MODEL_FALLBACK_SELECTOR) {
+      httpClient
+        .getProviderLLMModels()
+        .then((resp) => {
+          let models = resp.models;
+          if (
+            systemInfo.disable_models_service ||
+            userInfo?.account_type !== 'space'
+          ) {
+            models = models.filter(
+              (m) => m.provider?.requester !== 'space-chat-completions',
+            );
+          }
+          setLlmModels(models);
+        })
+        .catch((err) => {
+          toast.error('Failed to get LLM model list: ' + err.msg);
+        });
+    }
+  }, [config.type]);
+
   useEffect(() => {
     if (
       config.type === DynamicFormItemType.KNOWLEDGE_BASE_SELECTOR ||
@@ -171,12 +193,7 @@ export default function DynamicFormItemComponent({
       return <Textarea {...field} className="min-h-[120px]" />;
 
     case DynamicFormItemType.BOOLEAN:
-      return (
+      return <Switch checked={field.value} onCheckedChange={field.onChange} />;
-        <Switch
-          checked={field.value ?? false}
-          onCheckedChange={field.onChange}
-        />
-      );
 
     case DynamicFormItemType.STRING_ARRAY:
       return (
@@ -227,7 +244,7 @@ export default function DynamicFormItemComponent({
 
     case DynamicFormItemType.SELECT:
       return (
-        <Select value={field.value ?? ''} onValueChange={field.onChange}>
+        <Select value={field.value} onValueChange={field.onChange}>
           <SelectTrigger className="bg-[#ffffff] dark:bg-[#2a2a2e]">
             <SelectValue placeholder={t('common.select')} />
           </SelectTrigger>
@@ -318,6 +335,193 @@ export default function DynamicFormItemComponent({
         </Select>
       );
 
+    case DynamicFormItemType.MODEL_FALLBACK_SELECTOR: {
+      // Group models by provider
+      const groupedModelsForFallback = llmModels.reduce(
+        (acc, model) => {
+          const providerName =
+            model.provider?.name || model.provider?.requester || 'Unknown';
+          if (!acc[providerName]) acc[providerName] = [];
+          acc[providerName].push(model);
+          return acc;
+        },
+        {} as Record<string, LLMModel[]>,
+      );
+
+      const rawModelValue = field.value;
+      const modelValue: { primary: string; fallbacks: string[] } =
+        rawModelValue != null &&
+        typeof rawModelValue === 'object' &&
+        !Array.isArray(rawModelValue)
+          ? {
+              primary:
+                typeof (rawModelValue as Record<string, unknown>).primary ===
+                'string'
+                  ? ((rawModelValue as Record<string, unknown>)
+                      .primary as string)
+                  : '',
+              fallbacks: Array.isArray(
+                (rawModelValue as Record<string, unknown>).fallbacks,
+              )
+                ? (
+                    (rawModelValue as Record<string, unknown>)
+                      .fallbacks as unknown[]
+                  ).filter((v): v is string => typeof v === 'string')
+                : [],
+            }
+          : {
+              primary: typeof rawModelValue === 'string' ? rawModelValue : '',
+              fallbacks: [],
+            };
+
+      const renderModelSelect = (
+        value: string,
+        onChange: (val: string) => void,
+        placeholder: string,
+      ) => (
+        <Select value={value} onValueChange={onChange}>
+          <SelectTrigger className="bg-[#ffffff] dark:bg-[#2a2a2e]">
+            <SelectValue placeholder={placeholder} />
+          </SelectTrigger>
+          <SelectContent>
+            {Object.entries(groupedModelsForFallback).map(
+              ([providerName, models]) => (
+                <SelectGroup key={providerName}>
+                  <SelectLabel>{providerName}</SelectLabel>
+                  {models.map((model) => (
+                    <SelectItem key={model.uuid} value={model.uuid}>
+                      <span className="inline-flex items-center gap-1">
+                        {model.name}
+                        {model.abilities?.includes('vision') && (
+                          <Eye className="h-3 w-3 text-muted-foreground" />
+                        )}
+                        {model.abilities?.includes('func_call') && (
+                          <Wrench className="h-3 w-3 text-muted-foreground" />
+                        )}
+                      </span>
+                    </SelectItem>
+                  ))}
+                </SelectGroup>
+              ),
+            )}
+          </SelectContent>
+        </Select>
+      );
+
+      const updateValue = (patch: Partial<typeof modelValue>) => {
+        field.onChange({ ...modelValue, ...patch });
+      };
+
+      const addFallbackModel = () => {
+        updateValue({ fallbacks: [...modelValue.fallbacks, ''] });
+      };
+
+      const updateFallbackModel = (index: number, value: string) => {
+        const updated = [...modelValue.fallbacks];
+        updated[index] = value;
+        updateValue({ fallbacks: updated });
+      };
+
+      const removeFallbackModel = (index: number) => {
+        const updated = [...modelValue.fallbacks];
+        updated.splice(index, 1);
+        updateValue({ fallbacks: updated });
+      };
+
+      const moveFallbackModel = (index: number, direction: 'up' | 'down') => {
+        const updated = [...modelValue.fallbacks];
+        const newIndex = direction === 'up' ? index - 1 : index + 1;
+        if (newIndex < 0 || newIndex >= updated.length) return;
+        [updated[index], updated[newIndex]] = [
+          updated[newIndex],
+          updated[index],
+        ];
+        updateValue({ fallbacks: updated });
+      };
+
+      return (
+        <div className="space-y-3">
+          {/* Primary model selector */}
+          <div>
+            <p className="text-xs text-muted-foreground mb-1">
+              {t('models.fallback.primary')}
+            </p>
+            {renderModelSelect(
+              modelValue.primary,
+              (val) => updateValue({ primary: val }),
+              t('models.selectModel'),
+            )}
+          </div>
+
+          {/* Fallback models */}
+          {modelValue.fallbacks.length > 0 && (
+            <div className="space-y-2">
+              <p className="text-xs text-muted-foreground">
+                {t('models.fallback.fallbackList')}
+              </p>
+              {modelValue.fallbacks.map((fbUuid: string, index: number) => (
+                <div key={index} className="flex items-center gap-2">
+                  <span className="text-xs text-muted-foreground w-4 shrink-0">
+                    {index + 1}.
+                  </span>
+                  <div className="flex-1">
+                    {renderModelSelect(
+                      fbUuid,
+                      (val) => updateFallbackModel(index, val),
+                      t('models.selectModel'),
+                    )}
+                  </div>
+                  <div className="flex gap-1 shrink-0">
+                    <Button
+                      type="button"
+                      variant="ghost"
+                      size="sm"
+                      className="h-8 w-8 p-0"
+                      onClick={() => moveFallbackModel(index, 'up')}
+                      disabled={index === 0}
+                    >
+                      ↑
+                    </Button>
+                    <Button
+                      type="button"
+                      variant="ghost"
+                      size="sm"
+                      className="h-8 w-8 p-0"
+                      onClick={() => moveFallbackModel(index, 'down')}
+                      disabled={index === modelValue.fallbacks.length - 1}
+                    >
+                      ↓
+                    </Button>
+                    <Button
+                      type="button"
+                      variant="ghost"
+                      size="sm"
+                      className="h-8 w-8 p-0 text-destructive"
+                      onClick={() => removeFallbackModel(index)}
+                    >
+                      <X className="h-4 w-4" />
+                    </Button>
+                  </div>
+                </div>
+              ))}
+            </div>
+          )}
+
+          {/* Add fallback button */}
+          <Button
+            type="button"
+            variant="outline"
+            size="sm"
+            className="w-full"
+            onClick={addFallbackModel}
+          >
+            <Plus className="h-4 w-4 mr-1" />
+            {t('models.fallback.addFallback')}
+          </Button>
+        </div>
+      );
+    }
+
     case DynamicFormItemType.KNOWLEDGE_BASE_SELECTOR:
       // Group KBs by Knowledge Engine name
       const kbsByEngine = knowledgeBases.reduce(

web/src/app/home/components/models-dialog/ModelsDialog.tsx

@@ -463,14 +463,16 @@ export default function ModelsDialog({
                     )
                   : t('models.providerCount', { count: otherProviders.length })}
               </span>
-              <Button
+              <div className="flex gap-2">
-                size="sm"
+                <Button
-                variant="outline"
+                  size="sm"
-                onClick={handleCreateProvider}
+                  variant="outline"
-              >
+                  onClick={handleCreateProvider}
-                <Plus className="h-4 w-4 mr-1" />
+                >
-                {t('models.addProvider')}
+                  <Plus className="h-4 w-4 mr-1" />
-              </Button>
+                  {t('models.addProvider')}
+                </Button>
+              </div>
             </div>
 
             {/* Provider List */}

web/src/app/home/knowledge/components/kb-form/KBForm.tsx

@@ -1,4 +1,4 @@
-import { useEffect, useState } from 'react';
+import { useEffect, useMemo, useState } from 'react';
 import Link from 'next/link';
 import { useForm } from 'react-hook-form';
 import { zodResolver } from '@hookform/resolvers/zod';
@@ -242,11 +242,17 @@ export default function KBForm({
   };
 
   // Convert creation schema to dynamic form items (same as ExternalKBForm)
-  const configFormItems = parseCreationSchema(selectedEngine?.creation_schema);
+  // Memoize to avoid regenerating UUIDs on every render, which would cause
+  // DynamicFormComponent's useEffect to re-fire and trigger an infinite loop.
+  const configFormItems = useMemo(
+    () => parseCreationSchema(selectedEngine?.creation_schema),
+    [selectedEngine?.creation_schema],
+  );
 
   // Convert retrieval schema to dynamic form items
-  const retrievalFormItems = parseCreationSchema(
+  const retrievalFormItems = useMemo(
-    selectedEngine?.retrieval_schema,
+    () => parseCreationSchema(selectedEngine?.retrieval_schema),
+    [selectedEngine?.retrieval_schema],
   );
 
   // Show loading state

web/src/app/home/monitoring/hooks/useMonitoringData.ts

@@ -7,6 +7,7 @@ import {
   EmbeddingCall,
 } from '../types/monitoring';
 import { backendClient } from '@/app/infra/http';
+import { parseUTCTimestamp } from '../utils/dateUtils';
 
 /**
  * Custom hook for fetching and managing monitoring data
@@ -120,7 +121,7 @@ export function useMonitoringData(filterState: FilterState) {
             variables?: string;
           }) => ({
             id: msg.id,
-            timestamp: new Date(msg.timestamp),
+            timestamp: parseUTCTimestamp(msg.timestamp),
             botId: msg.bot_id,
             botName: msg.bot_name,
             pipelineId: msg.pipeline_id,
@@ -154,7 +155,7 @@ export function useMonitoringData(filterState: FilterState) {
             message_id?: string;
           }) => ({
             id: call.id,
-            timestamp: new Date(call.timestamp),
+            timestamp: parseUTCTimestamp(call.timestamp),
             modelName: call.model_name,
             tokens: {
               input: call.input_tokens,
@@ -190,7 +191,7 @@ export function useMonitoringData(filterState: FilterState) {
             call_type?: string;
           }) => ({
             id: call.id,
-            timestamp: new Date(call.timestamp),
+            timestamp: parseUTCTimestamp(call.timestamp),
             modelName: call.model_name,
             promptTokens: call.prompt_tokens,
             totalTokens: call.total_tokens,
@@ -227,10 +228,10 @@ export function useMonitoringData(filterState: FilterState) {
             pipelineName: session.pipeline_name,
             messageCount: session.message_count,
             duration:
-              new Date(session.last_activity).getTime() -
+              parseUTCTimestamp(session.last_activity).getTime() -
-              new Date(session.start_time).getTime(),
+              parseUTCTimestamp(session.start_time).getTime(),
-            lastActivity: new Date(session.last_activity),
+            lastActivity: parseUTCTimestamp(session.last_activity),
-            startTime: new Date(session.start_time),
+            startTime: parseUTCTimestamp(session.start_time),
             platform: session.platform,
             userId: session.user_id,
           }),
@@ -250,7 +251,7 @@ export function useMonitoringData(filterState: FilterState) {
             message_id?: string;
           }) => ({
             id: error.id,
-            timestamp: new Date(error.timestamp),
+            timestamp: parseUTCTimestamp(error.timestamp),
             errorType: error.error_type,
             errorMessage: error.error_message,
             botId: error.bot_id,

web/src/app/home/monitoring/utils/dateUtils.ts

@@ -97,3 +97,22 @@ export function isDateInRange(date: Date, range: DateRange | null): boolean {
 export function parseDate(dateStr: string): Date {
   return new Date(dateStr);
 }
+
+/**
+ * Parse a UTC timestamp string from the backend into a Date object.
+ *
+ * The backend stores all monitoring timestamps in UTC but serializes them
+ * as naive ISO strings (e.g. "2026-03-25T14:30:00") without a timezone
+ * designator. JavaScript's `new Date()` would treat such strings as local
+ * time, causing the displayed time to be off by the user's UTC offset.
+ *
+ * This function appends 'Z' when the string has no timezone info, so that
+ * `new Date()` correctly interprets it as UTC.
+ */
+export function parseUTCTimestamp(timestamp: string): Date {
+  // If the string already contains timezone info ('Z', '+', or '-' offset), parse as-is
+  if (/Z|[+-]\d{2}:\d{2}$/.test(timestamp)) {
+    return new Date(timestamp);
+  }
+  return new Date(timestamp + 'Z');
+}

web/src/app/home/pipelines/components/monitoring-tab/PipelineMonitoringTab.tsx

@@ -10,6 +10,7 @@ import { MessageContentRenderer } from '@/app/home/monitoring/components/Message
 import { LoadingSpinner } from '@/components/ui/loading-spinner';
 import { httpClient } from '@/app/infra/http/HttpClient';
 import { MessageDetails } from '@/app/home/monitoring/types/monitoring';
+import { parseUTCTimestamp } from '@/app/home/monitoring/utils/dateUtils';
 
 interface PipelineMonitoringTabProps {
   pipelineId: string;
@@ -120,7 +121,7 @@ export default function PipelineMonitoringTab({
                 message: result.message
                   ? {
                       id: result.message.id,
-                      timestamp: new Date(result.message.timestamp),
+                      timestamp: parseUTCTimestamp(result.message.timestamp),
                       botId: result.message.bot_id,
                       botName: result.message.bot_name,
                       pipelineId: result.message.pipeline_id,
@@ -137,7 +138,7 @@ export default function PipelineMonitoringTab({
                   : undefined,
                 llmCalls: result.llm_calls.map((call: RawLLMCallData) => ({
                   id: call.id,
-                  timestamp: new Date(call.timestamp),
+                  timestamp: parseUTCTimestamp(call.timestamp),
                   modelName: call.model_name,
                   status: call.status,
                   duration: call.duration,
@@ -150,7 +151,7 @@ export default function PipelineMonitoringTab({
                 })),
                 errors: result.errors.map((error: RawErrorData) => ({
                   id: error.id,
-                  timestamp: new Date(error.timestamp),
+                  timestamp: parseUTCTimestamp(error.timestamp),
                   errorType: error.error_type,
                   errorMessage: error.error_message,
                   stackTrace: error.stack_trace,

web/src/app/home/plugins/components/plugin-installed/PluginInstalledComponent.tsx

@@ -38,12 +38,10 @@ const PluginInstalledComponent = forwardRef<PluginInstalledComponentRef>(
   (props, ref) => {
     const { t } = useTranslation();
     const [pluginList, setPluginList] = useState<PluginCardVO[]>([]);
-    const [modalOpen, setModalOpen] = useState<boolean>(false);
+    const [detailModalOpen, setDetailModalOpen] = useState<boolean>(false);
     const [selectedPlugin, setSelectedPlugin] = useState<PluginCardVO | null>(
       null,
     );
-    const [readmeModalOpen, setReadmeModalOpen] = useState<boolean>(false);
-    const [readmePlugin, setReadmePlugin] = useState<PluginCardVO | null>(null);
     const [showOperationModal, setShowOperationModal] = useState(false);
     const [operationType, setOperationType] = useState<PluginOperationType>(
       PluginOperationType.DELETE,
@@ -166,12 +164,7 @@ const PluginInstalledComponent = forwardRef<PluginInstalledComponentRef>(
 
     function handlePluginClick(plugin: PluginCardVO) {
       setSelectedPlugin(plugin);
-      setModalOpen(true);
+      setDetailModalOpen(true);
-    }
-
-    function handleViewReadme(plugin: PluginCardVO) {
-      setReadmePlugin(plugin);
-      setReadmeModalOpen(true);
     }
 
     function handlePluginDelete(plugin: PluginCardVO) {
@@ -355,52 +348,46 @@ const PluginInstalledComponent = forwardRef<PluginInstalledComponentRef>(
           </div>
         ) : (
           <div className={`${styles.pluginListContainer}`}>
-            <Dialog open={modalOpen} onOpenChange={setModalOpen}>
+            <Dialog open={detailModalOpen} onOpenChange={setDetailModalOpen}>
-              <DialogContent className="w-[700px] max-h-[80vh] p-0 flex flex-col">
+              <DialogContent className="sm:max-w-[1100px] max-w-[95vw] max-h-[85vh] p-0 flex flex-col">
-                <DialogHeader className="px-6 pt-6 pb-2">
-                  <DialogTitle>{t('plugins.pluginConfig')}</DialogTitle>
-                </DialogHeader>
-                <div className="flex-1 overflow-y-auto px-6">
-                  {selectedPlugin && (
-                    <PluginForm
-                      pluginAuthor={selectedPlugin.author}
-                      pluginName={selectedPlugin.name}
-                      onFormSubmit={(timeout?: number) => {
-                        setModalOpen(false);
-                        if (timeout) {
-                          setTimeout(() => {
-                            getPluginList();
-                          }, timeout);
-                        } else {
-                          getPluginList();
-                        }
-                      }}
-                      onFormCancel={() => {
-                        setModalOpen(false);
-                      }}
-                    />
-                  )}
-                </div>
-              </DialogContent>
-            </Dialog>
-
-            <Dialog open={readmeModalOpen} onOpenChange={setReadmeModalOpen}>
-              <DialogContent className="sm:max-w-[900px] max-w-[90vw] max-h-[85vh] p-0 flex flex-col">
                 <DialogHeader className="px-6 pt-6 pb-2 border-b">
                   <DialogTitle>
-                    {readmePlugin &&
+                    {selectedPlugin &&
-                      `${readmePlugin.author}/${readmePlugin.name} - ${t(
+                      `${selectedPlugin.author}/${selectedPlugin.name}`}
-                        'plugins.readme',
-                      )}`}
                   </DialogTitle>
                 </DialogHeader>
-                <div className="flex-1 overflow-y-auto">
+                <div className="flex-1 flex flex-row overflow-hidden min-h-0">
-                  {readmePlugin && (
+                  {/* Left side - Readme */}
-                    <PluginReadme
+                  <div className="flex-1 overflow-y-auto border-r min-w-0">
-                      pluginAuthor={readmePlugin.author}
+                    {selectedPlugin && (
-                      pluginName={readmePlugin.name}
+                      <PluginReadme
-                    />
+                        pluginAuthor={selectedPlugin.author}
-                  )}
+                        pluginName={selectedPlugin.name}
+                      />
+                    )}
+                  </div>
+                  {/* Right side - Config */}
+                  <div className="w-[380px] flex-shrink-0 overflow-y-auto px-4">
+                    {selectedPlugin && (
+                      <PluginForm
+                        pluginAuthor={selectedPlugin.author}
+                        pluginName={selectedPlugin.name}
+                        onFormSubmit={(timeout?: number) => {
+                          setDetailModalOpen(false);
+                          if (timeout) {
+                            setTimeout(() => {
+                              getPluginList();
+                            }, timeout);
+                          } else {
+                            getPluginList();
+                          }
+                        }}
+                        onFormCancel={() => {
+                          setDetailModalOpen(false);
+                        }}
+                      />
+                    )}
+                  </div>
                 </div>
               </DialogContent>
             </Dialog>
@@ -413,7 +400,6 @@ const PluginInstalledComponent = forwardRef<PluginInstalledComponentRef>(
                     onCardClick={() => handlePluginClick(vo)}
                     onDeleteClick={() => handlePluginDelete(vo)}
                     onUpgradeClick={() => handlePluginUpdate(vo)}
-                    onViewReadme={() => handleViewReadme(vo)}
                   />
                 </div>
               );

web/src/app/home/plugins/components/plugin-installed/plugin-card/PluginCardComponent.tsx

@@ -2,15 +2,7 @@ import { PluginCardVO } from '@/app/home/plugins/components/plugin-installed/Plu
 import { useState } from 'react';
 import { Badge } from '@/components/ui/badge';
 import { useTranslation } from 'react-i18next';
-import {
+import { BugIcon, ExternalLink, Ellipsis, Trash, ArrowUp } from 'lucide-react';
-  BugIcon,
-  ExternalLink,
-  Ellipsis,
-  Trash,
-  ArrowUp,
-  Settings,
-  FileText,
-} from 'lucide-react';
 import { getCloudServiceClientSync, systemInfo } from '@/app/infra/http';
 import { httpClient } from '@/app/infra/http/HttpClient';
 import { Button } from '@/components/ui/button';
@@ -27,28 +19,20 @@ export default function PluginCardComponent({
   onCardClick,
   onDeleteClick,
   onUpgradeClick,
-  onViewReadme,
 }: {
   cardVO: PluginCardVO;
   onCardClick: () => void;
   onDeleteClick: (cardVO: PluginCardVO) => void;
   onUpgradeClick: (cardVO: PluginCardVO) => void;
-  onViewReadme: (cardVO: PluginCardVO) => void;
 }) {
   const { t } = useTranslation();
   const [dropdownOpen, setDropdownOpen] = useState(false);
-  const [isHovered, setIsHovered] = useState(false);
 
   return (
     <>
       <div
         className="w-[100%] h-[10rem] bg-white rounded-[10px] shadow-[0px_2px_2px_0_rgba(0,0,0,0.2)] p-[1.2rem] cursor-pointer dark:bg-[#1f1f22] relative transition-all duration-200 hover:shadow-[0px_3px_6px_0_rgba(0,0,0,0.12)] hover:scale-[1.005]"
-        onMouseEnter={() => setIsHovered(true)}
+        onClick={() => onCardClick()}
-        onMouseLeave={() => {
-          if (!dropdownOpen) {
-            setIsHovered(false);
-          }
-        }}
       >
         <div className="w-full h-full flex flex-row items-start justify-start gap-[1.2rem]">
           {/* Icon - fixed width */}
@@ -145,7 +129,10 @@ export default function PluginCardComponent({
           </div>
 
           {/* Menu button - fixed width and position */}
-          <div className="flex flex-col items-center justify-between h-full relative z-20 flex-shrink-0">
+          <div
+            className="flex flex-col items-center justify-between h-full relative z-20 flex-shrink-0"
+            onClick={(e) => e.stopPropagation()}
+          >
             <div className="flex items-center justify-center"></div>
 
             <div className="flex items-center justify-center">
@@ -153,9 +140,6 @@ export default function PluginCardComponent({
                 open={dropdownOpen}
                 onOpenChange={(open) => {
                   setDropdownOpen(open);
-                  if (!open) {
-                    setIsHovered(false);
-                  }
                 }}
               >
                 <DropdownMenuTrigger asChild>
@@ -233,45 +217,6 @@ export default function PluginCardComponent({
             </div>
           </div>
         </div>
-
-        {/* Hover overlay with action buttons */}
-        <div
-          className={`absolute inset-0 bg-gray-100/55 dark:bg-black/35 rounded-[10px] flex items-center justify-center gap-3 transition-all duration-200 z-10 ${
-            isHovered ? 'opacity-100' : 'opacity-0 pointer-events-none'
-          }`}
-        >
-          <Button
-            onClick={(e) => {
-              e.stopPropagation();
-              onViewReadme(cardVO);
-            }}
-            className={`bg-blue-600 hover:bg-blue-700 text-white px-4 py-2 rounded-lg shadow-sm flex items-center gap-2 transition-all duration-200 ${
-              isHovered
-                ? 'translate-y-0 opacity-100'
-                : 'translate-y-1 opacity-0'
-            }`}
-            style={{ transitionDelay: isHovered ? '10ms' : '0ms' }}
-          >
-            <FileText className="w-4 h-4" />
-            {t('plugins.readme')}
-          </Button>
-          <Button
-            onClick={(e) => {
-              e.stopPropagation();
-              onCardClick();
-            }}
-            variant="outline"
-            className={`bg-white hover:bg-gray-100 text-gray-900 dark:bg-white dark:hover:bg-gray-100 dark:text-gray-900 px-4 py-2 rounded-lg shadow-sm flex items-center gap-2 transition-all duration-200 ${
-              isHovered
-                ? 'translate-y-0 opacity-100'
-                : 'translate-y-1 opacity-0'
-            }`}
-            style={{ transitionDelay: isHovered ? '20ms' : '0ms' }}
-          >
-            <Settings className="w-4 h-4" />
-            {t('plugins.config')}
-          </Button>
-        </div>
       </div>
     </>
   );

web/src/app/home/plugins/components/plugin-market/PluginMarketComponent.tsx

@@ -1,13 +1,6 @@
 'use client';
 
-import {
+import { useState, useEffect, useCallback, useRef, Suspense } from 'react';
-  useState,
-  useEffect,
-  useCallback,
-  useRef,
-  Suspense,
-  useMemo,
-} from 'react';
 import { Input } from '@/components/ui/input';
 import {
   Select,
@@ -70,9 +63,10 @@ function MarketPageContent({
     RecommendationList[]
   >([]);
 
-  const pageSize = 16; // 每页16个，4行x4列
+  const pageSize = 12; // 每页12个
   const searchTimeoutRef = useRef<NodeJS.Timeout | null>(null);
   const scrollContainerRef = useRef<HTMLDivElement | null>(null);
+  const isComposingRef = useRef(false);
 
   // 排序选项
   const sortOptions: SortOption[] = [
@@ -257,10 +251,14 @@ function MarketPageContent({
         clearTimeout(searchTimeoutRef.current);
       }
 
+      if (isComposingRef.current) {
+        return;
+      }
+
       // 设置新的定时器
       searchTimeoutRef.current = setTimeout(() => {
         handleSearch(value);
-      }, 300);
+      }, 500);
     },
     [handleSearch],
   );
@@ -330,38 +328,7 @@ function MarketPageContent({
     };
   }, []);
 
-  // 计算所有推荐插件的 ID 集合
+  const visiblePlugins = plugins;
-  const recommendedPluginIds = useMemo(() => {
-    const ids = new Set<string>();
-    recommendationLists.forEach((list) => {
-      list.plugins.forEach((plugin) => {
-        ids.add(`${plugin.author} / ${plugin.name}`);
-      });
-    });
-    return ids;
-  }, [recommendationLists]);
-
-  // 过滤掉已在推荐列表中展示的插件
-  // 仅在显示推荐列表的条件下（无搜索、无筛选、第一页或后续页的累积数据中）进行过滤
-  // 注意：如果用户翻页，我们希望一直保持去重，否则推荐过的插件会在第二页出现
-  // 但是推荐列表只在第一页且无筛选时显示。
-  // 如果用户进行了筛选/搜索，推荐列表不显示，此时不需要去重。
-  const visiblePlugins = useMemo(() => {
-    const showRecommendations =
-      !searchQuery && componentFilter === 'all' && selectedTags.length === 0;
-
-    if (!showRecommendations) {
-      return plugins;
-    }
-
-    return plugins.filter((p) => !recommendedPluginIds.has(p.pluginId));
-  }, [
-    plugins,
-    recommendedPluginIds,
-    searchQuery,
-    componentFilter,
-    selectedTags,
-  ]);
 
   // 加载更多
   const loadMore = useCallback(() => {
@@ -436,6 +403,13 @@ function MarketPageContent({
               placeholder={t('market.searchPlaceholder')}
               value={searchQuery}
               onChange={(e) => handleSearchInputChange(e.target.value)}
+              onCompositionStart={() => {
+                isComposingRef.current = true;
+              }}
+              onCompositionEnd={(e) => {
+                isComposingRef.current = false;
+                handleSearchInputChange((e.target as HTMLInputElement).value);
+              }}
               onKeyPress={(e) => {
                 if (e.key === 'Enter') {
                   // Immediately search, clear debounce timer

web/src/app/home/plugins/components/plugin-market/RecommendationLists.tsx

@@ -47,10 +47,12 @@ function RecommendationListRow({
   list,
   tagNames,
   onInstall,
+  isLast,
 }: {
   list: RecommendationList;
   tagNames: Record<string, string>;
   onInstall: (author: string, pluginName: string) => void;
+  isLast: boolean;
 }) {
   const { t } = useTranslation();
   const [page, setPage] = useState(0);
@@ -143,7 +145,9 @@ function RecommendationListRow({
           />
         ))}
       </div>
-      {totalPages > 1 && <div className="border-b border-border mt-6" />}
+      {totalPages > 1 && !isLast && (
+        <div className="border-b border-border mt-6" />
+      )}
     </div>
   );
 }
@@ -161,12 +165,13 @@ export function RecommendationLists({
 
   return (
     <div className="mt-6">
-      {lists.map((list) => (
+      {lists.map((list, index) => (
         <RecommendationListRow
           key={list.uuid}
           list={list}
           tagNames={tagNames}
           onInstall={onInstall}
+          isLast={index === lists.length - 1}
         />
       ))}
       <div className="border-b border-border mb-6" />

web/src/app/home/plugins/components/plugin-market/plugin-market-card/PluginMarketCardComponent.tsx

@@ -17,7 +17,7 @@ import {
   FileText,
   Info,
 } from 'lucide-react';
-import { useState } from 'react';
+import { useState, useRef, useEffect } from 'react';
 import { Button } from '@/components/ui/button';
 
 export default function PluginMarketCardComponent({
@@ -31,6 +31,43 @@ export default function PluginMarketCardComponent({
 }) {
   const { t } = useTranslation();
   const [isHovered, setIsHovered] = useState(false);
+  const bottomRef = useRef<HTMLDivElement>(null);
+  const [visibleTags, setVisibleTags] = useState(2);
+
+  // Measure how many tags fit in the bottom row
+  useEffect(() => {
+    const tags = cardVO.tags;
+    if (!bottomRef.current || !tags || tags.length === 0) return;
+
+    const measure = () => {
+      const container = bottomRef.current;
+      if (!container) return;
+      const width = container.offsetWidth;
+      const availableForTags = width - 140 - 80;
+      if (availableForTags <= 0) {
+        setVisibleTags(0);
+        return;
+      }
+      const tagWidth = 80;
+      const plusBadgeWidth = 40;
+      const maxTags = Math.max(
+        0,
+        Math.floor((availableForTags - plusBadgeWidth) / tagWidth),
+      );
+      if (maxTags >= tags.length) {
+        setVisibleTags(tags.length);
+      } else {
+        setVisibleTags(Math.max(1, maxTags));
+      }
+    };
+
+    measure();
+    const observer = new ResizeObserver(measure);
+    observer.observe(bottomRef.current);
+    return () => observer.disconnect();
+  }, [cardVO.tags]);
+
+  const remainingTags = cardVO.tags ? cardVO.tags.length - visibleTags : 0;
 
   function handleInstallClick(e: React.MouseEvent) {
     e.stopPropagation();
@@ -135,10 +172,13 @@ export default function PluginMarketCardComponent({
         </div>
 
         {/* 下部分：下载量、标签和组件列表 */}
-        <div className="w-full flex flex-row items-center justify-between gap-2 px-0 sm:px-[0.4rem] flex-shrink-0">
+        <div
-          <div className="flex flex-row items-center justify-start gap-2 flex-wrap">
+          ref={bottomRef}
+          className="w-full flex flex-row items-center justify-between gap-2 px-0 sm:px-[0.4rem] flex-shrink-0 overflow-hidden"
+        >
+          <div className="flex flex-row items-center justify-start gap-2 min-w-0 overflow-hidden">
             {/* 下载数量 */}
-            <div className="flex flex-row items-center gap-[0.3rem] sm:gap-[0.4rem]">
+            <div className="flex flex-row items-center gap-[0.3rem] sm:gap-[0.4rem] flex-shrink-0">
               <svg
                 className="w-4 h-4 sm:w-[1.2rem] sm:h-[1.2rem] text-[#2563eb] dark:text-[#5b8def] flex-shrink-0"
                 xmlns="http://www.w3.org/2000/svg"
@@ -156,14 +196,14 @@ export default function PluginMarketCardComponent({
               </div>
             </div>
 
-            {/* Tags */}
+            {/* Tags - adaptive */}
-            {cardVO.tags && cardVO.tags.length > 0 && (
+            {cardVO.tags && cardVO.tags.length > 0 && visibleTags > 0 && (
-              <div className="flex flex-wrap gap-1.5">
+              <div className="flex flex-row items-center gap-1.5 overflow-hidden flex-shrink min-w-0">
-                {cardVO.tags.slice(0, 2).map((tag) => (
+                {cardVO.tags.slice(0, visibleTags).map((tag) => (
                   <Badge
                     key={tag}
                     variant="secondary"
-                    className="text-[0.65rem] sm:text-[0.7rem] px-2 py-0.5 h-5 flex items-center gap-1 flex-shrink-0"
+                    className="text-[0.65rem] sm:text-[0.7rem] px-2 py-0.5 h-5 flex items-center gap-1 flex-shrink-0 whitespace-nowrap"
                   >
                     <svg
                       className="w-2.5 h-2.5 flex-shrink-0"
@@ -178,15 +218,17 @@ export default function PluginMarketCardComponent({
                       <path d="M20.59 13.41l-7.17 7.17a2 2 0 0 1-2.83 0L2 12V2h10l8.59 8.59a2 2 0 0 1 0 2.82z" />
                       <line x1="7" y1="7" x2="7.01" y2="7" />
                     </svg>
-                    <span className="truncate">{tagNames[tag] || tag}</span>
+                    <span className="truncate max-w-[5rem]">
+                      {tagNames[tag] || tag}
+                    </span>
                   </Badge>
                 ))}
-                {cardVO.tags.length > 2 && (
+                {remainingTags > 0 && (
                   <Badge
                     variant="outline"
-                    className="text-[0.65rem] sm:text-[0.7rem] px-2 py-0.5 h-5 flex items-center flex-shrink-0"
+                    className="text-[0.65rem] sm:text-[0.7rem] px-1.5 py-0.5 h-5 flex items-center flex-shrink-0 whitespace-nowrap"
                   >
-                    +{cardVO.tags.length - 2}
+                    +{remainingTags}
                   </Badge>
                 )}
               </div>

web/src/app/infra/entities/form/dynamic.ts

@@ -35,6 +35,7 @@ export enum DynamicFormItemType {
   SELECT = 'select',
   LLM_MODEL_SELECTOR = 'llm-model-selector',
   EMBEDDING_MODEL_SELECTOR = 'embedding-model-selector',
+  MODEL_FALLBACK_SELECTOR = 'model-fallback-selector',
   PROMPT_EDITOR = 'prompt-editor',
   UNKNOWN = 'unknown',
   KNOWLEDGE_BASE_SELECTOR = 'knowledge-base-selector',

web/src/app/login/page.tsx

@@ -23,7 +23,7 @@ import {
 import { useEffect, useState } from 'react';
 import { httpClient, initializeUserInfo } from '@/app/infra/http';
 import { useRouter } from 'next/navigation';
-import { Mail, Lock, Loader2 } from 'lucide-react';
+import { Mail, Lock, Loader2, AlertCircle, RefreshCw } from 'lucide-react';
 import langbotIcon from '@/app/assets/langbot-logo.webp';
 import { toast } from 'sonner';
 import { useTranslation } from 'react-i18next';
@@ -46,6 +46,8 @@ export default function Login() {
   const [accountType, setAccountType] = useState<AccountType | null>(null);
   const [hasPassword, setHasPassword] = useState(false);
   const [loading, setLoading] = useState(true);
+  const [loadError, setLoadError] = useState<string | null>(null);
+  const [retrying, setRetrying] = useState(false);
 
   const form = useForm<z.infer<ReturnType<typeof formSchema>>>({
     resolver: zodResolver(formSchema(t)),
@@ -61,6 +63,7 @@ export default function Login() {
 
   async function checkAccountInfo() {
     try {
+      setLoadError(null);
       const res = await httpClient.getAccountInfo();
       if (!res.initialized) {
         router.push('/register');
@@ -72,11 +75,22 @@ export default function Login() {
 
       // Also check if already logged in
       checkIfAlreadyLoggedIn();
-    } catch {
+    } catch (err) {
+      const errorMessage =
+        err instanceof Error ? err.message : t('common.loginLoadError');
+      setLoadError(errorMessage);
       setLoading(false);
     }
   }
 
+  async function handleRetry() {
+    setRetrying(true);
+    setLoading(true);
+    setLoadError(null);
+    await checkAccountInfo();
+    setRetrying(false);
+  }
+
   function checkIfAlreadyLoggedIn() {
     httpClient
       .checkUserToken()
@@ -129,6 +143,54 @@ export default function Login() {
     );
   }
 
+  // Show error state when account info failed to load
+  if (loadError) {
+    return (
+      <div className="min-h-screen flex items-center justify-center bg-gray-50 dark:bg-neutral-900">
+        <Card className="w-[375px] shadow-lg dark:shadow-white/10">
+          <CardHeader>
+            <div className="flex justify-between items-center mb-6">
+              <ThemeToggle />
+              <LanguageSelector />
+            </div>
+            <img
+              src={langbotIcon.src}
+              alt="LangBot"
+              className="w-16 h-16 mb-4 mx-auto"
+            />
+            <CardTitle className="text-2xl text-center">
+              {t('common.welcome')}
+            </CardTitle>
+          </CardHeader>
+          <CardContent className="space-y-4">
+            <div className="flex flex-col items-center gap-3 py-4">
+              <AlertCircle className="h-10 w-10 text-destructive" />
+              <p className="text-sm text-center text-muted-foreground">
+                {t('common.loginLoadErrorDesc')}
+              </p>
+              <code className="text-xs bg-muted px-3 py-2 rounded max-w-full overflow-x-auto block text-center text-muted-foreground">
+                {loadError}
+              </code>
+              <Button
+                onClick={handleRetry}
+                disabled={retrying}
+                variant="outline"
+                className="mt-2 cursor-pointer"
+              >
+                {retrying ? (
+                  <Loader2 className="mr-2 h-4 w-4 animate-spin" />
+                ) : (
+                  <RefreshCw className="mr-2 h-4 w-4" />
+                )}
+                {t('common.retry')}
+              </Button>
+            </div>
+          </CardContent>
+        </Card>
+      </div>
+    );
+  }
+
   // Determine what to show based on account type
   const showLocalLogin =
     accountType === 'local' || (accountType === 'space' && hasPassword);

web/src/i18n/locales/en-US.ts

@@ -11,6 +11,10 @@ const enUS = {
     continueToLogin: 'Login to continue',
     loginSuccess: 'Login successful',
     loginFailed: 'Login failed, please check your email and password',
+    loginLoadError: 'Unable to connect to server',
+    loginLoadErrorDesc:
+      'Unable to connect to the LangBot backend. Please make sure the service is running and try again.',
+    retry: 'Retry',
     enterEmail: 'Enter email address',
     enterPassword: 'Enter password',
     invalidEmail: 'Please enter a valid email address',
@@ -236,6 +240,11 @@ const enUS = {
     modelsCount: '{{count}} model(s)',
     expandModels: 'Expand',
     collapseModels: 'Collapse',
+    fallback: {
+      primary: 'Primary Model',
+      fallbackList: 'Fallback Models',
+      addFallback: 'Add Fallback Model',
+    },
   },
   bots: {
     title: 'Bots',
@@ -279,6 +288,8 @@ const enUS = {
     webhookUrlCopied: 'Webhook URL copied',
     webhookUrlHint:
       'Click the input to select all, then press Ctrl+C (Mac: Cmd+C) to copy, or click the button',
+    webhookUrlHintEither:
+      'Use either of the two URLs above in your platform configuration',
     logLevel: 'Log Level',
     allLevels: 'All Levels',
     selectLevel: 'Select Level',

web/src/i18n/locales/ja-JP.ts

@@ -1,4 +1,4 @@
-const jaJP = {
+const jaJP = {
   common: {
     login: 'ログイン',
     logout: 'ログアウト',
@@ -12,6 +12,10 @@ const jaJP = {
     loginSuccess: 'ログインに成功しました',
     loginFailed:
       'ログインに失敗しました。メールアドレスまたはパスワードをご確認ください',
+    loginLoadError: 'サーバーに接続できません',
+    loginLoadErrorDesc:
+      'LangBot バックエンドに接続できません。サービスが起動していることを確認してから再試行してください。',
+    retry: '再試行',
     enterEmail: 'メールアドレスを入力',
     enterPassword: 'パスワードを入力',
     invalidEmail: '有効なメールアドレスを入力してください',
@@ -241,6 +245,11 @@ const jaJP = {
     modelsCount: '{{count}} 個のモデル',
     expandModels: '展開',
     collapseModels: '折りたたむ',
+    fallback: {
+      primary: 'プライマリモデル',
+      fallbackList: 'フォールバックモデル',
+      addFallback: 'フォールバックモデルを追加',
+    },
   },
   bots: {
     title: 'ボット',
@@ -284,6 +293,8 @@ const jaJP = {
     webhookUrlCopied: 'Webhook URL をコピーしました',
     webhookUrlHint:
       '入力ボックスをクリックして全選択し、Ctrl+C (Mac: Cmd+C) でコピーするか、右側のボタンをクリックしてください',
+    webhookUrlHintEither:
+      '上記の2つのURLのいずれかをプラットフォーム設定に使用してください',
     logLevel: 'ログレベル',
     allLevels: 'すべてのレベル',
     selectLevel: 'レベルを選択',

web/src/i18n/locales/zh-Hans.ts

@@ -11,6 +11,9 @@ const zhHans = {
     continueToLogin: '登录以继续',
     loginSuccess: '登录成功',
     loginFailed: '登录失败，请检查邮箱和密码是否正确',
+    loginLoadError: '无法连接到服务器',
+    loginLoadErrorDesc: '无法连接到 LangBot 后端服务，请确认服务已启动后重试。',
+    retry: '重试',
     enterEmail: '输入邮箱地址',
     enterPassword: '输入密码',
     invalidEmail: '请输入有效的邮箱地址',
@@ -227,6 +230,11 @@ const zhHans = {
     modelsCount: '{{count}} 个模型',
     expandModels: '展开',
     collapseModels: '收起',
+    fallback: {
+      primary: '主模型',
+      fallbackList: '备用模型',
+      addFallback: '添加备用模型',
+    },
   },
   bots: {
     title: '机器人',
@@ -268,6 +276,7 @@ const zhHans = {
     webhookUrlCopied: 'Webhook 地址已复制',
     webhookUrlHint:
       '点击输入框自动全选，然后按 Ctrl+C (Mac: Cmd+C) 复制，或点击右侧按钮',
+    webhookUrlHintEither: '以上两个地址任选其一填入平台配置即可',
     logLevel: '日志级别',
     allLevels: '全部级别',
     selectLevel: '选择级别',

web/src/i18n/locales/zh-Hant.ts

@@ -11,6 +11,9 @@ const zhHant = {
     continueToLogin: '登入以繼續',
     loginSuccess: '登入成功',
     loginFailed: '登入失敗，請檢查電子郵件和密碼是否正確',
+    loginLoadError: '無法連線到伺服器',
+    loginLoadErrorDesc: '無法連線到 LangBot 後端服務，請確認服務已啟動後重試。',
+    retry: '重試',
     enterEmail: '輸入電子郵件地址',
     enterPassword: '輸入密碼',
     invalidEmail: '請輸入有效的電子郵件地址',
@@ -226,6 +229,11 @@ const zhHant = {
     modelsCount: '{{count}} 個模型',
     expandModels: '展開',
     collapseModels: '收起',
+    fallback: {
+      primary: '主模型',
+      fallbackList: '備用模型',
+      addFallback: '新增備用模型',
+    },
   },
   bots: {
     title: '機器人',
@@ -267,6 +275,7 @@ const zhHant = {
     webhookUrlCopied: 'Webhook 位址已複製',
     webhookUrlHint:
       '點擊輸入框自動全選，然後按 Ctrl+C (Mac: Cmd+C) 複製，或點擊右側按鈕',
+    webhookUrlHintEither: '以上兩個地址任選其一填入平台配置即可',
     logLevel: '日誌級別',
     allLevels: '全部級別',
     selectLevel: '選擇級別',