fix(wecombot): preserve websocket compatibility with pull enhancements

Author: bjhx2003

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

@@ -11,6 +11,7 @@
 from ..logger import EventLogger
 from langbot.libs.wecom_ai_bot_api.wecombotevent import WecomBotEvent
 from langbot.libs.wecom_ai_bot_api.api import WecomBotClient
+from langbot.libs.wecom_ai_bot_api.ws_client import WecomBotWsClient
 
 
 DEFAULT_PULL_PENDING_PLACEHOLDER = 'AI 正在思考中，请稍候'
@@ -194,12 +195,13 @@ async def target2yiri(event: WecomBotEvent):
 
 
 class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
-    bot: WecomBotClient
+    bot: typing.Union[WecomBotClient, WecomBotWsClient]
     bot_account_id: str
     message_converter: WecomBotMessageConverter = WecomBotMessageConverter()
     event_converter: WecomBotEventConverter = WecomBotEventConverter()
     config: dict
     bot_uuid: str = None
+    _ws_mode: bool = False
 
     @staticmethod
     def _get_int_config(config: dict, key: str, default: int, min_value: int, max_value: int) -> int:
@@ -211,52 +213,60 @@ def _get_int_config(config: dict, key: str, default: int, min_value: int, max_va
         return max(min_value, min(max_value, value))
 
     def __init__(self, config: dict, logger: EventLogger):
-        enable_webhook = config.get('enable-webhook', True)
-        if not enable_webhook:
-            raise Exception('WecomBot websocket mode is not supported in this branch yet. Please enable webhook mode.')
-
-        required_keys = ['Token', 'EncodingAESKey', 'Corpid', 'BotId']
-        missing_keys = [key for key in required_keys if not config.get(key)]
-        if missing_keys:
-            raise Exception(f'WecomBot webhook mode missing config: {missing_keys}')
+        enable_webhook = config.get('enable-webhook', False)
+        normalized_config = dict(config)
 
-        pull_poll_timeout_ms = self._get_int_config(config, 'PullPollTimeoutMs', 500, 50, 2000)
-        pull_stream_max_lifetime_ms = self._get_int_config(config, 'PullStreamMaxLifetimeMs', 300000, 1000, 600000)
-        pending_placeholder_enabled = config.get('PullPendingPlaceholderEnabled', False)
-        pending_placeholder_delay_ms = self._get_int_config(config, 'PullPendingPlaceholderDelayMs', 3000, 0, 10000)
-        pending_placeholder = config.get('PullPendingPlaceholder', DEFAULT_PULL_PENDING_PLACEHOLDER)
+        if not enable_webhook:
+            bot = WecomBotWsClient(
+                bot_id=config['BotId'],
+                secret=config['Secret'],
+                logger=logger,
+                encoding_aes_key=config.get('EncodingAESKey', ''),
+            )
+            ws_mode = True
+        else:
+            required_keys = ['Token', 'EncodingAESKey', 'Corpid']
+            missing_keys = [key for key in required_keys if key not in config or not config[key]]
+            if missing_keys:
+                raise Exception(f'WecomBot webhook mode missing config: {missing_keys}')
+
+            pull_poll_timeout_ms = self._get_int_config(config, 'PullPollTimeoutMs', 500, 50, 2000)
+            pull_stream_max_lifetime_ms = self._get_int_config(config, 'PullStreamMaxLifetimeMs', 300000, 1000, 600000)
+            pending_placeholder_enabled = config.get('PullPendingPlaceholderEnabled', False)
+            pending_placeholder_delay_ms = self._get_int_config(config, 'PullPendingPlaceholderDelayMs', 3000, 0, 10000)
+            pending_placeholder = config.get('PullPendingPlaceholder', DEFAULT_PULL_PENDING_PLACEHOLDER)
+
+            normalized_config['PullPollTimeoutMs'] = pull_poll_timeout_ms
+            normalized_config['PullStreamMaxLifetimeMs'] = pull_stream_max_lifetime_ms
+            normalized_config['PullPendingPlaceholderEnabled'] = pending_placeholder_enabled
+            normalized_config['PullPendingPlaceholderDelayMs'] = pending_placeholder_delay_ms
+            normalized_config['PullPendingPlaceholder'] = pending_placeholder
+
+            effective_placeholder_delay = pending_placeholder_delay_ms / 1000 if pending_placeholder_enabled else 0
+            effective_placeholder = pending_placeholder if pending_placeholder_enabled else ''
+
+            bot = WecomBotClient(
+                Token=config['Token'],
+                EnCodingAESKey=config['EncodingAESKey'],
+                Corpid=config['Corpid'],
+                logger=logger,
+                unified_mode=True,
+                stream_poll_timeout=pull_poll_timeout_ms / 1000,
+                stream_max_lifetime=pull_stream_max_lifetime_ms / 1000,
+                pending_placeholder=effective_placeholder,
+                pending_placeholder_delay=effective_placeholder_delay,
+            )
+            ws_mode = False
 
-        normalized_config = dict(config)
-        normalized_config['enable-webhook'] = True
-        normalized_config['PullPollTimeoutMs'] = pull_poll_timeout_ms
-        normalized_config['PullStreamMaxLifetimeMs'] = pull_stream_max_lifetime_ms
-        normalized_config['PullPendingPlaceholderEnabled'] = pending_placeholder_enabled
-        normalized_config['PullPendingPlaceholderDelayMs'] = pending_placeholder_delay_ms
-        normalized_config['PullPendingPlaceholder'] = pending_placeholder
-
-        # 如果未开启首字等待占位，则将延迟设为0且占位文案设为空
-        effective_placeholder_delay = pending_placeholder_delay_ms / 1000 if pending_placeholder_enabled else 0
-        effective_placeholder = pending_placeholder if pending_placeholder_enabled else ''
-
-        bot = WecomBotClient(
-            Token=config['Token'],
-            EnCodingAESKey=config['EncodingAESKey'],
-            Corpid=config['Corpid'],
-            logger=logger,
-            unified_mode=True,
-            stream_poll_timeout=pull_poll_timeout_ms / 1000,
-            stream_max_lifetime=pull_stream_max_lifetime_ms / 1000,
-            pending_placeholder=effective_placeholder,
-            pending_placeholder_delay=effective_placeholder_delay,
-        )
-        bot_account_id = config['BotId']
+        bot_account_id = config.get('BotId', '')
 
         super().__init__(
             config=normalized_config,
             logger=logger,
             bot=bot,
             bot_account_id=bot_account_id,
         )
+        self._ws_mode = ws_mode
 
     async def reply_message(
         self,
@@ -265,7 +275,15 @@ async def reply_message(
         quote_origin: bool = False,
     ):
         content = await self.message_converter.yiri2target(message)
-        await self.bot.set_message(message_source.source_platform_object.message_id, content)
+        if self._ws_mode:
+            event = message_source.source_platform_object
+            req_id = event.get('req_id', '')
+            if req_id:
+                await self.bot.reply_text(req_id, content)
+            else:
+                await self.bot.set_message(event.message_id, content)
+        else:
+            await self.bot.set_message(message_source.source_platform_object.message_id, content)
 
     async def reply_message_chunk(
         self,
@@ -275,30 +293,22 @@ async def reply_message_chunk(
         quote_origin: bool = False,
         is_final: bool = False,
     ):
-        """将流水线增量输出写入企业微信 stream 会话。
-
-        Args:
-            message_source: 流水线提供的原始消息事件。
-            bot_message: 当前片段对应的模型元信息（未使用）。
-            message: 需要回复的消息链。
-            quote_origin: 是否引用原消息（企业微信暂不支持）。
-            is_final: 标记当前片段是否为最终回复。
-
-        Returns:
-            dict: 包含 `stream` 键，标识写入是否成功。
-
-        Example:
-            在流水线 `reply_message_chunk` 调用中自动触发，无需手动调用。
-        """
-        # 转换为纯文本（智能机器人当前协议仅支持文本流）
         content = await self.message_converter.yiri2target(message)
         msg_id = message_source.source_platform_object.message_id
-        # 将片段推送到 WecomBotClient 中的队列，返回值用于判断是否走降级逻辑
-        success = await self.bot.push_stream_chunk(msg_id, content, is_final=is_final)
-        if not success and is_final:
-            # 未命中流式队列时使用旧有 set_message 兜底
-            await self.bot.set_message(msg_id, content)
-        return {'stream': success}
+
+        if self._ws_mode:
+            success = await self.bot.push_stream_chunk(msg_id, content, is_final=is_final)
+            if not success and is_final:
+                event = message_source.source_platform_object
+                req_id = event.get('req_id', '')
+                if req_id:
+                    await self.bot.reply_text(req_id, content)
+            return {'stream': success}
+        else:
+            success = await self.bot.push_stream_chunk(msg_id, content, is_final=is_final)
+            if not success and is_final:
+                await self.bot.set_message(msg_id, content)
+            return {'stream': success}
 
     async def is_stream_output_supported(self) -> bool:
         """智能机器人侧默认开启流式能力。
@@ -315,7 +325,11 @@ async def create_message_card(self, message_id: str, event) -> bool:
         return False
 
     async def send_message(self, target_type, target_id, message):
-        pass
+        if self._ws_mode:
+            content = await self.message_converter.yiri2target(message)
+            await self.bot.send_message(target_id, content)
+        else:
+            pass
 
     def register_listener(
         self,
@@ -344,29 +358,25 @@ def set_bot_uuid(self, bot_uuid: str):
         self.bot_uuid = bot_uuid
 
     async def handle_unified_webhook(self, bot_uuid: str, path: str, request):
-        """处理统一 webhook 请求。
-
-        Args:
-            bot_uuid: Bot 的 UUID
-            path: 子路径（如果有的话）
-            request: Quart Request 对象
-
-        Returns:
-            响应数据
-        """
+        if self._ws_mode:
+            return None
         return await self.bot.handle_unified_webhook(request)
 
     async def run_async(self):
-        # 统一 webhook 模式下，不启动独立的 Quart 应用
-        # 保持运行但不启动独立端口
+        if self._ws_mode:
+            await self.bot.connect()
+        else:
 
-        async def keep_alive():
-            while True:
-                await asyncio.sleep(1)
+            async def keep_alive():
+                while True:
+                    await asyncio.sleep(1)
 
-        await keep_alive()
+            await keep_alive()
 
     async def kill(self) -> bool:
+        if self._ws_mode:
+            await self.bot.disconnect()
+            return True
         return False
 
     async def unregister_listener(

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

@@ -21,20 +21,20 @@ spec:
     - name: enable-webhook
       label:
         en_US: Enable Webhook Mode
-        zh_Hans: 启用 Webhook 模式
+        zh_Hans: 启用Webhook模式
       description:
-        en_US: Switch between Webhook mode and WebSocket mode config. This branch currently supports Webhook mode only.
-        zh_Hans: 在 Webhook / WebSocket 配置之间切换展示。当前分支仅支持 Webhook 模式。
+        en_US: If enabled, the bot will use webhook mode to receive messages. Otherwise, it will use WS long connection mode
+        zh_Hans: 如果启用，机器人将使用 Webhook 模式接收消息。否则，将使用 WS 长连接模式
       type: boolean
       required: true
-      default: true
+      default: false
     - name: Secret
       label:
         en_US: Secret
         zh_Hans: 机器人密钥 (Secret)
       description:
-        en_US: Reserved for WebSocket mode.
-        zh_Hans: WebSocket 模式预留配置项。
+        en_US: Required for WebSocket long connection mode
+        zh_Hans: 使用 WS 长连接模式时必填
       type: string
       required: false
       default: ""
@@ -43,6 +43,9 @@ spec:
       label:
         en_US: Corpid
         zh_Hans: 企业ID
+      description:
+        en_US: Required for Webhook mode
+        zh_Hans: 使用 Webhook 模式时必填
       type: string
       required: false
       default: ""
@@ -51,6 +54,9 @@ spec:
       label:
         en_US: Token
         zh_Hans: 令牌 (Token)
+      description:
+        en_US: Required for Webhook mode
+        zh_Hans: 使用 Webhook 模式时必填
       type: string
       required: false
       default: ""
@@ -59,6 +65,9 @@ spec:
       label:
         en_US: EncodingAESKey
         zh_Hans: 消息加解密密钥 (EncodingAESKey)
+      description:
+        en_US: Required for Webhook mode. Optional for WebSocket mode (used for file decryption)
+        zh_Hans: 使用 Webhook 模式时必填。WebSocket 模式下可选（用于文件解密）
       type: string
       required: false
       default: ""

tests/unit_tests/pipeline/test_wecombot_dify_minfix.py

@@ -409,8 +409,6 @@ async def test_wecom_dispatch_exception_forces_finish():
     assert chunk.content == client.stream_error_final_text
 
 
-
-
 def test_wecom_client_defaults_match_master_polling_behavior():
     WecomBotClient = get_wecom_client()
     client = WecomBotClient(
@@ -441,18 +439,22 @@ def get_wecom_adapter():
     return import_module('langbot.pkg.platform.sources.wecombot').WecomBotAdapter
 
 
-def test_wecombot_adapter_rejects_websocket_mode_in_current_branch():
+def test_wecombot_adapter_supports_websocket_mode_from_master():
     WecomBotAdapter = get_wecom_adapter()
+    ws_module = import_module('langbot.libs.wecom_ai_bot_api.ws_client')
 
-    with pytest.raises(Exception, match='websocket mode is not supported'):
-        WecomBotAdapter(
-            {
-                'BotId': 'bot-id',
-                'enable-webhook': False,
-                'Secret': 'secret',
-            },
-            make_async_logger(),
-        )
+    adapter = WecomBotAdapter(
+        {
+            'BotId': 'bot-id',
+            'enable-webhook': False,
+            'Secret': 'secret',
+        },
+        make_valid_event_logger(),
+    )
+
+    assert adapter._ws_mode is True
+    assert isinstance(adapter.bot, ws_module.WecomBotWsClient)
+    assert adapter.config['enable-webhook'] is False
 
 
 def test_wecombot_adapter_webhook_mode_normalizes_pull_config():