feat: Add pipeline_uuid handling across multiple adapters and update related tests

feat: Add missing completed input lines handling in DingTalk adapter
Merge remote-tracking branch 'origin/master' into feat/card_dify_human_input
2026-09-05 17:17:14 +00:00 · 2026-07-12 19:40:59 +08:00 · 2026-07-12 18:53:13 +08:00 · 2026-07-12 18:39:48 +08:00 · 2026-07-12 00:31:12 +08:00 · 2026-07-11 22:40:08 +08:00
35 changed files with 272 additions and 1516 deletions
@@ -92,17 +92,6 @@ docker compose --profile all up -d
 ---
 ## Live Demo
 **Try it now:** https://demo.langbot.dev/
 - Email: `demo@langbot.app`
 - Password: `langbot123456`
 _Note: Public demo environment. Do not enter sensitive information._
 ---
 ## Supported Platforms
 | Platform | Status | Notes |
@@ -178,6 +167,17 @@ LangBot is **agent-friendly by design** — your coding agents (Claude Code, Cod
 ---
 ## Live Demo
 **Try it now:** https://demo.langbot.dev/
 - Email: `demo@langbot.app`
 - Password: `langbot123456`
 _Note: Public demo environment. Do not enter sensitive information._
 ---
 ## Community
 [![Discord](https://img.shields.io/discord/1335141740050649118?logo=discord&label=Discord)](https://discord.gg/wdNEHETs87)
@@ -186,6 +186,12 @@ LangBot is **agent-friendly by design** — your coding agents (Claude Code, Cod
 ---
 ## Star History
 [![Star History Chart](https://api.star-history.com/svg?repos=langbot-app/LangBot&type=Date)](https://star-history.com/#langbot-app/LangBot&Date)
 ---
 ## Contributors
 Thanks to all [contributors](https://github.com/langbot-app/LangBot/graphs/contributors) who have helped make LangBot better:
@@ -92,16 +92,6 @@ docker compose --profile all up -d
 ---
 ## 在线演示
 **立即体验：** https://demo.langbot.dev/
 - 邮箱：`demo@langbot.app`
 - 密码：`langbot123456`
 *注意：公开演示环境，请不要在其中填入任何敏感信息。*
 ---
 ## 支持的平台
 | 平台 | 状态 | 备注 |
@@ -180,6 +170,16 @@ docker compose --profile all up -d
 ---
 ## 在线演示
 **立即体验：** https://demo.langbot.dev/
 - 邮箱：`demo@langbot.app`
 - 密码：`langbot123456`
 *注意：公开演示环境，请不要在其中填入任何敏感信息。*
 ---
 ## 为 AI Agent 而生 🤖
 LangBot **从设计上就对 Agent 友好** —— 你的编码 Agent（Claude Code、Codex、Copilot、Cursor 等）可以一等公民般地操作、扩展和部署 LangBot：
@@ -203,6 +203,12 @@ LangBot **从设计上就对 Agent 友好** —— 你的编码 Agent（Claude C
 ---
 ## Star 趋势
 [![Star History Chart](https://api.star-history.com/svg?repos=langbot-app/LangBot&type=Date)](https://star-history.com/#langbot-app/LangBot&Date)
 ---
 ## 贡献者
 感谢所有[贡献者](https://github.com/langbot-app/LangBot/graphs/contributors)对 LangBot 的帮助：
@@ -91,16 +91,6 @@ docker compose --profile all up -d
 ---
 ## Demo en Vivo
 **Pruébelo ahora:** https://demo.langbot.dev/
 - Correo electrónico: `demo@langbot.app`
 - Contraseña: `langbot123456`
 *Nota: Entorno de demostración público. No ingrese información confidencial.*
 ---
 ## Plataformas Soportadas
 | Plataforma | Estado | Notas |
@@ -163,6 +153,14 @@ docker compose --profile all up -d
 ---
 ## Demo en Vivo
 **Pruébelo ahora:** https://demo.langbot.dev/
 - Correo electrónico: `demo@langbot.app`
 - Contraseña: `langbot123456`
 *Nota: Entorno de demostración público. No ingrese información confidencial.*
 ## Diseñado para Agentes de IA 🤖
 LangBot es **agent-friendly por diseño** —— tus agentes de codificación (Claude Code, Codex, Copilot, Cursor, …) pueden operar, extender y desplegar LangBot con soporte de primera clase:
@@ -184,6 +182,12 @@ LangBot es **agent-friendly por diseño** —— tus agentes de codificación (C
 ---
 ## Historial de Stars
 [![Star History Chart](https://api.star-history.com/svg?repos=langbot-app/LangBot&type=Date)](https://star-history.com/#langbot-app/LangBot&Date)
 ---
 ## Colaboradores
 Gracias a todos los [colaboradores](https://github.com/langbot-app/LangBot/graphs/contributors) que han ayudado a mejorar LangBot:
@@ -91,16 +91,6 @@ docker compose --profile all up -d
 ---
 ## Démo en Ligne
 **Essayez maintenant :** https://demo.langbot.dev/
 - Email : `demo@langbot.app`
 - Mot de passe : `langbot123456`
 *Note : Environnement de démonstration public. Ne saisissez pas d'informations sensibles.*
 ---
 ## Plateformes Supportées
 | Plateforme | Statut | Notes |
@@ -163,6 +153,14 @@ docker compose --profile all up -d
 ---
 ## Démo en Ligne
 **Essayez maintenant :** https://demo.langbot.dev/
 - Email : `demo@langbot.app`
 - Mot de passe : `langbot123456`
 *Note : Environnement de démonstration public. Ne saisissez pas d'informations sensibles.*
 ## Conçu pour les agents IA 🤖
 LangBot est **agent-friendly par conception** —— vos agents de codage (Claude Code, Codex, Copilot, Cursor, …) peuvent exploiter, étendre et déployer LangBot avec un support de premier ordre :
@@ -184,6 +182,12 @@ LangBot est **agent-friendly par conception** —— vos agents de codage (Claud
 ---
 ## Historique des Stars
 [![Star History Chart](https://api.star-history.com/svg?repos=langbot-app/LangBot&type=Date)](https://star-history.com/#langbot-app/LangBot&Date)
 ---
 ## Contributeurs
 Merci à tous les [contributeurs](https://github.com/langbot-app/LangBot/graphs/contributors) qui ont aidé à améliorer LangBot :
@@ -91,16 +91,6 @@ docker compose --profile all up -d
 ---
 ## ライブデモ
 **今すぐ試す:** https://demo.langbot.dev/
 - メール: `demo@langbot.app`
 - パスワード: `langbot123456`
 *注意: 公開デモ環境です。機密情報を入力しないでください。*
 ---
 ## 対応プラットフォーム
 | プラットフォーム | ステータス | 備考 |
@@ -163,6 +153,14 @@ docker compose --profile all up -d
 ---
 ## ライブデモ
 **今すぐ試す:** https://demo.langbot.dev/
 - メール: `demo@langbot.app`
 - パスワード: `langbot123456`
 *注意: 公開デモ環境です。機密情報を入力しないでください。*
 ## AI エージェントのために 🤖
 LangBot は **設計段階からエージェントフレンドリー** です。お使いのコーディングエージェント（Claude Code、Codex、Copilot、Cursor など）が、ファーストクラスのサポートで LangBot を操作・拡張・デプロイできます：
@@ -184,6 +182,12 @@ LangBot は **設計段階からエージェントフレンドリー** です。
 ---
 ## Star 推移
 [![Star History Chart](https://api.star-history.com/svg?repos=langbot-app/LangBot&type=Date)](https://star-history.com/#langbot-app/LangBot&Date)
 ---
 ## コントリビューター
 LangBot をより良くするために貢献してくださったすべての[コントリビューター](https://github.com/langbot-app/LangBot/graphs/contributors)に感謝します:
@@ -91,16 +91,6 @@ docker compose --profile all up -d
 ---
 ## 라이브 데모
 **지금 체험:** https://demo.langbot.dev/
 - 이메일: `demo@langbot.app`
 - 비밀번호: `langbot123456`
 *참고: 공개 데모 환경입니다. 민감한 정보를 입력하지 마세요.*
 ---
 ## 지원 플랫폼
 | 플랫폼 | 상태 | 비고 |
@@ -163,6 +153,14 @@ docker compose --profile all up -d
 ---
 ## 라이브 데모
 **지금 체험:** https://demo.langbot.dev/
 - 이메일: `demo@langbot.app`
 - 비밀번호: `langbot123456`
 *참고: 공개 데모 환경입니다. 민감한 정보를 입력하지 마세요.*
 ## AI 에이전트를 위한 설계 🤖
 LangBot은 **설계 단계부터 에이전트 친화적**입니다 —— 코딩 에이전트(Claude Code, Codex, Copilot, Cursor 등)가 일급 지원으로 LangBot을 운영·확장·배포할 수 있습니다:
@@ -184,6 +182,12 @@ LangBot은 **설계 단계부터 에이전트 친화적**입니다 —— 코딩
 ---
 ## Star 추이
 [![Star History Chart](https://api.star-history.com/svg?repos=langbot-app/LangBot&type=Date)](https://star-history.com/#langbot-app/LangBot&Date)
 ---
 ## 기여자
 LangBot을 더 나은 프로젝트로 만들어 주신 모든 [기여자](https://github.com/langbot-app/LangBot/graphs/contributors)분들께 감사드립니다:
@@ -91,16 +91,6 @@ docker compose --profile all up -d
 ---
 ## Демо
 **Попробуйте прямо сейчас:** https://demo.langbot.dev/
 - Email: `demo@langbot.app`
 - Пароль: `langbot123456`
 *Примечание: Публичная демо-среда. Не вводите конфиденциальную информацию.*
 ---
 ## Поддерживаемые платформы
 | Платформа | Статус | Примечания |
@@ -163,6 +153,14 @@ docker compose --profile all up -d
 ---
 ## Демо
 **Попробуйте прямо сейчас:** https://demo.langbot.dev/
 - Email: `demo@langbot.app`
 - Пароль: `langbot123456`
 *Примечание: Публичная демо-среда. Не вводите конфиденциальную информацию.*
 ## Создано для ИИ-агентов 🤖
 LangBot **дружелюбен к агентам по своей архитектуре** —— ваши кодинг-агенты (Claude Code, Codex, Copilot, Cursor и др.) могут управлять, расширять и развёртывать LangBot с первоклассной поддержкой:
@@ -184,6 +182,12 @@ LangBot **дружелюбен к агентам по своей архитек
 ---
 ## История Stars
 [![Star History Chart](https://api.star-history.com/svg?repos=langbot-app/LangBot&type=Date)](https://star-history.com/#langbot-app/LangBot&Date)
 ---
 ## Участники
 Спасибо всем [участникам](https://github.com/langbot-app/LangBot/graphs/contributors), которые помогли сделать LangBot лучше:
@@ -93,16 +93,6 @@ docker compose --profile all up -d
 ---
 ## 線上演示
 **立即體驗：** https://demo.langbot.dev/
 - 信箱：`demo@langbot.app`
 - 密碼：`langbot123456`
 *注意：公開演示環境，請不要在其中填入任何敏感資訊。*
 ---
 ## 支援的平台
 | 平台 | 狀態 | 備註 |
@@ -179,6 +169,14 @@ docker compose --profile all up -d
 ---
 ## 線上演示
 **立即體驗：** https://demo.langbot.dev/
 - 信箱：`demo@langbot.app`
 - 密碼：`langbot123456`
 *注意：公開演示環境，請不要在其中填入任何敏感資訊。*
 ## 為 AI Agent 而生 🤖
 LangBot **從設計上就對 Agent 友善** —— 你的編碼 Agent（Claude Code、Codex、Copilot、Cursor 等）可以一等公民般地操作、擴充和部署 LangBot：
@@ -202,6 +200,12 @@ LangBot **從設計上就對 Agent 友善** —— 你的編碼 Agent（Claude C
 ---
 ## Star 趨勢
 [![Star History Chart](https://api.star-history.com/svg?repos=langbot-app/LangBot&type=Date)](https://star-history.com/#langbot-app/LangBot&Date)
 ---
 ## 貢獻者
 感謝所有[貢獻者](https://github.com/langbot-app/LangBot/graphs/contributors)對 LangBot 的幫助：
@@ -91,16 +91,6 @@ docker compose --profile all up -d
 ---
 ## Demo trực tuyến
 **Thử ngay:** https://demo.langbot.dev/
 - Email: `demo@langbot.app`
 - Mật khẩu: `langbot123456`
 *Lưu ý: Môi trường demo công khai. Không nhập thông tin nhạy cảm.*
 ---
 ## Nền tảng được hỗ trợ
 | Nền tảng | Trạng thái | Ghi chú |
@@ -163,6 +153,14 @@ docker compose --profile all up -d
 ---
 ## Demo trực tuyến
 **Thử ngay:** https://demo.langbot.dev/
 - Email: `demo@langbot.app`
 - Mật khẩu: `langbot123456`
 *Lưu ý: Môi trường demo công khai. Không nhập thông tin nhạy cảm.*
 ## Được xây dựng cho AI Agent 🤖
 LangBot **thân thiện với agent ngay từ thiết kế** —— các coding agent của bạn (Claude Code, Codex, Copilot, Cursor, …) có thể vận hành, mở rộng và triển khai LangBot với sự hỗ trợ hạng nhất:
@@ -184,6 +182,12 @@ LangBot **thân thiện với agent ngay từ thiết kế** —— các coding
 ---
 ## Lịch sử Star
 [![Star History Chart](https://api.star-history.com/svg?repos=langbot-app/LangBot&type=Date)](https://star-history.com/#langbot-app/LangBot&Date)
 ---
 ## Người đóng góp
 Cảm ơn tất cả [người đóng góp](https://github.com/langbot-app/LangBot/graphs/contributors) đã giúp LangBot trở nên tốt hơn:
@@ -31,10 +31,8 @@ Both are compatible with LangBot.
 ## Installation
-Valkey Search support is included automatically on Linux and macOS. The official `valkey-glide`
+Valkey Search support is included when you install LangBot — the `valkey-glide` dependency is
-client does not currently publish a Windows package, so LangBot skips this optional dependency on
+declared in `pyproject.toml`. To install manually:
 Windows; LangBot remains usable there, but the Valkey Search backend is unavailable. To install the
 client manually on a supported platform:
 ```bash
 pip install 'valkey-glide>=2.4.1,<3.0.0'
@@ -1,6 +1,6 @@
 [project]
 name = "langbot"
-version = "4.10.6"
+version = "4.10.5"
 description = "Production-grade platform for building agentic IM bots"
 readme = "README.md"
 license-files = ["LICENSE"]
@@ -80,7 +80,7 @@ dependencies = [
    "pgvector>=0.4.1",
    "botocore>=1.42.39",
    "litellm>=1.0.0",
-    "valkey-glide>=2.4.1,<3.0.0; sys_platform != 'win32'", # No Windows wheels are published
+    "valkey-glide>=2.4.1,<3.0.0",
 ]
 keywords = [
    "bot",
@@ -885,10 +885,7 @@ class DingTalkClient:
        while not self._stopped:
            try:
-                # open_connection performs blocking network I/O in the DingTalk SDK.
+                connection = self.client.open_connection()
                # Run it off the event loop so connection stalls do not block the
                # LangBot HTTP server and other async tasks.
                connection = await asyncio.to_thread(self.client.open_connection)
                if not connection:
                    if self.logger:
@@ -21,7 +21,7 @@ import quart
 from ... import group
 from ......utils import paths
-from ......platform.sources.websocket_manager import is_valid_session_id, ws_connection_manager
+from ......platform.sources.websocket_manager import ws_connection_manager
 logger = logging.getLogger(__name__)
@@ -203,15 +203,11 @@ class EmbedRouterGroup(group.RouterGroup):
                if session_type not in ['person', 'group']:
                    return self.http_status(400, -1, 'session_type must be person or group')
                session_id = quart.request.args.get('session_id', '')
                if not is_valid_session_id(session_id):
                    return self.http_status(400, -1, 'Valid session_id is required')
                websocket_adapter = self.ap.platform_mgr.websocket_proxy_bot.adapter
                if not websocket_adapter:
                    return self.http_status(404, -1, 'WebSocket adapter not found')
-                messages = websocket_adapter.get_websocket_messages(pipeline_uuid, session_type, session_id)
+                messages = websocket_adapter.get_websocket_messages(pipeline_uuid, session_type)
                return self.success(data={'messages': messages})
            except Exception as e:
@@ -231,15 +227,11 @@ class EmbedRouterGroup(group.RouterGroup):
                if session_type not in ['person', 'group']:
                    return self.http_status(400, -1, 'session_type must be person or group')
                session_id = quart.request.args.get('session_id', '')
                if not is_valid_session_id(session_id):
                    return self.http_status(400, -1, 'Valid session_id is required')
                websocket_adapter = self.ap.platform_mgr.websocket_proxy_bot.adapter
                if not websocket_adapter:
                    return self.http_status(404, -1, 'WebSocket adapter not found')
-                websocket_adapter.reset_session(pipeline_uuid, session_type, session_id)
+                websocket_adapter.reset_session(pipeline_uuid, session_type)
                return self.success(data={'message': 'Session reset successfully'})
            except Exception as e:
@@ -302,11 +294,6 @@ class EmbedRouterGroup(group.RouterGroup):
                )
                return
            session_id = quart.websocket.args.get('session_id', '')
            if not is_valid_session_id(session_id):
                await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Valid session_id is required'}))
                return
            websocket_adapter = self.ap.platform_mgr.websocket_proxy_bot.adapter
            if not websocket_adapter:
                await quart.websocket.send(json.dumps({'type': 'error', 'message': 'WebSocket adapter not found'}))
@@ -317,7 +304,6 @@ class EmbedRouterGroup(group.RouterGroup):
                    websocket=quart.websocket._get_current_object(),
                    pipeline_uuid=pipeline_uuid,
                    session_type=session_type,
                    session_id=session_id,
                    metadata={'user_agent': quart.websocket.headers.get('User-Agent', '')},
                )
@@ -491,11 +491,7 @@ class AiocqhttpAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
    message_converter: AiocqhttpMessageConverter = AiocqhttpMessageConverter()
    event_converter: AiocqhttpEventConverter = pydantic.Field(default_factory=AiocqhttpEventConverter)
-    on_websocket_connection_event_cache: list[aiocqhttp.Event] = []
+    on_websocket_connection_event_cache: typing.List[typing.Callable[[aiocqhttp.Event], None]] = []
    _listener_wrappers: dict[
        tuple[typing.Type[platform_events.Event], typing.Callable],
        tuple[str, typing.Callable],
    ] = {}
    def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger):
        super().__init__(
@@ -510,7 +506,6 @@ class AiocqhttpAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
        self.config['shutdown_trigger'] = shutdown_trigger_placeholder
        self.on_websocket_connection_event_cache = []
        self._listener_wrappers = {}
        if 'access-token' in config:
            self.bot = aiocqhttp.CQHttp(access_token=config['access-token'])
@@ -518,16 +513,6 @@ class AiocqhttpAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
        else:
            self.bot = aiocqhttp.CQHttp()
        self.bot.on_websocket_connection(self._on_websocket_connection)
    async def _on_websocket_connection(self, event: aiocqhttp.Event):
        for cached_event in self.on_websocket_connection_event_cache:
            if cached_event.self_id == event.self_id and cached_event.time == event.time:
                return
        self.on_websocket_connection_event_cache.append(event)
        await self.logger.info(f'WebSocket connection established, bot id: {event.self_id}')
    async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
        # Check if message contains a Forward component
        forward_msg = message.get_first(platform_message.Forward)
@@ -663,14 +648,22 @@ class AiocqhttpAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
        if event_type == platform_events.GroupMessage:
            self.bot.on_message('group')(on_message)
            self._listener_wrappers[(event_type, callback)] = ('message.group', on_message)
            # self.bot.on_notice()(on_message)
        elif event_type == platform_events.FriendMessage:
            self.bot.on_message('private')(on_message)
            self._listener_wrappers[(event_type, callback)] = ('message.private', on_message)
            # self.bot.on_notice()(on_message)
        # print(event_type)
        async def on_websocket_connection(event: aiocqhttp.Event):
            for event in self.on_websocket_connection_event_cache:
                if event.self_id == event.self_id and event.time == event.time:
                    return
            self.on_websocket_connection_event_cache.append(event)
            await self.logger.info(f'WebSocket connection established, bot id: {event.self_id}')
        self.bot.on_websocket_connection(on_websocket_connection)
    def unregister_listener(
        self,
        event_type: typing.Type[platform_events.Event],
@@ -678,12 +671,7 @@ class AiocqhttpAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
            [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], None
        ],
    ):
-        listener = self._listener_wrappers.pop((event_type, callback), None)
+        return super().unregister_listener(event_type, callback)
        if listener is None:
            return
        event_name, wrapper = listener
        self.bot._bus.unsubscribe(event_name, wrapper)
    async def run_async(self):
        await self.bot._server_app.run_task(**self.config)
@@ -242,33 +242,6 @@ def _lark_extract_action_form_inputs(action: typing.Any, action_value_obj: dict)
    return form_inputs
 class NonBlockingLarkWSClient(lark_oapi.ws.Client):
    """Keep the SDK's synchronous connection lookup off LangBot's event loop.
    lark-oapi performs ``requests.post`` inside its async ``_connect`` method.
    A stalled TLS handshake therefore freezes Quart and every other adapter in
    the process. Pre-fetch the URL in a worker thread, then let the SDK finish
    the WebSocket setup on the main loop with the already-resolved URL.
    """
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._langbot_connect_lock = asyncio.Lock()
    async def _connect(self) -> None:
        async with self._langbot_connect_lock:
            if self._conn is not None:
                return
            conn_url = await asyncio.to_thread(self._get_conn_url)
            original_get_conn_url = self._get_conn_url
            self._get_conn_url = lambda: conn_url
            try:
                await super()._connect()
            finally:
                self._get_conn_url = original_get_conn_url
 class AESCipher(object):
    def __init__(self, key):
        self.bs = AES.block_size
@@ -1267,9 +1240,7 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
        bot_account_id = config['bot_name']
        domain = self._resolve_domain(config)
-        bot = NonBlockingLarkWSClient(
+        bot = lark_oapi.ws.Client(config['app_id'], config['app_secret'], event_handler=event_handler, domain=domain)
            config['app_id'], config['app_secret'], event_handler=event_handler, domain=domain
        )
        api_client = self.build_api_client(config)
        cipher = AESCipher(config.get('encrypt-key', ''))
        self.request_app_ticket(api_client, config)
@@ -1436,27 +1407,6 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
            )
        return api_client
    @staticmethod
    def _has_markdown_table(text_elements: list) -> bool:
        """Check if text elements contain markdown table syntax (|...|).
        A markdown table requires:
        - At least one line with pipe characters (|)
        - A separator line with pipes and dashes (|---|)
        """
        # Regex to detect markdown table: lines with pipes and separator lines with dashes
        table_pattern = re.compile(r'^\s*\|.+\|\s*$', re.MULTILINE)
        separator_pattern = re.compile(r'^\s*\|\s*-+(\s*\|\s*-+)*\s*\|\s*$', re.MULTILINE)
        for paragraph in text_elements:
            for ele in paragraph:
                if ele.get('tag') == 'md':
                    text = ele.get('text', '')
                    # Quick check: if has pipes and separator line, it's likely a table
                    if '|' in text and table_pattern.search(text) and separator_pattern.search(text):
                        return True
        return False
    async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
        text_elements, media_items = await self.message_converter.yiri2target(message, self.api_client)
@@ -1470,10 +1420,7 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
        # Send text message if there are text elements
        if text_elements:
-            # Use 'post' format if: has @mentions OR has markdown tables
+            needs_post = any(ele['tag'] == 'at' for paragraph in text_elements for ele in paragraph)
            has_at = any(ele['tag'] == 'at' for paragraph in text_elements for ele in paragraph)
            has_table = self._has_markdown_table(text_elements)
            needs_post = has_at or has_table
            if needs_post:
                msg_type = 'post'
@@ -1967,10 +1914,9 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
        # Send text message if there are text elements
        if text_elements:
-            # Use 'post' format if: has @mentions OR has markdown tables
+            # Determine msg_type based on content: use 'post' if at mentions
-            has_at = any(ele['tag'] == 'at' for paragraph in text_elements for ele in paragraph)
+            # are present (requires post paragraph structure), otherwise 'text'
-            has_table = self._has_markdown_table(text_elements)
+            needs_post = any(ele['tag'] == 'at' for paragraph in text_elements for ele in paragraph)
            needs_post = has_at or has_table
            if needs_post:
                msg_type = 'post'
@@ -13,7 +13,7 @@ import langbot_plugin.api.entities.builtin.platform.events as platform_events
 import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
 import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
 from ...core import app
-from .websocket_manager import WebSocketConnection, is_valid_session_id, ws_connection_manager
+from .websocket_manager import ws_connection_manager, WebSocketConnection
 logger = logging.getLogger(__name__)
@@ -91,59 +91,6 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
        self.outbound_message_queue = asyncio.Queue()
        self.stream_enabled = True
    @staticmethod
    def _conversation_key(pipeline_uuid: str, session_id: str | None = None) -> str:
        """Return the history key for a pipeline/client conversation."""
        return f'{pipeline_uuid}:{session_id}' if session_id else pipeline_uuid
    @staticmethod
    def _parse_embed_target(target_id: str) -> tuple[str, str] | None:
        """Extract pipeline and session identifiers from a stable embed launcher."""
        target_value = str(target_id)
        for prefix in ('websocket_', 'websocketgroup_'):
            if target_value.startswith(prefix):
                target = target_value[len(prefix) :]
                break
        else:
            return None
        if ':' not in target:
            return None
        pipeline_uuid, session_id = target.rsplit(':', 1)
        if not pipeline_uuid or not is_valid_session_id(session_id):
            return None
        return pipeline_uuid, session_id
    @classmethod
    async def _get_connection_from_target(cls, target_id: str):
        """Resolve a person or group WebSocket launcher to its connection."""
        target_value = str(target_id)
        for prefix in ('websocket_', 'websocketgroup_'):
            if target_value.startswith(prefix):
                target = target_value[len(prefix) :]
                break
        else:
            return None
        connection = await ws_connection_manager.get_connection(target)
        if connection is not None:
            return connection
        embed_target = cls._parse_embed_target(target_id)
        if embed_target is not None:
            pipeline_uuid, session_id = embed_target
            return await ws_connection_manager.get_connection_by_session_id(session_id, pipeline_uuid)
        return await ws_connection_manager.get_connection_by_session_id(target)
    async def _get_message_context(self, message_source) -> tuple[str, str | None]:
        """Resolve the originating pipeline and browser session for a reply."""
        sender = getattr(message_source, 'sender', None)
        sender_id = getattr(sender, 'id', '')
        connection = await self._get_connection_from_target(sender_id)
        if connection is not None:
            return connection.pipeline_uuid, connection.session_id
        embed_target = self._parse_embed_target(sender_id)
        if embed_target is not None:
            return embed_target
        return typing.cast(str, self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid), None
    async def send_message(
        self,
        target_type: str,
@@ -156,26 +103,15 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
        target_id 可能是 launcher_id（如 websocket_xxx）或 pipeline_uuid。
        我们需要尝试两种方式来确保消息能够送达。
        """
-        connection = await self._get_connection_from_target(target_id)
+        # 获取当前的 pipeline_uuid
-        if connection is not None:
+        pipeline_uuid = self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid
            pipeline_uuid = connection.pipeline_uuid
            session_id = connection.session_id
        else:
            embed_target = self._parse_embed_target(target_id)
            if embed_target is not None:
                pipeline_uuid, session_id = embed_target
            else:
                pipeline_uuid = typing.cast(
                    str,
                    self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid,
                )
                session_id = None
        session_type = 'group' if target_type == 'group' else 'person'
        conversation_key = self._conversation_key(pipeline_uuid, session_id)
        # 选择会话
        session = self.websocket_group_session if session_type == 'group' else self.websocket_person_session
-        msg_id = len(session.get_message_list(conversation_key)) + 1
+        # 生成唯一消息ID
        msg_id = len(session.get_message_list(pipeline_uuid)) + 1
        message_data = WebSocketMessage(
            id=msg_id,
@@ -186,8 +122,10 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
            is_final=True,
        )
-        session.get_message_list(conversation_key).append(message_data)
+        # 保存到历史记录
        session.get_message_list(pipeline_uuid).append(message_data)
        # 直接广播到当前pipeline的连接
        await ws_connection_manager.broadcast_to_pipeline(
            pipeline_uuid,
            {
@@ -196,7 +134,6 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
                'data': message_data.model_dump(),
            },
            session_type=session_type,
            session_id=session_id,
        )
        return message_data.model_dump()
@@ -215,11 +152,12 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
            else self.websocket_person_session
        )
-        pipeline_uuid, session_id = await self._get_message_context(message_source)
+        # 从message_source获取pipeline_uuid和connection_id
        pipeline_uuid = self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid
        session_type = 'group' if isinstance(message_source, platform_events.GroupMessage) else 'person'
        conversation_key = self._conversation_key(pipeline_uuid, session_id)
-        msg_id = len(session.get_message_list(conversation_key)) + 1
+        # 生成新的消息ID
        msg_id = len(session.get_message_list(pipeline_uuid)) + 1
        message_data = WebSocketMessage(
            id=msg_id,
@@ -230,8 +168,10 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
            is_final=True,
        )
-        session.get_message_list(conversation_key).append(message_data)
+        # 保存到历史记录
        session.get_message_list(pipeline_uuid).append(message_data)
        # 直接广播到所有该pipeline的连接，包含session_type信息
        await ws_connection_manager.broadcast_to_pipeline(
            pipeline_uuid,
            {
@@ -240,7 +180,6 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
                'data': message_data.model_dump(),
            },
            session_type=session_type,
            session_id=session_id,
        )
        return message_data.model_dump()
@@ -261,11 +200,10 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
            else self.websocket_person_session
        )
-        pipeline_uuid, session_id = await self._get_message_context(message_source)
+        pipeline_uuid = self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid
        session_type = 'group' if isinstance(message_source, platform_events.GroupMessage) else 'person'
-        conversation_key = self._conversation_key(pipeline_uuid, session_id)
+        message_list = session.get_message_list(pipeline_uuid)
-        message_list = session.get_message_list(conversation_key)
+        stream_message_indexes = session.get_stream_message_indexes(pipeline_uuid)
        stream_message_indexes = session.get_stream_message_indexes(conversation_key)
        # Streaming messages in LangBot have a stable resp_message_id during the same assistant reply.
        # Use it as the primary key to avoid overwriting an old card from a previous reply.
@@ -309,6 +247,7 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
        if message_is_final and resp_message_id:
            stream_message_indexes.pop(resp_message_id, None)
        # 直接广播到所有该pipeline的连接，包含session_type信息
        await ws_connection_manager.broadcast_to_pipeline(
            pipeline_uuid,
            {
@@ -317,7 +256,6 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
                'data': message_data.model_dump(),
            },
            session_type=session_type,
            session_id=session_id,
        )
        return message_data.model_dump()
@@ -443,19 +381,23 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
        """
        pipeline_uuid = connection.pipeline_uuid
        session_type = connection.session_type
        conversation_key = self._conversation_key(pipeline_uuid, connection.session_id)
        # 获取stream参数，默认为True
        self.stream_enabled = message_data.get('stream', True)
        # 选择会话
        use_session = self.websocket_group_session if session_type == 'group' else self.websocket_person_session
        # 解析消息链
        message_chain_obj = message_data.get('message', [])
        # 处理图片组件：将path转换为base64
        await self._process_image_components(message_chain_obj)
        message_chain = platform_message.MessageChain.model_validate(message_chain_obj)
-        message_id = len(use_session.get_message_list(conversation_key)) + 1
+        # 生成消息ID
        message_id = len(use_session.get_message_list(pipeline_uuid)) + 1
        # 保存用户消息
        user_message = WebSocketMessage(
@@ -467,8 +409,9 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
            connection_id=connection.connection_id,
            is_final=True,  # 用户消息始终是完整的，非流式
        )
-        use_session.get_message_list(conversation_key).append(user_message)
+        use_session.get_message_list(pipeline_uuid).append(user_message)
        # 广播用户消息到所有连接（包括发送者），包含session_type信息
        await ws_connection_manager.broadcast_to_pipeline(
            pipeline_uuid,
            {
@@ -477,27 +420,25 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
                'data': user_message.model_dump(),
            },
            session_type=session_type,
            session_id=connection.session_id,
        )
        # 添加消息源
        message_chain.insert(0, platform_message.Source(id=message_id, time=datetime.now().timestamp()))
        # 创建事件
        launcher_id = f'{pipeline_uuid}:{connection.session_id}' if connection.session_id else connection.connection_id
        if session_type == 'person':
-            sender = platform_entities.Friend(id=f'websocket_{launcher_id}', nickname='User', remark='User')
+            sender = platform_entities.Friend(
                id=f'websocket_{connection.connection_id}', nickname='User', remark='User'
            )
            event = platform_events.FriendMessage(
                sender=sender, message_chain=message_chain, time=datetime.now().timestamp()
            )
        else:
            group = platform_entities.Group(
-                id=f'websocketgroup_{launcher_id}' if connection.session_id else 'websocketgroup',
+                id='websocketgroup', name='Group', permission=platform_entities.Permission.Member
                name='Group',
                permission=platform_entities.Permission.Member,
            )
            sender = platform_entities.GroupMember(
-                id=f'websocket_{launcher_id}',
+                id=f'websocket_{connection.connection_id}',
                member_name='User',
                group=group,
                permission=platform_entities.Permission.Member,
@@ -527,47 +468,22 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
        if event.__class__ in listeners:
            asyncio.create_task(listeners[event.__class__](event, callback_adapter))
-    def get_websocket_messages(
+    def get_websocket_messages(self, pipeline_uuid: str, session_type: str) -> list[dict]:
-        self,
+        """获取消息历史"""
-        pipeline_uuid: str,
+        if session_type == 'person':
-        session_type: str,
+            return [message.model_dump() for message in self.websocket_person_session.get_message_list(pipeline_uuid)]
-        session_id: str | None = None,
+        else:
-    ) -> list[dict]:
+            return [message.model_dump() for message in self.websocket_group_session.get_message_list(pipeline_uuid)]
        """Return history for one pipeline/client conversation."""
        conversation_key = self._conversation_key(pipeline_uuid, session_id)
        session = self.websocket_person_session if session_type == 'person' else self.websocket_group_session
        return [message.model_dump() for message in session.message_lists.get(conversation_key, [])]
-    def reset_session(
+    def reset_session(self, pipeline_uuid: str, session_type: str):
-        self,
+        """重置会话"""
-        pipeline_uuid: str,
+        if session_type == 'person':
-        session_type: str,
+            if pipeline_uuid in self.websocket_person_session.message_lists:
-        session_id: str | None = None,
+                self.websocket_person_session.message_lists[pipeline_uuid] = []
-    ):
+            if pipeline_uuid in self.websocket_person_session.stream_message_indexes:
-        """Reset one pipeline/client conversation."""
+                self.websocket_person_session.stream_message_indexes[pipeline_uuid] = {}
-        conversation_key = self._conversation_key(pipeline_uuid, session_id)
+        else:
-        session = self.websocket_person_session if session_type == 'person' else self.websocket_group_session
+            if pipeline_uuid in self.websocket_group_session.message_lists:
-        if conversation_key in session.message_lists:
+                self.websocket_group_session.message_lists[pipeline_uuid] = []
-            session.message_lists[conversation_key] = []
+            if pipeline_uuid in self.websocket_group_session.stream_message_indexes:
-        if conversation_key in session.stream_message_indexes:
+                self.websocket_group_session.stream_message_indexes[pipeline_uuid] = {}
            session.stream_message_indexes[conversation_key] = {}
        if session_id:
            launcher_id = (
                f'websocketgroup_{pipeline_uuid}:{session_id}'
                if session_type == 'group'
                else f'websocket_{pipeline_uuid}:{session_id}'
            )
            self.ap.sess_mgr.session_list = [
                candidate_session
                for candidate_session in self.ap.sess_mgr.session_list
                if not (
                    str(
                        candidate_session.launcher_type.value
                        if hasattr(candidate_session.launcher_type, 'value')
                        else candidate_session.launcher_type
                    )
                    == session_type
                    and str(candidate_session.launcher_id) == launcher_id
                )
            ]
@@ -9,16 +9,6 @@ from datetime import datetime
 import pydantic
 logger = logging.getLogger(__name__)
 _SESSION_FILTER_UNSET = object()
 def is_valid_session_id(value: str) -> bool:
    """Accept only canonical random UUIDs for client conversation identifiers."""
    try:
        parsed = uuid.UUID(value)
    except (ValueError, TypeError, AttributeError):
        return False
    return parsed.version == 4 and str(parsed) == value
 class WebSocketConnection(pydantic.BaseModel):
@@ -35,9 +25,6 @@ class WebSocketConnection(pydantic.BaseModel):
    session_type: str  # 'person' or 'group'
    """会话类型"""
    session_id: str | None = None
    """Optional client conversation identifier used by embed widgets."""
    websocket: typing.Any = pydantic.Field(exclude=True)
    """WebSocket连接对象 (quart.websocket)"""
@@ -78,15 +65,13 @@ class WebSocketConnectionManager:
        websocket: typing.Any,
        pipeline_uuid: str,
        session_type: str,
-        metadata: dict | None = None,
+        metadata: dict = None,
        session_id: str | None = None,
    ) -> WebSocketConnection:
-        """Register a WebSocket connection and its optional embed session."""
+        """添加新的WebSocket连接"""
        async with self._lock:
            connection = WebSocketConnection(
                pipeline_uuid=pipeline_uuid,
                session_type=session_type,
                session_id=session_id,
                websocket=websocket,
                metadata=metadata or {},
            )
@@ -135,25 +120,10 @@ class WebSocketConnectionManager:
            logger.debug(f'WebSocket connection disconnected: {connection_id}')
-    async def get_connection(self, connection_id: str) -> WebSocketConnection | None:
+    async def get_connection(self, connection_id: str) -> typing.Optional[WebSocketConnection]:
-        """Get a connection by its transport identifier."""
+        """获取指定连接"""
        return self.connections.get(connection_id)
    async def get_connection_by_session_id(
        self,
        session_id: str,
        pipeline_uuid: str | None = None,
    ) -> WebSocketConnection | None:
        """Get an active embed connection by its stable browser session identifier."""
        for connection in self.connections.values():
            if (
                connection.session_id == session_id
                and connection.is_active
                and (pipeline_uuid is None or connection.pipeline_uuid == pipeline_uuid)
            ):
                return connection
        return None
    async def get_connections_by_pipeline(self, pipeline_uuid: str) -> list[WebSocketConnection]:
        """获取指定流水线的所有连接"""
        connection_ids = self.pipeline_connections.get(pipeline_uuid, set())
@@ -164,30 +134,20 @@ class WebSocketConnectionManager:
        connection_ids = self.session_connections.get(session_type, set())
        return [self.connections[cid] for cid in connection_ids if cid in self.connections]
-    async def broadcast_to_pipeline(
+    async def broadcast_to_pipeline(self, pipeline_uuid: str, message: dict, session_type: str = None):
-        self,
+        """向指定流水线的所有连接广播消息
        pipeline_uuid: str,
        message: dict,
        session_type: str | None = None,
        session_id: typing.Any = _SESSION_FILTER_UNSET,
    ):
        """Broadcast a message to matching connections for one pipeline.
        Args:
-            pipeline_uuid: Pipeline identifier.
+            pipeline_uuid: 流水线UUID
-            message: Serialized message to enqueue.
+            message: 要广播的消息
-            session_type: Optional session-type filter.
+            session_type: 可选的会话类型过滤器，如果提供则只向匹配的session_type连接广播
            session_id: Embed conversation filter. Omit it to broadcast across
                conversations; pass ``None`` to target non-embed connections.
        """
        connections = await self.get_connections_by_pipeline(pipeline_uuid)
        # 如果指定了session_type，只向匹配的连接广播
        if session_type is not None:
            connections = [conn for conn in connections if conn.session_type == session_type]
        if session_id is not _SESSION_FILTER_UNSET:
            connections = [conn for conn in connections if conn.session_id == session_id]
        tasks = []
        for conn in connections:
            tasks.append(self.send_to_connection(conn.connection_id, message))
@@ -13,151 +13,6 @@ import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
 import langbot_plugin.api.entities.builtin.provider.message as provider_message
 class _ThinkStripState:
    """Stateful filter that drops think blocks across chunks."""
    _THINK_OPEN = '<think>'
    _THINK_CLOSE = '</think>'
    _LEGACY_OPEN = 'CRETIRE_REASONING_BEGINk'
    _LEGACY_CLOSE = 'CRETIRE_REASONING_ENDk'
    def __init__(self) -> None:
        self._pairs: tuple[tuple[str, str], ...] = (
            (self._THINK_OPEN, self._THINK_CLOSE),
            (self._LEGACY_OPEN, self._LEGACY_CLOSE),
        )
        self._open_tags = tuple(open_tag for open_tag, _close_tag in self._pairs)
        self._buf = ''
        self._close_tag: str | None = None
        self._pending_initial = True
    def feed(self, chunk: str) -> str:
        """Feed a streaming delta and return user-visible content."""
        if not chunk:
            return chunk
        text = self._buf + chunk
        if self._close_tag is not None:
            return self._consume_think_body(text)
        return self._process_visible_text(text)
    def flush(self) -> str:
        """Release buffered visible content when the stream ends."""
        if self._close_tag is not None:
            self._buf = ''
            self._close_tag = None
            return ''
        pending, self._buf = self._buf, ''
        self._close_tag = None
        return pending
    def _consume_think_body(self, text: str) -> str:
        close_tag = self._close_tag
        if close_tag is None:
            return text
        close_idx = text.find(close_tag)
        if close_idx != -1:
            self._close_tag = None
            self._buf = ''
            self._pending_initial = False
            return self._process_visible_text(text[close_idx + len(close_tag) :])
        self._buf = self._close_prefix(text, close_tag)
        return ''
    def _process_visible_text(self, text: str) -> str:
        out: list[str] = []
        index = 0
        while index < len(text):
            if self._pending_initial:
                open_idx, open_tag, close_tag = self._find_next_open(text, index)
                orphan_close_idx, orphan_close_tag = self._find_next_close(text, index)
                if orphan_close_idx != -1 and (open_idx == -1 or orphan_close_idx < open_idx):
                    self._pending_initial = False
                    index = orphan_close_idx + len(orphan_close_tag)
                    continue
                if open_idx == -1:
                    self._buf = text[index:]
                    return ''.join(out)
                if open_idx > index:
                    self._pending_initial = False
                    out.append(text[index:open_idx])
                    index = open_idx
                    continue
            open_idx, open_tag, close_tag = self._find_next_open(text, index)
            if open_idx == -1:
                emit_end = self._visible_emit_end(text, index)
                out.append(text[index:emit_end])
                if emit_end > index:
                    self._pending_initial = False
                self._buf = text[emit_end:]
                return ''.join(out)
            out.append(text[index:open_idx])
            if open_idx > index:
                self._pending_initial = False
            body_start = open_idx + len(open_tag)
            close_idx = text.find(close_tag, body_start)
            if close_idx == -1:
                self._close_tag = close_tag
                self._buf = self._close_prefix(text[body_start:], close_tag)
                return ''.join(out)
            self._pending_initial = False
            index = close_idx + len(close_tag)
        self._buf = ''
        return ''.join(out)
    def _find_next_open(self, text: str, start: int) -> tuple[int, str, str]:
        best_idx = -1
        best_open = ''
        best_close = ''
        for open_tag, close_tag in self._pairs:
            idx = text.find(open_tag, start)
            if idx != -1 and (best_idx == -1 or idx < best_idx):
                best_idx = idx
                best_open = open_tag
                best_close = close_tag
        return best_idx, best_open, best_close
    def _find_next_close(self, text: str, start: int) -> tuple[int, str]:
        best_idx = -1
        best_close = ''
        for _open_tag, close_tag in self._pairs:
            idx = text.find(close_tag, start)
            if idx != -1 and (best_idx == -1 or idx < best_idx):
                best_idx = idx
                best_close = close_tag
        return best_idx, best_close
    def _visible_emit_end(self, text: str, start: int) -> int:
        visible = text[start:]
        limit = min(len(visible), max(len(open_tag) for open_tag in self._open_tags) - 1)
        for keep in range(limit, 0, -1):
            suffix = visible[-keep:]
            if any(open_tag.startswith(suffix) for open_tag in self._open_tags):
                return len(text) - keep
        return len(text)
    @staticmethod
    def _close_prefix(text: str, close_tag: str) -> str:
        limit = min(len(text), len(close_tag) - 1)
        for keep in range(limit, 0, -1):
            suffix = text[-keep:]
            if close_tag.startswith(suffix):
                return suffix
        return ''
 class LiteLLMRequester(requester.ProviderAPIRequester):
    """LiteLLM unified API requester supporting chat, embedding, and rerank."""
@@ -382,25 +237,6 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
        return req_messages
    _THINK_PATTERNS: tuple[str, ...] = (
        r'^\s*(?:(?!<think>).)*?</think>\s*',
        r'^\s*(?:(?!CRETIRE_REASONING_BEGINk).)*?CRETIRE_REASONING_ENDk\s*',
        r'<think>.*?</think>',
        r'CRETIRE_REASONING_BEGINk.*?CRETIRE_REASONING_ENDk',
    )
    @classmethod
    def _strip_think(cls, content: str) -> str:
        """Strip chain-of-thought blocks from ``content``."""
        if not content:
            return content
        import re
        for pattern in cls._THINK_PATTERNS:
            content = re.sub(pattern, '', content, flags=re.DOTALL)
        return content.strip()
    def _process_thinking_content(self, content: str, reasoning_content: str | None, remove_think: bool) -> str:
        """Process thinking/reasoning content.
@@ -412,12 +248,20 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
        Returns:
            Processed content string
        """
-        if remove_think and content:
+        # Extract and handle thinking tags
-            content = self._strip_think(content)
+        if content and 'CRETIRE_REASONING_BEGINk' in content and 'CRETIRE_REASONING_ENDk' in content:
            import re
-        if reasoning_content and not remove_think:
+            think_pattern = r'CRETIRE_REASONING_BEGINk(.*?)CRETIRE_REASONING_ENDk'
            content = f'<think>\n{reasoning_content}\n</think>\n{content or ""}'.strip()
            if remove_think:
                # Remove thinking tags and their content from output
                content = re.sub(think_pattern, '', content, flags=re.DOTALL).strip()
            # else: preserve thinking content as-is
        # Handle separate reasoning_content field
        # Currently we don't include reasoning_content in user-facing output regardless of remove_think
        # because it's typically internal model reasoning, not user-visible thinking
        return content or ''
    @staticmethod
@@ -726,7 +570,6 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
        chunk_idx = 0
        role = 'assistant'
        tool_call_state: dict[int, dict[str, typing.Any]] = {}
        think_state = _ThinkStripState() if remove_think else None
        try:
            response = await acompletion(**args)
@@ -770,12 +613,6 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
                        # Use reasoning_content as the displayed content
                        delta_content = reasoning_content
                if think_state is not None and delta_content:
                    delta_content = think_state.feed(delta_content)
                    if not delta_content:
                        chunk_idx += 1
                        continue
                tool_calls = self._normalize_stream_tool_calls(delta.get('tool_calls'), tool_call_state)
                if chunk_idx == 0 and not delta_content and not tool_calls:
@@ -797,15 +634,6 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
                yield provider_message.MessageChunk(**chunk_data)
                chunk_idx += 1
            if think_state is not None:
                pending_content = think_state.flush()
                if pending_content:
                    yield provider_message.MessageChunk(
                        role=role,
                        content=pending_content,
                        is_final=True,
                    )
        except Exception as e:
            self._handle_litellm_error(e)
@@ -1,5 +0,0 @@
 <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 64 64" role="img" aria-label="TokenLab">
  <rect width="64" height="64" rx="14" fill="#111827"/>
  <path fill="#38bdf8" d="M17 14h30v8H36v28h-8V22H17z"/>
  <path fill="#a7f3d0" d="M40 30h8v20H28v-8h12z"/>
 </svg>
@@ -1,29 +0,0 @@
 apiVersion: v1
 kind: LLMAPIRequester
 metadata:
  name: tokenlab-chat-completions
  label:
    en_US: TokenLab
    zh_Hans: TokenLab
  icon: tokenlab.svg
 spec:
  litellm_provider: openai
  config:
  - name: base_url
    label:
      en_US: Base URL
      zh_Hans: 基础 URL
    type: string
    required: true
    default: https://api.tokenlab.sh/v1
  - name: timeout
    label:
      en_US: Timeout
      zh_Hans: 超时时间
    type: integer
    required: true
    default: 120
  alias: "tokenlab TokenLab gpt claude gemini deepseek qwen kimi minimax glm grok openai-compatible"
  support_type:
  - llm
  provider_category: maas
@@ -49,23 +49,12 @@ def _model_has_ability(model: modelmgr_requester.RuntimeLLMModel, ability: str)
 class _StreamAccumulator:
    """Accumulate streamed content and fragmented OpenAI-style tool calls."""
-    def __init__(
+    def __init__(self, msg_sequence: int = 0, initial_content: str | None = None):
        self,
        msg_sequence: int = 0,
        initial_content: str | None = None,
        remove_think: bool = False,
    ):
        self.tool_calls_map: dict[str, provider_message.ToolCall] = {}
        self.msg_idx = 0
        self.accumulated_content = initial_content or ''
        self.last_role = 'assistant'
        self.msg_sequence = msg_sequence
        self.remove_think = remove_think
        self._think_state = None
        if remove_think:
            from ..modelmgr.requesters.litellmchat import _ThinkStripState
            self._think_state = _ThinkStripState()
    def add(self, msg: provider_message.MessageChunk) -> provider_message.MessageChunk | None:
        self.msg_idx += 1
@@ -74,10 +63,7 @@ class _StreamAccumulator:
            self.last_role = msg.role
        if msg.content:
-            content = msg.content
+            self.accumulated_content += msg.content
            if self._think_state is not None:
                content = self._think_state.feed(content)
            self.accumulated_content += content
        if msg.tool_calls:
            for tool_call in msg.tool_calls:
@@ -93,14 +79,11 @@ class _StreamAccumulator:
                if tool_call.function and tool_call.function.arguments:
                    self.tool_calls_map[tool_call.id].function.arguments += tool_call.function.arguments
        if msg.is_final:
            self._flush_think_state()
        if self.msg_idx % 8 == 0 or msg.is_final:
            self.msg_sequence += 1
            return provider_message.MessageChunk(
                role=self.last_role,
-                content=self._maybe_strip_think(self.accumulated_content),
+                content=self.accumulated_content,
                tool_calls=list(self.tool_calls_map.values()) if (self.tool_calls_map and msg.is_final) else None,
                is_final=msg.is_final,
                msg_sequence=self.msg_sequence,
@@ -109,29 +92,13 @@ class _StreamAccumulator:
        return None
    def final_message(self) -> provider_message.MessageChunk:
        self._flush_think_state()
        return provider_message.MessageChunk(
            role=self.last_role,
-            content=self._maybe_strip_think(self.accumulated_content),
+            content=self.accumulated_content,
            tool_calls=list(self.tool_calls_map.values()) if self.tool_calls_map else None,
            msg_sequence=self.msg_sequence,
        )
    def _maybe_strip_think(self, content: str) -> str:
        if not self.remove_think or not content:
            return content
        from ..modelmgr.requesters.litellmchat import LiteLLMRequester
        return LiteLLMRequester._strip_think(content)
    def _flush_think_state(self) -> None:
        if self._think_state is None:
            return
        pending = self._think_state.flush()
        if pending:
            self.accumulated_content += pending
@runner.runner_class('local-agent')
 class LocalAgentRunner(runner.RequestRunner):
@@ -481,7 +448,7 @@ class LocalAgentRunner(runner.RequestRunner):
        except AttributeError:
            is_stream = False
-        remove_think = ((query.pipeline_config.get('output') or {}).get('misc') or {}).get('remove-think', False)
+        remove_think = query.pipeline_config['output'].get('misc', '').get('remove-think')
        # Build ordered candidate list (primary + fallbacks)
        candidates = await self._get_model_candidates(query)
@@ -505,7 +472,7 @@ class LocalAgentRunner(runner.RequestRunner):
            final_msg = msg
        else:
            # Streaming: invoke with fallback
-            stream_accumulator = _StreamAccumulator(msg_sequence=1, remove_think=remove_think)
+            stream_accumulator = _StreamAccumulator(msg_sequence=1)
            stream_src, use_llm_model = await self._invoke_stream_with_fallback(
                query,
@@ -609,7 +576,6 @@ class LocalAgentRunner(runner.RequestRunner):
                stream_accumulator = _StreamAccumulator(
                    msg_sequence=first_end_sequence,
                    initial_content=first_content,
                    remove_think=remove_think,
                )
                tool_stream_src = use_llm_model.provider.invoke_llm_stream(
@@ -196,16 +196,6 @@ class _TransportReconnect(Exception):
    """
 class _CallerReconnect(Exception):
    """Internal signal: a tool/resource call hit a server-expired session
    (e.g. a UDP/HTTP MCP server's own session timeout) and asked the owning
    lifecycle loop to rebuild the connection.
    Like _TransportReconnect, this does NOT consume the fatal retry budget —
    the server-side timeout is expected, recurring behavior, not a failure.
    """
 class RuntimeMCPSession:
    """运行时 MCP 会话"""
@@ -275,12 +265,6 @@ class RuntimeMCPSession:
        self._lifecycle_task = None
        self._shutdown_event = asyncio.Event()
        self._ready_event = asyncio.Event()
        # Signaled by callers (invoke_mcp_tool / read_resource_envelope) when
        # they see a server-expired session; the lifecycle loop reconnects and
        # sets _reconnected_event so all callers waiting on this cycle resume
        # together, instead of each racing to rebuild the session itself.
        self._reconnect_event = asyncio.Event()
        self._reconnected_event: asyncio.Event | None = None
        # Set transiently when a WS transport drop should NOT stop the managed
        # process (it will be re-attached on the next initialize()).
        self._preserve_managed_process = False
@@ -445,19 +429,12 @@ class RuntimeMCPSession:
            if self._uses_box_stdio():
                monitor_task = asyncio.create_task(self._box_stdio_runtime.monitor_process_health())
                shutdown_task = asyncio.create_task(self._shutdown_event.wait())
                reconnect_task = asyncio.create_task(self._reconnect_event.wait())
                done, pending = await asyncio.wait(
-                    [shutdown_task, monitor_task, reconnect_task],
+                    [shutdown_task, monitor_task],
                    return_when=asyncio.FIRST_COMPLETED,
                )
                for task in pending:
                    task.cancel()
                if reconnect_task in done and not self._shutdown_event.is_set():
                    self._reconnect_event.clear()
                    self.ap.logger.info(
                        f'MCP session {self.server_name}: caller requested reconnect (server session expired)'
                    )
                    raise _CallerReconnect('Caller requested reconnect after session expiry')
                for task in done:
                    if task is monitor_task and not self._shutdown_event.is_set():
                        # The monitor completed. This is EITHER the managed
@@ -485,20 +462,7 @@ class RuntimeMCPSession:
                        self.error_phase = MCPSessionErrorPhase.RUNTIME
                        raise Exception('Box managed process exited unexpectedly')
            else:
-                shutdown_task = asyncio.create_task(self._shutdown_event.wait())
+                await self._shutdown_event.wait()
                reconnect_task = asyncio.create_task(self._reconnect_event.wait())
                done, pending = await asyncio.wait(
                    [shutdown_task, reconnect_task],
                    return_when=asyncio.FIRST_COMPLETED,
                )
                for task in pending:
                    task.cancel()
                if reconnect_task in done and not self._shutdown_event.is_set():
                    self._reconnect_event.clear()
                    self.ap.logger.info(
                        f'MCP session {self.server_name}: caller requested reconnect (server session expired)'
                    )
                    raise _CallerReconnect('Caller requested reconnect after session expiry')
        except _ColdStartRetry:
            # Cold-start in progress: set the preserve flag BEFORE the finally
@@ -557,43 +521,6 @@ class RuntimeMCPSession:
                self.error_phase = None
                await asyncio.sleep(1)
                continue
            except _CallerReconnect:
                # A tool/resource call hit a server-expired session and asked us
                # to rebuild the connection. Same treatment as _TransportReconnect:
                # reconnect immediately WITHOUT consuming the fatal retry budget.
                # Wake any callers waiting in _trigger_reconnect() regardless of
                # outcome so they don't block for the full timeout.
                reconnected_event = self._reconnected_event
                if self._shutdown_event.is_set():
                    if reconnected_event is not None:
                        reconnected_event.set()
                    return
                self.status = MCPSessionStatus.CONNECTING
                self.error_message = None
                self.error_phase = None
                try:
                    if self.server_config['mode'] == 'stdio':
                        await self._init_stdio_python_server()
                    elif self.server_config['mode'] == 'remote':
                        await self._init_remote_server()
                    elif self.server_config['mode'] == 'sse':
                        await self._init_sse_server()
                    elif self.server_config['mode'] == 'http':
                        await self._init_streamable_http_server()
                    await self.refresh()
                    self.status = MCPSessionStatus.CONNECTED
                    self.ap.logger.info(f'MCP session {self.server_name} reconnected successfully after session expiry')
                except Exception as reconnect_err:
                    self.status = MCPSessionStatus.ERROR
                    self.error_message = str(reconnect_err)
                    self.ap.logger.error(
                        f'MCP session {self.server_name}: reconnect after session expiry failed: '
                        f'{self._describe_exception(reconnect_err)}'
                    )
                finally:
                    if reconnected_event is not None:
                        reconnected_event.set()
                continue
            except _ColdStartRetry as e:
                # The managed process is alive but still cold-starting (e.g.
                # `npx -y <pkg>` is still installing) and cannot yet answer the
@@ -696,47 +623,6 @@ class RuntimeMCPSession:
                return True
        return False
    @staticmethod
    def _is_session_terminated(exc: BaseException) -> bool:
        """Whether exc indicates the server-side session expired.
        Long-lived MCP servers (notably UDP/HTTP transports) commonly enforce
        their own session timeout (e.g. ~30 minutes); once it fires, any call
        on the cached session raises this rather than a transport-level error.
        """
        for leaf in RuntimeMCPSession._iter_exception_leaves(exc):
            msg = str(leaf).lower()
            if 'session terminated' in msg or 'session expired' in msg:
                return True
        return False
    _RECONNECT_WAIT_TIMEOUT = 30.0
    async def _trigger_reconnect(self) -> bool:
        """Ask the owning lifecycle loop to rebuild the session and wait for it.
        Concurrent callers that hit the timeout at the same time all signal
        the same _reconnect_event and await the same _reconnected_event, so
        the lifecycle loop reconnects once and every caller resumes together
        — instead of each caller racing to rebuild the session itself.
        Returns True if reconnection succeeded within the timeout.
        """
        if self._shutdown_event.is_set():
            return False
        if self._reconnected_event is None or self._reconnected_event.is_set():
            self._reconnected_event = asyncio.Event()
        reconnected_event = self._reconnected_event
        self._reconnect_event.set()
        try:
            await asyncio.wait_for(reconnected_event.wait(), timeout=self._RECONNECT_WAIT_TIMEOUT)
            return self.status == MCPSessionStatus.CONNECTED
        except asyncio.TimeoutError:
            self.ap.logger.warning(f'MCP session {self.server_name} reconnect timed out')
            return False
    _MONITOR_POLL_INTERVAL = 5
    _MONITOR_MAX_CONSECUTIVE_ERRORS = 3
@@ -945,34 +831,21 @@ class RuntimeMCPSession:
        arguments: dict,
        query: pipeline_query.Query | None = None,
    ) -> list[provider_message.ContentElement]:
-        for attempt in range(2):
+        if not self.session:
-            if not self.session:
+            raise Exception('MCP session is not connected')
                raise Exception('MCP session is not connected')
-            try:
+        result = await self.session.call_tool(tool_name, arguments)
-                result = await self.session.call_tool(tool_name, arguments)
+        if result.isError:
-            except Exception as e:
+            error_texts = []
                if attempt == 0 and self._is_session_terminated(e):
                    self.ap.logger.warning(
                        f'MCP tool {tool_name} on {self.server_name} got session terminated, triggering reconnect...'
                    )
                    if await self._trigger_reconnect():
                        continue
                raise
            if result.isError:
                error_texts = []
                for content in result.content:
                    if getattr(content, 'type', '') == 'text':
                        error_texts.append(content.text)
                raise Exception('\n'.join(error_texts) if error_texts else 'Unknown error from MCP tool')
            result_contents: list[provider_message.ContentElement] = []
            for content in result.content:
-                result_contents.extend(self._content_to_provider_elements(content, query=query, source_tool=tool_name))
+                if getattr(content, 'type', '') == 'text':
-            return result_contents
+                    error_texts.append(content.text)
            raise Exception('\n'.join(error_texts) if error_texts else 'Unknown error from MCP tool')
-        raise Exception('MCP session is not connected')
+        result_contents: list[provider_message.ContentElement] = []
        for content in result.content:
            result_contents.extend(self._content_to_provider_elements(content, query=query, source_tool=tool_name))
        return result_contents
    def get_tools(self) -> list[resource_tool.LLMTool]:
        return self.functions
@@ -1037,21 +910,7 @@ class RuntimeMCPSession:
            self._record_resource_read_trace(query, envelope)
            return envelope
-        result = None
+        result = await self.session.read_resource(AnyUrl(uri))
        for attempt in range(2):
            if not self.session:
                raise Exception('MCP session is not connected')
            try:
                result = await self.session.read_resource(AnyUrl(uri))
                break
            except Exception as e:
                if attempt == 0 and self._is_session_terminated(e):
                    self.ap.logger.warning(
                        f'MCP resource read on {self.server_name} got session terminated, triggering reconnect...'
                    )
                    if await self._trigger_reconnect():
                        continue
                raise
        contents: list[dict] = []
        total_bytes = 0
        truncated_any = False
@@ -119,8 +119,7 @@ class ValkeySearchVectorDatabase(VectorDatabase):
    def __init__(self, ap: app.Application):
        if not VALKEY_SEARCH_AVAILABLE:
            raise ImportError(
-                'valkey-glide is not installed or is unavailable on this platform. '
+                "valkey-glide is not installed. Install it with: pip install 'valkey-glide>=2.4.1,<3.0.0'"
                "On Linux or macOS, install it with: pip install 'valkey-glide>=2.4.1,<3.0.0'"
            )
        self.ap = ap
@@ -303,60 +303,11 @@
    '<svg viewBox="0 0 24 24"><path d="M14 2H6a2 2 0 00-2 2v16a2 2 0 002 2h12a2 2 0 002-2V8l-6-6zM6 20V4h7v5h5v11H6z"/><path d="M8 17l2.5-3.5L13 17l2-2.5L18 17H8z"/></svg>';
  // ========== State ==========
  function createSessionId() {
    if (window.crypto && typeof window.crypto.randomUUID === "function") {
      return window.crypto.randomUUID();
    }
    var bytes = new Uint8Array(16);
    if (!window.crypto || typeof window.crypto.getRandomValues !== "function") {
      throw new Error("Secure random number generation is unavailable");
    }
    window.crypto.getRandomValues(bytes);
    bytes[6] = (bytes[6] & 15) | 64;
    bytes[8] = (bytes[8] & 63) | 128;
    var hex = Array.prototype.map
      .call(bytes, function (value) {
        return value.toString(16).padStart(2, "0");
      })
      .join("");
    return (
      hex.slice(0, 8) +
      "-" +
      hex.slice(8, 12) +
      "-" +
      hex.slice(12, 16) +
      "-" +
      hex.slice(16, 20) +
      "-" +
      hex.slice(20)
    );
  }
  function getOrCreateSessionId() {
    var storageKey = "langbot_embed_session_" + CONFIG.botUuid;
    try {
      var stored = window.sessionStorage.getItem(storageKey);
      if (
        /^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$/.test(
          stored || "",
        )
      )
        return stored;
      var created = createSessionId();
      window.sessionStorage.setItem(storageKey, created);
      return created;
    } catch (e) {
      return createSessionId();
    }
  }
  var state = {
    isOpen: false,
    isConnected: false,
    ws: null,
    connectionId: null,
    sessionId: getOrCreateSessionId(),
    reconnectAttempts: 0,
    heartbeatTimer: null,
    messages: [],
@@ -364,9 +315,6 @@
    isStreaming: false,
    streamingMsgId: null,
    historyLoaded: false,
    hasConnected: false,
    messageVersion: 0,
    historyReloadTimer: null,
    pendingImage: null,
    feedbackState: {},
  };
@@ -525,9 +473,7 @@
      "/api/v1/embed/" +
      CONFIG.botUuid +
      "/ws/connect?session_type=" +
-      CONFIG.sessionType +
+      CONFIG.sessionType;
      "&session_id=" +
      encodeURIComponent(state.sessionId);
    try {
      state.ws = new WebSocket(url);
@@ -574,8 +520,6 @@
      case "connected":
        state.isConnected = true;
        state.connectionId = data.connection_id;
        if (state.hasConnected) loadHistory(true);
        state.hasConnected = true;
        updateStatusDot();
        updateSendBtn();
        break;
@@ -643,7 +587,6 @@
    if (existingIdx >= 0) {
      state.messages[existingIdx] = msg;
      state.messageVersion++;
      updateMessageEl(existingIdx, msg);
    } else {
      addMessage(msg);
@@ -713,27 +656,16 @@
  }
  // ========== Message History ==========
-  function scheduleHistoryReload() {
+  function loadHistory() {
-    if (state.historyReloadTimer) clearTimeout(state.historyReloadTimer);
+    if (state.historyLoaded) return;
    state.historyReloadTimer = setTimeout(function () {
      state.historyReloadTimer = null;
      loadHistory(true);
    }, 100);
  }
  function loadHistory(force) {
    if (state.historyLoaded && !force) return;
    state.historyLoaded = true;
    var messageVersion = state.messageVersion;
    var url =
      CONFIG.baseUrl +
      "/api/v1/embed/" +
      CONFIG.botUuid +
      "/messages/" +
-      CONFIG.sessionType +
+      CONFIG.sessionType;
      "?session_id=" +
      encodeURIComponent(state.sessionId);
    var headers = {};
    if (state.sessionToken)
      headers["Authorization"] = "Bearer " + state.sessionToken;
@@ -743,16 +675,6 @@
      })
      .then(function (json) {
        if (json.code === 0 && json.data && json.data.messages) {
          if (force && messageVersion !== state.messageVersion) {
            scheduleHistoryReload();
            return;
          }
          if (force) {
            state.messages = [];
            state.isStreaming = false;
            state.streamingMsgId = null;
            renderMessages();
          }
          var msgs = json.data.messages;
          for (var i = 0; i < msgs.length; i++) {
            addMessage(msgs[i], true);
@@ -771,16 +693,13 @@
      "/api/v1/embed/" +
      CONFIG.botUuid +
      "/reset/" +
-      CONFIG.sessionType +
+      CONFIG.sessionType;
      "?session_id=" +
      encodeURIComponent(state.sessionId);
    var headers = {};
    if (state.sessionToken)
      headers["Authorization"] = "Bearer " + state.sessionToken;
    fetch(url, { method: "POST", headers: headers })
      .then(function () {
        state.messages = [];
        state.messageVersion++;
        state.isStreaming = false;
        state.streamingMsgId = null;
        state.historyLoaded = true;
@@ -794,7 +713,6 @@
  // ========== UI Rendering ==========
  function addMessage(msg, silent) {
    state.messages.push(msg);
    if (!silent) state.messageVersion++;
    var el = createMessageEl(msg);
    if (els.welcome) {
      els.welcome.style.display = "none";
@@ -15,7 +15,6 @@ from tests.factories import FakeApp
 pytestmark = pytest.mark.integration
 SESSION_ID = '31c0f2e9-b115-4ee6-8f15-3e624d6456b1'
@pytest.fixture(scope='module')
@@ -192,10 +191,10 @@ class TestEmbedMessagesEndpoint:
    """Tests for messages endpoint."""
    @pytest.mark.asyncio
-    async def test_get_messages_person_success(self, quart_test_client, fake_embed_app):
+    async def test_get_messages_person_success(self, quart_test_client):
        """GET messages/person returns messages."""
        response = await quart_test_client.get(
-            f'/api/v1/embed/a1b2c3d4-5678-90ab-cdef-123456789abc/messages/person?session_id={SESSION_ID}',
+            '/api/v1/embed/a1b2c3d4-5678-90ab-cdef-123456789abc/messages/person',
            headers={'Authorization': 'Bearer 1234567890.dummy'},
        )
@@ -203,30 +202,17 @@ class TestEmbedMessagesEndpoint:
        data = await response.get_json()
        assert data['code'] == 0
        assert 'messages' in data['data']
        fake_embed_app.platform_mgr.websocket_proxy_bot.adapter.get_websocket_messages.assert_called_with(
            'test-pipeline-uuid', 'person', SESSION_ID
        )
    @pytest.mark.asyncio
    async def test_get_messages_group_success(self, quart_test_client):
        """GET messages/group returns messages."""
        response = await quart_test_client.get(
-            f'/api/v1/embed/a1b2c3d4-5678-90ab-cdef-123456789abc/messages/group?session_id={SESSION_ID}',
+            '/api/v1/embed/a1b2c3d4-5678-90ab-cdef-123456789abc/messages/group',
            headers={'Authorization': 'Bearer 1234567890.dummy'},
        )
        assert response.status_code == 200
    @pytest.mark.asyncio
    async def test_get_messages_requires_session_id(self, quart_test_client):
        """GET messages without a client session identifier returns 400."""
        response = await quart_test_client.get(
            '/api/v1/embed/a1b2c3d4-5678-90ab-cdef-123456789abc/messages/person',
            headers={'Authorization': 'Bearer 1234567890.dummy'},
        )
        assert response.status_code == 400
    @pytest.mark.asyncio
    async def test_get_messages_invalid_session_type(self, quart_test_client):
        """GET messages with invalid session_type returns 400."""
@@ -243,29 +229,16 @@ class TestEmbedResetEndpoint:
    """Tests for session reset endpoint."""
    @pytest.mark.asyncio
-    async def test_reset_session_person_success(self, quart_test_client, fake_embed_app):
+    async def test_reset_session_person_success(self, quart_test_client):
        """POST reset/person resets session."""
        response = await quart_test_client.post(
-            f'/api/v1/embed/a1b2c3d4-5678-90ab-cdef-123456789abc/reset/person?session_id={SESSION_ID}',
+            '/api/v1/embed/a1b2c3d4-5678-90ab-cdef-123456789abc/reset/person',
            headers={'Authorization': 'Bearer 1234567890.dummy'},
        )
        assert response.status_code == 200
        data = await response.get_json()
        assert data['code'] == 0
        fake_embed_app.platform_mgr.websocket_proxy_bot.adapter.reset_session.assert_called_with(
            'test-pipeline-uuid', 'person', SESSION_ID
        )
    @pytest.mark.asyncio
    async def test_reset_session_requires_session_id(self, quart_test_client):
        """POST reset without a client session identifier returns 400."""
        response = await quart_test_client.post(
            '/api/v1/embed/a1b2c3d4-5678-90ab-cdef-123456789abc/reset/person',
            headers={'Authorization': 'Bearer 1234567890.dummy'},
        )
        assert response.status_code == 400
    @pytest.mark.asyncio
    async def test_reset_session_invalid_uuid(self, quart_test_client):
@@ -77,12 +77,11 @@ async def backend(valkey_config):
        ValkeySearchVectorDatabase,
        VALKEY_SEARCH_AVAILABLE,
    )
    from glide import ft
    if not VALKEY_SEARCH_AVAILABLE:
        pytest.skip('valkey-glide not installed')
    from glide import ft
    ap = _make_ap(valkey_config)
    db = ValkeySearchVectorDatabase(ap)
    client = await db._ensure_client()
@@ -2,7 +2,6 @@ import pytest
 import aiocqhttp
 import langbot_plugin.api.entities.builtin.platform.message as platform_message
 import langbot_plugin.api.entities.builtin.platform.events as platform_events
 from langbot.pkg.platform.sources.aiocqhttp import (
    AiocqhttpAdapter,
    AiocqhttpEventConverter,
@@ -16,72 +15,6 @@ async def _convert_single(component: platform_message.MessageComponent):
    return message[0]
 class _TestLogger:
    def __init__(self):
        self.messages = []
    async def info(self, message):
        self.messages.append(message)
 def _make_adapter():
    logger = _TestLogger()
    adapter = AiocqhttpAdapter.model_construct(
        config={},
        logger=logger,
        bot=aiocqhttp.CQHttp(),
        on_websocket_connection_event_cache=[],
        _listener_wrappers={},
    )
    adapter.bot.on_websocket_connection(adapter._on_websocket_connection)
    return adapter, logger
 def test_connection_listener_is_registered_once_for_multiple_message_listeners():
    adapter, _ = _make_adapter()
    async def callback(event, source_adapter):
        return None
    adapter.register_listener(platform_events.FriendMessage, callback)
    adapter.register_listener(platform_events.GroupMessage, callback)
    adapter.register_listener(platform_events.FeedbackEvent, callback)
    assert len(adapter.bot._bus._subscribers['meta_event.lifecycle.connect']) == 1
@pytest.mark.asyncio
 async def test_connection_listener_only_suppresses_exact_duplicates():
    adapter, logger = _make_adapter()
    first = aiocqhttp.Event({'self_id': 1001, 'time': 10})
    duplicate = aiocqhttp.Event({'self_id': 1001, 'time': 10})
    second = aiocqhttp.Event({'self_id': 2002, 'time': 20})
    await adapter._on_websocket_connection(first)
    await adapter._on_websocket_connection(duplicate)
    await adapter._on_websocket_connection(second)
    assert adapter.on_websocket_connection_event_cache == [first, second]
    assert logger.messages == [
        'WebSocket connection established, bot id: 1001',
        'WebSocket connection established, bot id: 2002',
    ]
 def test_unregister_listener_removes_registered_wrapper():
    adapter, _ = _make_adapter()
    async def callback(event, source_adapter):
        return None
    adapter.register_listener(platform_events.GroupMessage, callback)
    assert len(adapter.bot._bus._subscribers['message.group']) == 1
    adapter.unregister_listener(platform_events.GroupMessage, callback)
    assert not adapter.bot._bus._subscribers['message.group']
@pytest.mark.asyncio
@pytest.mark.parametrize(
    ('payload', 'expected'),
@@ -1,39 +0,0 @@
 import asyncio
 import threading
 import pytest
 from langbot.pkg.platform.sources.lark import NonBlockingLarkWSClient
@pytest.mark.asyncio
 async def test_lark_connection_url_lookup_does_not_block_main_event_loop(monkeypatch):
    client = NonBlockingLarkWSClient('app-id', 'app-secret')
    lookup_started = threading.Event()
    release_lookup = threading.Event()
    base_connect_urls: list[str] = []
    def blocking_get_conn_url() -> str:
        lookup_started.set()
        if not release_lookup.wait(timeout=2):
            raise TimeoutError('test did not release the connection lookup')
        return 'wss://example.invalid/connect?device_id=device&service_id=service'
    async def fake_base_connect(self):
        base_connect_urls.append(self._get_conn_url())
    monkeypatch.setattr(client, '_get_conn_url', blocking_get_conn_url)
    monkeypatch.setattr('lark_oapi.ws.Client._connect', fake_base_connect)
    connect_task = asyncio.create_task(client._connect())
    await asyncio.wait_for(asyncio.to_thread(lookup_started.wait, 1), timeout=1)
    # If the SDK's synchronous requests.post still runs on the event-loop
    # thread, this sleep cannot complete until release_lookup is set.
    await asyncio.wait_for(asyncio.sleep(0.01), timeout=0.1)
    assert not connect_task.done()
    release_lookup.set()
    await asyncio.wait_for(connect_task, timeout=1)
    assert base_connect_urls == ['wss://example.invalid/connect?device_id=device&service_id=service']
@@ -1,267 +0,0 @@
 """Regression tests for isolated embed-widget conversations."""
 import asyncio
 from pathlib import Path
 from unittest.mock import AsyncMock, Mock
 import pytest
 import langbot_plugin.api.entities.builtin.platform.events as platform_events
 from langbot.pkg.platform.sources import websocket_adapter as websocket_adapter_module
 from langbot.pkg.platform.sources.websocket_adapter import WebSocketAdapter, WebSocketMessage, WebSocketSession
 from langbot.pkg.platform.sources.websocket_manager import WebSocketConnectionManager, is_valid_session_id
@pytest.mark.asyncio
 async def test_broadcast_only_reaches_connections_in_same_browser_session():
    manager = WebSocketConnectionManager()
    first = await manager.add_connection(
        websocket=Mock(),
        pipeline_uuid='pipeline-1',
        session_type='person',
        session_id='session-a',
    )
    second = await manager.add_connection(
        websocket=Mock(),
        pipeline_uuid='pipeline-1',
        session_type='person',
        session_id='session-b',
    )
    dashboard = await manager.add_connection(
        websocket=Mock(),
        pipeline_uuid='pipeline-1',
        session_type='person',
    )
    await manager.broadcast_to_pipeline(
        'pipeline-1',
        {'type': 'response'},
        session_type='person',
        session_id='session-a',
    )
    assert await first.send_queue.get() == {'type': 'response'}
    assert second.send_queue.empty()
    assert dashboard.send_queue.empty()
    await manager.broadcast_to_pipeline(
        'pipeline-1',
        {'type': 'dashboard-response'},
        session_type='person',
        session_id=None,
    )
    assert await dashboard.send_queue.get() == {'type': 'dashboard-response'}
    assert first.send_queue.empty()
    assert second.send_queue.empty()
@pytest.mark.asyncio
 async def test_embed_event_uses_stable_session_launcher(monkeypatch):
    manager = WebSocketConnectionManager()
    session_id = '31c0f2e9-b115-4ee6-8f15-3e624d6456b1'
    connection = await manager.add_connection(
        websocket=Mock(),
        pipeline_uuid='pipeline-1',
        session_type='person',
        session_id=session_id,
    )
    monkeypatch.setattr(websocket_adapter_module, 'ws_connection_manager', manager)
    adapter = WebSocketAdapter.model_construct(ap=Mock(), logger=AsyncMock())
    adapter.websocket_person_session = WebSocketSession(id='person')
    adapter.websocket_group_session = WebSocketSession(id='group')
    received = []
    async def listener(event, _callback_adapter):
        received.append(event)
    adapter.listeners = {platform_events.FriendMessage: listener}
    await adapter.handle_websocket_message(
        connection,
        {'message': [{'type': 'Plain', 'text': 'hello'}], 'stream': False},
    )
    await asyncio.sleep(0)
    assert received[0].sender.id == f'websocket_pipeline-1:{session_id}'
@pytest.mark.asyncio
 async def test_embed_group_event_uses_stable_session_launcher(monkeypatch):
    manager = WebSocketConnectionManager()
    session_id = '31c0f2e9-b115-4ee6-8f15-3e624d6456b1'
    connection = await manager.add_connection(
        websocket=Mock(),
        pipeline_uuid='pipeline-1',
        session_type='group',
        session_id=session_id,
    )
    monkeypatch.setattr(websocket_adapter_module, 'ws_connection_manager', manager)
    adapter = WebSocketAdapter.model_construct(ap=Mock(), logger=AsyncMock())
    adapter.websocket_person_session = WebSocketSession(id='person')
    adapter.websocket_group_session = WebSocketSession(id='group')
    received = []
    async def listener(event, _callback_adapter):
        received.append(event)
    adapter.listeners = {platform_events.GroupMessage: listener}
    await adapter.handle_websocket_message(
        connection,
        {'message': [{'type': 'Plain', 'text': 'hello'}], 'stream': False},
    )
    await asyncio.sleep(0)
    assert received[0].sender.id == f'websocket_pipeline-1:{session_id}'
    assert received[0].sender.group.id == f'websocketgroup_pipeline-1:{session_id}'
    dashboard = await manager.add_connection(
        websocket=Mock(),
        pipeline_uuid='pipeline-1',
        session_type='group',
    )
    await adapter.handle_websocket_message(
        dashboard,
        {'stream': False, 'message': [{'type': 'Plain', 'text': 'dashboard'}]},
        None,
    )
    await asyncio.sleep(0)
    assert received[1].sender.id == f'websocket_{dashboard.connection_id}'
    assert received[1].sender.group.id == 'websocketgroup'
@pytest.mark.asyncio
 async def test_stable_session_launcher_resolves_to_active_connection(monkeypatch):
    manager = WebSocketConnectionManager()
    session_id = '31c0f2e9-b115-4ee6-8f15-3e624d6456b1'
    await manager.add_connection(
        websocket=Mock(),
        pipeline_uuid='pipeline-2',
        session_type='person',
        session_id=session_id,
    )
    connection = await manager.add_connection(
        websocket=Mock(),
        pipeline_uuid='pipeline-1',
        session_type='person',
        session_id=session_id,
    )
    monkeypatch.setattr(websocket_adapter_module, 'ws_connection_manager', manager)
    adapter = WebSocketAdapter.model_construct(ap=Mock(), logger=AsyncMock())
    message_source = Mock()
    message_source.sender.id = f'websocket_pipeline-1:{session_id}'
    assert await adapter._get_message_context(message_source) == ('pipeline-1', session_id)
    assert await adapter._get_connection_from_target(f'websocketgroup_pipeline-1:{session_id}') is connection
    assert await manager.get_connection_by_session_id(session_id, 'pipeline-1') is connection
    await manager.remove_connection(connection.connection_id)
    assert await adapter._get_message_context(message_source) == ('pipeline-1', session_id)
    assert await manager.get_connection_by_session_id(session_id, 'pipeline-1') is None
 def test_session_ids_must_be_canonical_random_uuids():
    assert is_valid_session_id('31c0f2e9-b115-4ee6-8f15-3e624d6456b1')
    assert not is_valid_session_id('session-a')
    assert not is_valid_session_id('00000000-0000-0000-0000-000000000000')
 def test_history_read_does_not_allocate_unknown_session():
    adapter = WebSocketAdapter.model_construct(ap=Mock(), logger=AsyncMock())
    adapter.websocket_person_session = WebSocketSession(id='person')
    adapter.websocket_group_session = WebSocketSession(id='group')
    assert adapter.get_websocket_messages('pipeline-1', 'person', 'missing-session') == []
    assert adapter.websocket_person_session.message_lists == {}
 def test_history_and_reset_are_scoped_to_browser_session():
    matching_provider_session = Mock(
        launcher_type=Mock(value='person'),
        launcher_id='websocket_pipeline-1:session-a',
    )
    matching_group_provider_session = Mock(
        launcher_type=Mock(value='group'),
        launcher_id='websocketgroup_pipeline-1:session-a',
    )
    other_session = Mock(
        launcher_type=Mock(value='person'),
        launcher_id='websocket_pipeline-1:session-b',
    )
    ap = Mock()
    ap.sess_mgr.session_list = [
        matching_provider_session,
        matching_group_provider_session,
        other_session,
    ]
    adapter = WebSocketAdapter.model_construct(
        ap=ap,
        logger=AsyncMock(),
    )
    adapter.websocket_person_session = Mock()
    adapter.websocket_group_session = Mock()
    session_a = [
        WebSocketMessage(
            id=1,
            role='user',
            content='private-a',
            message_chain=[],
            timestamp='2026-07-13T00:00:00',
        )
    ]
    session_b = [
        WebSocketMessage(
            id=1,
            role='user',
            content='private-b',
            message_chain=[],
            timestamp='2026-07-13T00:00:00',
        )
    ]
    histories = {
        'pipeline-1:session-a': session_a,
        'pipeline-1:session-b': session_b,
    }
    stream_indexes = {
        'pipeline-1:session-a': {'response-a': 0},
        'pipeline-1:session-b': {'response-b': 0},
    }
    adapter.websocket_person_session.get_message_list.side_effect = histories.__getitem__
    adapter.websocket_person_session.message_lists = histories
    adapter.websocket_person_session.stream_message_indexes = stream_indexes
    adapter.websocket_group_session.message_lists = {}
    adapter.websocket_group_session.stream_message_indexes = {}
    assert adapter.get_websocket_messages('pipeline-1', 'person', 'session-a')[0]['content'] == 'private-a'
    assert adapter.get_websocket_messages('pipeline-1', 'person', 'session-b')[0]['content'] == 'private-b'
    adapter.reset_session('pipeline-1', 'person', 'session-a')
    assert histories['pipeline-1:session-a'] == []
    assert stream_indexes['pipeline-1:session-a'] == {}
    assert histories['pipeline-1:session-b'] == session_b
    assert stream_indexes['pipeline-1:session-b'] == {'response-b': 0}
    assert ap.sess_mgr.session_list == [matching_group_provider_session, other_session]
    adapter.reset_session('pipeline-1', 'group', 'session-a')
    assert ap.sess_mgr.session_list == [other_session]
 def test_widget_sends_stable_session_id_to_all_conversation_endpoints():
    widget_path = Path(__file__).parents[3] / 'src/langbot/templates/embed/widget.js'
    widget = widget_path.read_text(encoding='utf-8')
    assert 'langbot_embed_session_' in widget
    assert 'window.sessionStorage' in widget
    assert 'window.localStorage' not in widget
    assert 'session_id=' in widget
    assert widget.count('encodeURIComponent(state.sessionId)') >= 3
    assert 'loadHistory(true)' in widget
    assert 'messageVersion !== state.messageVersion' in widget
    assert 'scheduleHistoryReload();' in widget
@@ -279,122 +279,6 @@ class TestInvokeLLMStreamUsage:
        assert query.variables['_stream_usage']['total_tokens'] == 12
    @pytest.mark.asyncio
    async def test_stream_removes_leading_think_across_chunks(self):
        """A leading think block split across chunks must be removed."""
        import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
        import langbot_plugin.api.entities.builtin.provider.message as provider_message
        mock_ap = Mock()
        mock_ap.tool_mgr = Mock()
        mock_ap.tool_mgr.generate_tools_for_openai = AsyncMock(return_value=None)
        requester = litellmchat.LiteLLMRequester(ap=mock_ap, config={})
        model = MockRuntimeModel('minimax-m3', 'test-api-key')
        chunks = [
            self._make_chunk(content='<thi'),
            self._make_chunk(content='nk>hidden'),
            self._make_chunk(content=' reasoning</thi'),
            self._make_chunk(content='nk>Visible answer', finish_reason='stop'),
        ]
        async def _aiter(*args, **kwargs):
            for c in chunks:
                yield c
        query = Mock(spec=pipeline_query.Query)
        query.variables = {}
        messages = [provider_message.Message(role='user', content='Hi')]
        with patch.object(litellmchat, 'acompletion', new=AsyncMock(side_effect=lambda **kw: _aiter())):
            collected = [
                chunk
                async for chunk in requester.invoke_llm_stream(
                    query=query,
                    model=model,
                    messages=messages,
                    remove_think=True,
                )
            ]
        assert ''.join(chunk.content or '' for chunk in collected) == 'Visible answer'
    @pytest.mark.asyncio
    async def test_stream_removes_initial_orphan_think_close(self):
        """Initial reasoning content without an open tag is removed until </think>."""
        import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
        import langbot_plugin.api.entities.builtin.provider.message as provider_message
        mock_ap = Mock()
        mock_ap.tool_mgr = Mock()
        mock_ap.tool_mgr.generate_tools_for_openai = AsyncMock(return_value=None)
        requester = litellmchat.LiteLLMRequester(ap=mock_ap, config={})
        model = MockRuntimeModel('minimax-m3', 'test-api-key')
        chunks = [
            self._make_chunk(content='hidden reasoning'),
            self._make_chunk(content=' still hidden</thi'),
            self._make_chunk(content='nk>Visible answer', finish_reason='stop'),
        ]
        async def _aiter(*args, **kwargs):
            for c in chunks:
                yield c
        query = Mock(spec=pipeline_query.Query)
        query.variables = {}
        messages = [provider_message.Message(role='user', content='Hi')]
        with patch.object(litellmchat, 'acompletion', new=AsyncMock(side_effect=lambda **kw: _aiter())):
            collected = [
                chunk
                async for chunk in requester.invoke_llm_stream(
                    query=query,
                    model=model,
                    messages=messages,
                    remove_think=True,
                )
            ]
        assert ''.join(chunk.content or '' for chunk in collected) == 'Visible answer'
    @pytest.mark.asyncio
    async def test_stream_removes_non_leading_think_content(self):
        """A think block in the answer body is removed with its content."""
        import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
        import langbot_plugin.api.entities.builtin.provider.message as provider_message
        mock_ap = Mock()
        mock_ap.tool_mgr = Mock()
        mock_ap.tool_mgr.generate_tools_for_openai = AsyncMock(return_value=None)
        requester = litellmchat.LiteLLMRequester(ap=mock_ap, config={})
        model = MockRuntimeModel('gpt-4o', 'test-api-key')
        chunks = [
            self._make_chunk(content='Use <think>x</think> as an XML-like example.', finish_reason='stop'),
        ]
        async def _aiter(*args, **kwargs):
            for c in chunks:
                yield c
        query = Mock(spec=pipeline_query.Query)
        query.variables = {}
        messages = [provider_message.Message(role='user', content='Hi')]
        with patch.object(litellmchat, 'acompletion', new=AsyncMock(side_effect=lambda **kw: _aiter())):
            collected = [
                chunk
                async for chunk in requester.invoke_llm_stream(
                    query=query,
                    model=model,
                    messages=messages,
                    remove_think=True,
                )
            ]
        assert ''.join(chunk.content or '' for chunk in collected) == 'Use  as an XML-like example.'
    @pytest.mark.asyncio
    async def test_stream_tool_call_delta_missing_id_and_name(self):
        """LiteLLM may stream tool-call argument deltas with id/name set to None."""
@@ -598,38 +482,6 @@ class TestProcessThinkingContent:
        result = requester._process_thinking_content(content, None, remove_think=True)
        assert result == 'The answer is 42.'
    def test_remove_leading_think_tag(self):
        """Test removing a leading <think> block when remove_think=True"""
        requester = litellmchat.LiteLLMRequester(ap=Mock(), config={})
        content = '<think>Let me think...</think> The answer is 42.'
        result = requester._process_thinking_content(content, None, remove_think=True)
        assert result == 'The answer is 42.'
    def test_remove_non_leading_think_tag(self):
        """Test removing <think> and its content in the answer body"""
        requester = litellmchat.LiteLLMRequester(ap=Mock(), config={})
        content = 'Use <think>example</think> in the document.'
        result = requester._process_thinking_content(content, None, remove_think=True)
        assert result == 'Use  in the document.'
    def test_remove_initial_orphan_think_close(self):
        """Test removing leading reasoning content when only </think> is visible"""
        requester = litellmchat.LiteLLMRequester(ap=Mock(), config={})
        content = 'hidden reasoning</think> Visible answer.'
        result = requester._process_thinking_content(content, None, remove_think=True)
        assert result == 'Visible answer.'
    def test_remove_multiple_think_tags(self):
        """Test removing multiple <think> blocks"""
        requester = litellmchat.LiteLLMRequester(ap=Mock(), config={})
        content = '<think>hidden</think> Keep <think>example</think>.'
        result = requester._process_thinking_content(content, None, remove_think=True)
        assert result == 'Keep .'
    def test_preserve_thinking_markers(self):
        """Test preserving thinking markers when remove_think=False"""
        requester = litellmchat.LiteLLMRequester(ap=Mock(), config={})
@@ -639,20 +491,6 @@ class TestProcessThinkingContent:
        assert 'CRETIRE_REASONING_BEGINk' in result
        assert 'The answer is 42.' in result
    def test_preserve_reasoning_content_when_remove_think_false(self):
        """Test showing separate reasoning_content when remove_think=False"""
        requester = litellmchat.LiteLLMRequester(ap=Mock(), config={})
        result = requester._process_thinking_content('The answer is 42.', 'Let me think...', remove_think=False)
        assert result == '<think>\nLet me think...\n</think>\nThe answer is 42.'
    def test_hide_reasoning_content_when_remove_think_true(self):
        """Test hiding separate reasoning_content when remove_think=True"""
        requester = litellmchat.LiteLLMRequester(ap=Mock(), config={})
        result = requester._process_thinking_content('The answer is 42.', 'Let me think...', remove_think=True)
        assert result == 'The answer is 42.'
    def test_empty_content(self):
        """Test empty content"""
        requester = litellmchat.LiteLLMRequester(ap=Mock(), config={})
@@ -163,51 +163,6 @@ def test_stream_accumulator_merges_fragmented_tool_call_arguments():
    assert final_msg.tool_calls[0].function.arguments == '{"command":"pwd"}'
 def test_stream_accumulator_strips_leading_think_from_tool_round_content():
    accumulator = _StreamAccumulator(
        msg_sequence=3,
        initial_content='I will search for LangBot.',
        remove_think=True,
    )
    assert accumulator.add(provider_message.MessageChunk(role='assistant', content='<thi')) is None
    assert accumulator.add(provider_message.MessageChunk(role='assistant', content='nk>hidden')) is None
    emitted = accumulator.add(
        provider_message.MessageChunk(
            role='assistant',
            content=' reasoning</think>Here is the answer.',
            is_final=True,
        )
    )
    assert emitted is not None
    assert emitted.content == 'I will search for LangBot.Here is the answer.'
    assert '<think>' not in emitted.content
    assert 'hidden reasoning' not in emitted.content
 def test_stream_accumulator_strips_initial_orphan_think_close_from_tool_round_content():
    accumulator = _StreamAccumulator(
        msg_sequence=3,
        initial_content='I will search for LangBot.',
        remove_think=True,
    )
    assert accumulator.add(provider_message.MessageChunk(role='assistant', content='hidden reasoning')) is None
    emitted = accumulator.add(
        provider_message.MessageChunk(
            role='assistant',
            content=' still hidden</think>Here is the answer.',
            is_final=True,
        )
    )
    assert emitted is not None
    assert emitted.content == 'I will search for LangBot.Here is the answer.'
    assert '</think>' not in emitted.content
    assert 'hidden reasoning' not in emitted.content
@pytest.mark.asyncio
 async def test_localagent_uses_exec_for_exact_calculation():
    provider = RecordingProvider()
@@ -357,13 +357,10 @@ class TestCredentialsBuild:
            cred_calls.append(kwargs)
            return ('CRED', kwargs)
-        # These names are absent when the optional valkey-glide dependency is
+        monkeypatch.setattr(mod, 'GlideClient', _FakeClient)
-        # unavailable (for example, on Windows), so allow the test doubles to
+        monkeypatch.setattr(mod, 'ServerCredentials', _fake_credentials)
-        # create them on the module.
+        monkeypatch.setattr(mod, 'GlideClientConfiguration', lambda **kw: kw)
-        monkeypatch.setattr(mod, 'GlideClient', _FakeClient, raising=False)
+        monkeypatch.setattr(mod, 'NodeAddress', lambda *a, **k: ('node', a, k))
        monkeypatch.setattr(mod, 'ServerCredentials', _fake_credentials, raising=False)
        monkeypatch.setattr(mod, 'GlideClientConfiguration', lambda **kw: kw, raising=False)
        monkeypatch.setattr(mod, 'NodeAddress', lambda *a, **k: ('node', a, k), raising=False)
        return backend, created, cred_calls, warnings
    async def test_username_without_password_fails_closed(self, monkeypatch):
@@ -2008,7 +2008,7 @@ wheels = [
 [[package]]
 name = "langbot"
-version = "4.10.6"
+version = "4.10.5"
 source = { editable = "." }
 dependencies = [
    { name = "aiocqhttp" },
@@ -2084,7 +2084,7 @@ dependencies = [
    { name = "tiktoken" },
    { name = "urllib3" },
    { name = "uv" },
-    { name = "valkey-glide", marker = "sys_platform != 'win32'" },
+    { name = "valkey-glide" },
    { name = "websockets" },
 ]
@@ -2173,7 +2173,7 @@ requires-dist = [
    { name = "tiktoken", specifier = ">=0.9.0" },
    { name = "urllib3", specifier = ">=2.7.0" },
    { name = "uv", specifier = ">=0.11.15" },
-    { name = "valkey-glide", marker = "sys_platform != 'win32'", specifier = ">=2.4.1,<3.0.0" },
+    { name = "valkey-glide", specifier = ">=2.4.1,<3.0.0" },
    { name = "websockets", specifier = ">=15.0.1" },
 ]
@@ -5991,9 +5991,9 @@ name = "valkey-glide"
 version = "2.4.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "anyio", marker = "sys_platform != 'win32'" },
+    { name = "anyio" },
-    { name = "protobuf", marker = "sys_platform != 'win32'" },
+    { name = "protobuf" },
-    { name = "sniffio", marker = "sys_platform != 'win32'" },
+    { name = "sniffio" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/72/a2/582b34c6acc8dc857c537f6007459cba48dfa0dc404789a657e5c1a998c0/valkey_glide-2.4.1.tar.gz", hash = "sha256:f1155d84156d11b90488aa67e90102f0bf98a45314f5b99308ac9074c05f7241", size = 898030, upload-time = "2026-05-28T21:41:55.881Z" }
 wheels = [
Author	SHA1	Message	Date
fdc310	1bdf371b5d	feat: Add pipeline_uuid handling across multiple adapters and update related tests	2026-07-12 19:40:59 +08:00
fdc310	cdfde83e39	feat: Add missing completed input lines handling in DingTalk adapter	2026-07-12 18:53:13 +08:00
fdc310	4a753d98f5	Merge remote-tracking branch 'origin/master' into feat/card_dify_human_input	2026-07-12 18:39:48 +08:00
fdc310	354aadfce0	feat: Enhance select input handling and validation in Dify API runner and Telegram adapter	2026-07-12 00:31:12 +08:00
fdc310	4b6931d294	feat: Enhance error handling and caching for Dify and WeCom interactions	2026-07-11 22:40:08 +08:00
fdc310	5eb2b90ae6	feat: Refactor pending form handling to isolate by bot and pipeline	2026-07-11 20:46:42 +08:00
fdc310	d8c14fc969	feat: Update the human-computer interaction template cards, and optimize the prompt information and content display.	2026-07-11 20:05:05 +08:00
fdc310	6cae954a67	feat: Refactor input handling and feedback messages across multiple adapters	2026-07-11 19:23:42 +08:00
fdc310	183fb87649	Add tests for DingTalk adapter content processing and markdown formatting - Updated the assertion in `test_dingtalk_completed_input_lines_include_text_and_select_values` to remove unnecessary markdown formatting. - Added new tests to verify that `_dingtalk_clean_form_content` maintains the order of prompts and completed values in various scenarios. - Introduced `test_dingtalk_card_markdown_preserves_internal_line_breaks` to ensure internal line breaks are correctly converted to HTML line breaks.	2026-07-11 17:28:31 +08:00
fdc310	56b3c15213	feat(lark): add functions for current input definitions and visible form content handling feat(qqofficial): update fallback text handling for non-streaming scenarios feat(difysvapi): enhance form content processing for interactive fields and actions test: add unit tests for Lark and QQ Official adapter functionalities	2026-07-11 15:31:29 +08:00
fdc310	0cde4c71cd	feat: Enhance Telegram and QQ Official adapters with select field handling and form action processing - Added support for select fields in Telegram adapter, including option extraction and callback handling. - Implemented form action processing for Telegram callbacks, improving user interaction feedback. - Introduced new helper functions for building keyboards and resolving select button actions in QQ Official adapter. - Enhanced DifyServiceAPIRunner to handle cumulative streaming responses and improve error handling during workflow resumes. - Added unit tests for new functionalities in Telegram and QQ Official adapters, ensuring robust behavior for select fields and form actions.	2026-07-11 11:16:59 +08:00
fdc310	98a633c712	Add unit tests for DingTalk, Lark, WeComBot, and Dify service API runners - Implement tests for DingTalk adapter helper functions including form content cleaning, input extraction, and completed input lines. - Create unit tests for Lark adapter helper functions focusing on input extraction and completed input lines. - Add tests for WeComBot template card functionalities, including event extraction and payload building for human input. - Enhance Dify service API runner tests to cover human input forms, including input collection, action handling, and form snapshot extraction.	2026-07-10 18:25:24 +08:00
fdc310	776d48380e	feat(telegram): enhance message handling with group stream deletion and form placeholder detection	2026-06-24 14:14:14 +08:00
fdc310	5e6b3cc550	feat(dingtalk): add download link for human input card template and enhance dynamic form configuration	2026-06-24 11:23:02 +08:00
fdc310	ea4cd80939	feat(qqofficial): add method to identify invisible form placeholder chunks in messages	2026-06-23 16:34:38 +08:00
fdc310	ff7fd2f5e0	feat(qqofficial): synchronize passive-reply counter with inbound message sequence	2026-06-23 16:04:29 +08:00
fdc310	3b3aa93cf8	feat(wecom): add functions for template card action extraction and update, enhance button interaction handling	2026-06-23 15:25:39 +08:00
fdc310	f658f0ede5	feat(wecom): add optional source block to interactive template cards for enhanced branding	2026-06-23 00:02:37 +08:00
fdc310	47d75082e8	fix: add Avatar component to dingtalk_human_input_card.json for enhanced user interaction	2026-06-22 20:16:10 +08:00
fdc310	8849fb172e	feat(dingtalk): add feedback functionality to cards with 👍/👎 buttons - Implemented feedback state management for cards, allowing users to provide feedback via thumbs up/down buttons. - Enhanced card rendering to include feedback buttons when appropriate. - Registered feedback listeners to handle feedback events and update card states accordingly. - Updated the card template to support dynamic button rendering for feedback actions. - Improved error handling and logging for feedback actions and card updates.	2026-06-22 15:48:35 +08:00
fdc310	6818ae2783	feat(dingtalk): update display formatting for card content with HTML line breaks	2026-06-21 17:57:39 +08:00
fdc310	e7f1059194	feat(lark): enhance LarkAdapter to store form content for resume notices	2026-06-21 17:17:59 +08:00
fdc310	537677044e	feat(telegram): add action title feedback for user selections in Telegram messages	2026-06-21 15:12:47 +08:00
fdc310	321891441c	feat(dify): add support for chatflow app type and enhance human input handling	2026-06-21 13:50:50 +08:00
fdc310	7f36103aa4	fix(telegram): correct group chat type check and handle oversized callback data for Telegram actions fix(difysvapi): ensure safe access to remove-think configuration in pipeline settings	2026-06-18 11:53:50 +08:00
fdc310	b1750882ed	feat(discord): implement Discord form view with button interactions for Dify actions	2026-06-17 11:25:59 +08:00
fdc310	b55f073e62	feat(qqofficial): implement one-click QR binding and enhance localization support	2026-06-17 00:31:29 +08:00
fdc310	e6cfee541f	feat(qqofficial): implement Dify human input button interaction handling and markdown keyboard support	2026-06-16 19:37:16 +08:00
fdc310	f10ea82418	feat(wecom): implement Dify human input pause handling with button interaction support	2026-06-16 00:07:11 +08:00
fdc310	a32c4d152f	feat(dingtalk): enhance human input card functionality with streaming support and active turn management - Updated the DingTalk card template to enable streaming mode and multi-update configuration. - Removed the obsolete delete_card method from DingTalkClient to streamline card management. - Enhanced DingTalkAdapter to manage active turn cards and accumulated streaming text, ensuring a seamless user experience during human input prompts. - Modified the create_message_card method to utilize existing active cards for resumed workflows, preventing duplication. - Improved the _paint_form_on_card method to update existing cards with human input prompts and buttons dynamically. - Updated the dingtalk_human_input_card.json template to reflect the new streaming capabilities and configuration options.	2026-06-15 17:45:09 +08:00
fdc310	83b0d26e99	feat(dingtalk): implement human input card support and card action handling - Add a new module `card_callback.py` to handle card action button clicks from DingTalk. - Introduce `DingTalkCardActionHandler` to process card action callbacks and extract parameters. - Update `DingTalkAdapter` to manage card state and handle form input through a single card template. - Add configuration for `human_input_card_template_id` in `dingtalk.yaml` to specify the template for human input. - Create a new card template `dingtalk_human_input_card.json` for rendering human input prompts and buttons.	2026-06-15 15:51:52 +08:00
fdc310	e08b5db625	feat: Add the function for formatting human input text to support adapters without rich UI.	2026-06-10 15:41:18 +08:00
fdc310	d0f65b17ec	feat: Improve TelegramAdapter message handling with enhanced error management and draft message support	2026-06-02 01:14:25 +08:00
fdc310	2b533c4a00	feat: Enhance TelegramAdapter to handle form action buttons and message threading	2026-06-02 00:25:52 +08:00
fdc310	f663d87a60	feat: Enhance Lark and Telegram adapters with new form handling for paused workflows	2026-06-01 23:48:59 +08:00
fdc310	60e5b873ee	feat: Add '_routed_by_rule' variable to form action in Lark and Telegram adapters	2026-05-30 12:22:00 +08:00
fdc310	b96f209b98	feat: Implement workflow form handling for paused workflows - Added module-level storage for pending forms to manage state across sessions. - Introduced functions to set, get, and clear pending forms with expiration handling. - Enhanced DifyServiceAPIRunner to support resuming paused workflows via form actions. - Implemented logic to yield human input requests and display appropriate messages. - Updated workflow submission methods to handle paused states and resume actions. - Ensured proper merging of pending form actions with user inputs for seamless interaction.	2026-05-28 23:32:46 +08:00