feat: add WebChat adapter for pipeline debugging (#1510)

* feat: add WebChat adapter for pipeline debugging

- Create WebChatAdapter for handling debug messages in pipeline testing
- Add HTTP API endpoints for debug message sending and retrieval
- Implement frontend debug dialog with session switching (private/group chat)
- Add Chinese i18n translations for debug interface
- Auto-create default WebChat bot during database initialization
- Support fixed session IDs: webchatperson and webchatgroup for testing

Co-Authored-By: Junyan Qin <Chin>, 秦骏言 in Chinese, you can call me my english name Rock Chin. <rockchinq@gmail.com>

* perf: ui for webchat

* feat: complete webchat backend

* feat: core chat apis

* perf: button style in pipeline card

* perf: log btn in bot card

* perf: webchat entities definition

* fix: bugs

* perf: web chat

* perf: dialog styles

* perf: styles

* perf: styles

* fix: group invalid in webchat

* perf: simulate real im message

* perf: group timeout toast

* feat(webchat): add supports for mentioning bot in group

* perf(webchat): at component styles

* perf: at badge display in message

* fix: linter errors

* fix: webchat was listed on adapter list

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Junyan Qin <Chin>, 秦骏言 in Chinese, you can call me my english name Rock Chin. <rockchinq@gmail.com>

pkg/api/http/controller/groups/pipelines/pipelines.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import quart
 
-from .. import group
+from ... import group
 
 
 @group.group_class('pipelines', '/api/v1/pipelines')

pkg/api/http/controller/groups/pipelines/webchat.py

@@ -0,0 +1,79 @@
+import quart
+
+from ... import group
+
+
+@group.group_class('webchat', '/api/v1/pipelines/<pipeline_uuid>/chat')
+class WebChatDebugRouterGroup(group.RouterGroup):
+    async def initialize(self) -> None:
+        @self.route('/send', methods=['POST'])
+        async def send_message(pipeline_uuid: str) -> str:
+            """发送调试消息到流水线"""
+            try:
+                data = await quart.request.get_json()
+                session_type = data.get('session_type', 'person')
+                message_chain_obj = data.get('message', [])
+
+                if not message_chain_obj:
+                    return self.http_status(400, -1, 'message is required')
+
+                if session_type not in ['person', 'group']:
+                    return self.http_status(400, -1, 'session_type must be person or group')
+
+                webchat_adapter = self.ap.platform_mgr.webchat_proxy_bot.adapter
+
+                if not webchat_adapter:
+                    return self.http_status(404, -1, 'WebChat adapter not found')
+
+                result = await webchat_adapter.send_webchat_message(pipeline_uuid, session_type, message_chain_obj)
+
+                return self.success(
+                    data={
+                        'message': result,
+                    }
+                )
+
+            except Exception as e:
+                return self.http_status(500, -1, f'Internal server error: {str(e)}')
+
+        @self.route('/messages/<session_type>', methods=['GET'])
+        async def get_messages(pipeline_uuid: str, session_type: str) -> str:
+            """获取调试消息历史"""
+            try:
+                if session_type not in ['person', 'group']:
+                    return self.http_status(400, -1, 'session_type must be person or group')
+
+                webchat_adapter = self.ap.platform_mgr.webchat_proxy_bot.adapter
+
+                if not webchat_adapter:
+                    return self.http_status(404, -1, 'WebChat adapter not found')
+
+                messages = webchat_adapter.get_webchat_messages(pipeline_uuid, session_type)
+
+                return self.success(data={'messages': messages})
+
+            except Exception as e:
+                return self.http_status(500, -1, f'Internal server error: {str(e)}')
+
+        @self.route('/reset/<session_type>', methods=['POST'])
+        async def reset_session(session_type: str) -> str:
+            """重置调试会话"""
+            try:
+                if session_type not in ['person', 'group']:
+                    return self.http_status(400, -1, 'session_type must be person or group')
+
+                webchat_adapter = None
+                for bot in self.ap.platform_mgr.bots:
+                    if hasattr(bot.adapter, '__class__') and bot.adapter.__class__.__name__ == 'WebChatAdapter':
+                        webchat_adapter = bot.adapter
+                        break
+
+                if not webchat_adapter:
+                    return self.http_status(404, -1, 'WebChat adapter not found')
+
+                webchat_adapter.reset_debug_session(session_type)
+
+                return self.success(data={'message': 'Session reset successfully'})
+
+            except Exception as e:
+                return self.http_status(500, -1, f'Internal server error: {str(e)}')

pkg/api/http/controller/main.py

@@ -13,10 +13,12 @@ from . import groups
 from . import group
 from .groups import provider as groups_provider
 from .groups import platform as groups_platform
+from .groups import pipelines as groups_pipelines
 
 importutil.import_modules_in_pkg(groups)
 importutil.import_modules_in_pkg(groups_provider)
 importutil.import_modules_in_pkg(groups_platform)
+importutil.import_modules_in_pkg(groups_pipelines)
 
 
 class HTTPController:

pkg/persistence/mgr.py

@@ -66,13 +66,15 @@ class PersistenceManager:
 
         # write default pipeline
         result = await self.execute_async(sqlalchemy.select(pipeline.LegacyPipeline))
+        default_pipeline_uuid = None
         if result.first() is None:
             self.ap.logger.info('Creating default pipeline...')
 
             pipeline_config = json.load(open('templates/default-pipeline-config.json', 'r', encoding='utf-8'))
 
+            default_pipeline_uuid = str(uuid.uuid4())
             pipeline_data = {
-                'uuid': str(uuid.uuid4()),
+                'uuid': default_pipeline_uuid,
                 'for_version': self.ap.ver_mgr.get_current_version(),
                 'stages': pipeline_service.default_stage_order,
                 'is_default': True,
@@ -82,6 +84,7 @@ class PersistenceManager:
             }
 
             await self.execute_async(sqlalchemy.insert(pipeline.LegacyPipeline).values(pipeline_data))
+
         # =================================
 
         # run migrations

pkg/pipeline/controller.py

@@ -51,11 +51,10 @@ class Controller:
                             # find pipeline
                             # Here firstly find the bot, then find the pipeline, in case the bot adapter's config is not the latest one.
                             # Like aiocqhttp, once a client is connected, even the adapter was updated and restarted, the existing client connection will not be affected.
-                            bot = await self.ap.platform_mgr.get_bot_by_uuid(selected_query.bot_uuid)
-                            if bot:
-                                pipeline = await self.ap.pipeline_mgr.get_pipeline_by_uuid(
-                                    bot.bot_entity.use_pipeline_uuid
-                                )
+                            pipeline_uuid = selected_query.pipeline_uuid
+
+                            if pipeline_uuid:
+                                pipeline = await self.ap.pipeline_mgr.get_pipeline_by_uuid(pipeline_uuid)
                                 if pipeline:
                                     await pipeline.run(selected_query)
 

pkg/pipeline/pool.py

@@ -35,6 +35,7 @@ class QueryPool:
         message_event: platform_events.MessageEvent,
         message_chain: platform_message.MessageChain,
         adapter: msadapter.MessagePlatformAdapter,
+        pipeline_uuid: typing.Optional[str] = None,
     ) -> entities.Query:
         async with self.condition:
             query = entities.Query(
@@ -48,6 +49,7 @@ class QueryPool:
                 resp_messages=[],
                 resp_message_chain=[],
                 adapter=adapter,
+                pipeline_uuid=pipeline_uuid,
             )
             self.queries.append(query)
             self.query_id_counter += 1

pkg/platform/botmgr.py

@@ -5,7 +5,6 @@ import asyncio
 import traceback
 import sqlalchemy
 
-
 #     FriendMessage, Image, MessageChain, Plain
 from . import adapter as msadapter
 
@@ -78,6 +77,7 @@ class RuntimeBot:
                 message_event=event,
                 message_chain=event.message_chain,
                 adapter=adapter,
+                pipeline_uuid=self.bot_entity.use_pipeline_uuid,
             )
 
         async def on_group_message(
@@ -102,6 +102,7 @@ class RuntimeBot:
                 message_event=event,
                 message_chain=event.message_chain,
                 adapter=adapter,
+                pipeline_uuid=self.bot_entity.use_pipeline_uuid,
             )
 
         self.adapter.register_listener(platform_events.FriendMessage, on_friend_message)
@@ -144,6 +145,8 @@ class PlatformManager:
 
     bots: list[RuntimeBot]
 
+    webchat_proxy_bot: RuntimeBot
+
     adapter_components: list[engine.Component]
 
     adapter_dict: dict[str, type[msadapter.MessagePlatformAdapter]]
@@ -161,6 +164,31 @@ class PlatformManager:
             adapter_dict[component.metadata.name] = component.get_python_component_class()
         self.adapter_dict = adapter_dict
 
+        webchat_adapter_class = self.adapter_dict['webchat']
+
+        # initialize webchat adapter
+        webchat_logger = EventLogger(name='webchat-adapter', ap=self.ap)
+        webchat_adapter_inst = webchat_adapter_class(
+            {},
+            self.ap,
+            webchat_logger,
+        )
+
+        self.webchat_proxy_bot = RuntimeBot(
+            ap=self.ap,
+            bot_entity=persistence_bot.Bot(
+                uuid='webchat-proxy-bot',
+                name='WebChat',
+                description='',
+                adapter='webchat',
+                adapter_config={},
+                enable=True,
+            ),
+            adapter=webchat_adapter_inst,
+            logger=webchat_logger,
+        )
+        await self.webchat_proxy_bot.initialize()
+
         await self.load_bots_from_db()
 
     def get_running_adapters(self) -> list[msadapter.MessagePlatformAdapter]:
@@ -220,7 +248,9 @@ class PlatformManager:
                 return
 
     def get_available_adapters_info(self) -> list[dict]:
-        return [component.to_plain_dict() for component in self.adapter_components]
+        return [
+            component.to_plain_dict() for component in self.adapter_components if component.metadata.name != 'webchat'
+        ]
 
     def get_available_adapter_info_by_name(self, name: str) -> dict | None:
         for component in self.adapter_components:
@@ -273,6 +303,8 @@ class PlatformManager:
 
     async def run(self):
         # This method will only be called when the application launching
+        await self.webchat_proxy_bot.run()
+
         for bot in self.bots:
             if bot.enable:
                 await bot.run()

pkg/platform/sources/webchat.py

@@ -0,0 +1,209 @@
+import asyncio
+import logging
+import typing
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from .. import adapter as msadapter
+from ..types import events as platform_events, message as platform_message, entities as platform_entities
+from ...core import app
+from ..logger import EventLogger
+
+logger = logging.getLogger(__name__)
+
+
+class WebChatMessage(BaseModel):
+    id: int
+    role: str
+    content: str
+    message_chain: list[dict]
+    timestamp: str
+
+
+class WebChatSession:
+    id: str
+    message_lists: dict[str, list[WebChatMessage]] = {}
+    resp_waiters: dict[int, asyncio.Future[WebChatMessage]]
+
+    def __init__(self, id: str):
+        self.id = id
+        self.message_lists = {}
+        self.resp_waiters = {}
+
+    def get_message_list(self, pipeline_uuid: str) -> list[WebChatMessage]:
+        if pipeline_uuid not in self.message_lists:
+            self.message_lists[pipeline_uuid] = []
+
+        return self.message_lists[pipeline_uuid]
+
+
+class WebChatAdapter(msadapter.MessagePlatformAdapter):
+    """WebChat调试适配器，用于流水线调试"""
+
+    webchat_person_session: WebChatSession
+    webchat_group_session: WebChatSession
+
+    listeners: typing.Dict[
+        typing.Type[platform_events.Event],
+        typing.Callable[[platform_events.Event, msadapter.MessagePlatformAdapter], None],
+    ] = {}
+
+    def __init__(self, config: dict, ap: app.Application, logger: EventLogger):
+        self.ap = ap
+        self.logger = logger
+        self.config = config
+
+        self.webchat_person_session = WebChatSession(id='webchatperson')
+        self.webchat_group_session = WebChatSession(id='webchatgroup')
+
+        self.bot_account_id = 'webchatbot'
+
+    async def send_message(
+        self,
+        target_type: str,
+        target_id: str,
+        message: platform_message.MessageChain,
+    ) -> dict:
+        """发送消息到调试会话"""
+        session_key = target_id
+
+        if session_key not in self.debug_messages:
+            self.debug_messages[session_key] = []
+
+        message_data = {
+            'id': len(self.debug_messages[session_key]) + 1,
+            'type': 'bot',
+            'content': str(message),
+            'timestamp': datetime.now().isoformat(),
+            'message_chain': [component.__dict__ for component in message],
+        }
+
+        self.debug_messages[session_key].append(message_data)
+
+        await self.logger.info(f'Send message to {session_key}: {message}')
+
+        return message_data
+
+    async def reply_message(
+        self,
+        message_source: platform_events.MessageEvent,
+        message: platform_message.MessageChain,
+        quote_origin: bool = False,
+    ) -> dict:
+        """回复消息"""
+        message_data = WebChatMessage(
+            id=-1,
+            role='assistant',
+            content=str(message),
+            message_chain=[component.__dict__ for component in message],
+            timestamp=datetime.now().isoformat(),
+        )
+
+        # notify waiter
+        if isinstance(message_source, platform_events.FriendMessage):
+            self.webchat_person_session.resp_waiters[message_source.message_chain.message_id].set_result(message_data)
+        elif isinstance(message_source, platform_events.GroupMessage):
+            self.webchat_group_session.resp_waiters[message_source.message_chain.message_id].set_result(message_data)
+
+        return message_data.model_dump()
+
+    def register_listener(
+        self,
+        event_type: typing.Type[platform_events.Event],
+        func: typing.Callable[[platform_events.Event, msadapter.MessagePlatformAdapter], typing.Awaitable[None]],
+    ):
+        """注册事件监听器"""
+        self.listeners[event_type] = func
+
+    def unregister_listener(
+        self,
+        event_type: typing.Type[platform_events.Event],
+        func: typing.Callable[[platform_events.Event, msadapter.MessagePlatformAdapter], typing.Awaitable[None]],
+    ):
+        """取消注册事件监听器"""
+        del self.listeners[event_type]
+
+    async def run_async(self):
+        """运行适配器"""
+        await self.logger.info('WebChat调试适配器已启动')
+
+        try:
+            while True:
+                await asyncio.sleep(1)
+        except asyncio.CancelledError:
+            await self.logger.info('WebChat调试适配器已停止')
+            raise
+
+    async def kill(self):
+        """停止适配器"""
+        await self.logger.info('WebChat调试适配器正在停止')
+
+    async def send_webchat_message(
+        self, pipeline_uuid: str, session_type: str, message_chain_obj: typing.List[dict]
+    ) -> dict:
+        """发送调试消息到流水线"""
+        if session_type == 'person':
+            use_session = self.webchat_person_session
+        else:
+            use_session = self.webchat_group_session
+
+        message_chain = platform_message.MessageChain.parse_obj(message_chain_obj)
+
+        message_id = len(use_session.get_message_list(pipeline_uuid)) + 1
+
+        use_session.get_message_list(pipeline_uuid).append(
+            WebChatMessage(
+                id=message_id,
+                role='user',
+                content=str(message_chain),
+                message_chain=message_chain_obj,
+                timestamp=datetime.now().isoformat(),
+            )
+        )
+
+        message_chain.insert(0, platform_message.Source(id=message_id, time=datetime.now().timestamp()))
+
+        if session_type == 'person':
+            sender = platform_entities.Friend(id='webchatperson', nickname='User')
+            event = platform_events.FriendMessage(
+                sender=sender, message_chain=message_chain, time=datetime.now().timestamp()
+            )
+        else:
+            group = platform_entities.Group(
+                id='webchatgroup', name='Group', permission=platform_entities.Permission.Member
+            )
+            sender = platform_entities.GroupMember(
+                id='webchatperson',
+                member_name='User',
+                group=group,
+                permission=platform_entities.Permission.Member,
+            )
+            event = platform_events.GroupMessage(
+                sender=sender, message_chain=message_chain, time=datetime.now().timestamp()
+            )
+
+        self.ap.platform_mgr.webchat_proxy_bot.bot_entity.use_pipeline_uuid = pipeline_uuid
+
+        if event.__class__ in self.listeners:
+            await self.listeners[event.__class__](event, self)
+
+        # set waiter
+        waiter = asyncio.Future[WebChatMessage]()
+        use_session.resp_waiters[message_id] = waiter
+        waiter.add_done_callback(lambda future: use_session.resp_waiters.pop(message_id))
+
+        resp_message = await waiter
+
+        resp_message.id = len(use_session.get_message_list(pipeline_uuid)) + 1
+
+        use_session.get_message_list(pipeline_uuid).append(resp_message)
+
+        return resp_message.model_dump()
+
+    def get_webchat_messages(self, pipeline_uuid: str, session_type: str) -> list[dict]:
+        """获取调试消息历史"""
+        if session_type == 'person':
+            return [message.model_dump() for message in self.webchat_person_session.get_message_list(pipeline_uuid)]
+        else:
+            return [message.model_dump() for message in self.webchat_group_session.get_message_list(pipeline_uuid)]

pkg/platform/sources/webchat.yaml

@@ -0,0 +1,16 @@
+apiVersion: v1
+kind: MessagePlatformAdapter
+metadata:
+  name: webchat
+  label:
+    en_US: "WebChat Debug"
+    zh_Hans: "网页聊天调试"
+  description:
+    en_US: "WebChat adapter for pipeline debugging"
+    zh_Hans: "用于流水线调试的网页聊天适配器"
+  icon: ""
+spec: {}
+execution:
+  python:
+    path: "webchat.py"
+    attr: "WebChatAdapter"