refactor: consolidate bot event routing

src/langbot/pkg/api/http/controller/groups/pipelines/embed.py

@@ -3,7 +3,7 @@
 All user-facing URLs are keyed by **bot_uuid** (not pipeline_uuid) so that
 internal pipeline identifiers are never exposed to end-users.  Each handler
 resolves the bot_uuid to the owning ``web_page_bot`` RuntimeBot and extracts
-the bound pipeline_uuid for internal routing.
+the message event route's Pipeline target for internal routing.
 """
 
 import asyncio
@@ -62,16 +62,17 @@ class EmbedRouterGroup(group.RouterGroup):
         """Resolve *bot_uuid* to ``(runtime_bot, pipeline_uuid)``.
 
         Returns ``(None, None)`` when the bot does not exist, is not a
-        ``web_page_bot``, is disabled, or has no pipeline bound.
+        ``web_page_bot``, is disabled, or has no Pipeline target for messages.
         """
         for bot in self.ap.platform_mgr.bots:
+            pipeline_uuid = bot.get_pipeline_target_for_event_type('message.received')
             if (
                 bot.bot_entity.uuid == bot_uuid
                 and bot.bot_entity.adapter == 'web_page_bot'
                 and bot.bot_entity.enable
-                and bot.bot_entity.use_pipeline_uuid
+                and pipeline_uuid
             ):
-                return bot, bot.bot_entity.use_pipeline_uuid
+                return bot, pipeline_uuid
         return None, None
 
     def _get_bot_config(self, bot_uuid: str) -> dict:

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

@@ -17,7 +17,7 @@ AGENT_DEFAULT_EVENT_PATTERNS = ['*']
 
 
 class AgentService:
-    """Unified product surface for Agent orchestration instances and Pipelines."""
+    """Unified product surface for Agent processors and Pipelines."""
 
     ap: app.Application
 

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

@@ -15,6 +15,15 @@ class BotService:
     """Bot service"""
 
     ap: app.Application
+    BOT_FIELDS = {
+        'uuid',
+        'name',
+        'description',
+        'adapter',
+        'adapter_config',
+        'enable',
+        'event_bindings',
+    }
 
     def __init__(self, ap: app.Application) -> None:
         self.ap = ap
@@ -115,6 +124,17 @@ class BotService:
 
         return normalized
 
+    async def _prepare_bot_data(self, bot_data: dict, *, include_uuid: bool) -> dict:
+        """Normalize Bot write payloads to the current event-routing model."""
+        update_data = bot_data.copy()
+        if not include_uuid:
+            update_data.pop('uuid', None)
+
+        update_data = {key: value for key, value in update_data.items() if key in self.BOT_FIELDS}
+        if 'event_bindings' in update_data:
+            update_data['event_bindings'] = await self._normalize_event_bindings(update_data.get('event_bindings'))
+        return update_data
+
     async def get_bots(self, include_secret: bool = True) -> list[dict]:
         """获取所有机器人"""
         result = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_bot.Bot))
@@ -188,7 +208,9 @@ class BotService:
                 raise ValueError(f'Maximum number of bots ({max_bots}) reached')
 
         # TODO: 检查配置信息格式
+        bot_data = await self._prepare_bot_data(bot_data, include_uuid=True)
         bot_data['uuid'] = str(uuid.uuid4())
+        bot_data.setdefault('event_bindings', [])
 
         await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_bot.Bot).values(bot_data))
 
@@ -200,18 +222,7 @@ class BotService:
 
     async def update_bot(self, bot_uuid: str, bot_data: dict) -> None:
         """Update bot"""
-        update_data = bot_data.copy()
-
-        if 'uuid' in update_data:
-            del update_data['uuid']
-
-        if 'event_bindings' in update_data:
-            update_data['event_bindings'] = await self._normalize_event_bindings(update_data.get('event_bindings'))
-
-        # clear legacy routing fields — routing is now fully managed via event_bindings
-        update_data.pop('use_pipeline_uuid', None)
-        update_data.pop('use_pipeline_name', None)
-        update_data.pop('pipeline_routing_rules', None)
+        update_data = await self._prepare_bot_data(bot_data, include_uuid=False)
 
         await self.ap.persistence_mgr.execute_async(
             sqlalchemy.update(persistence_bot.Bot).values(update_data).where(persistence_bot.Bot.uuid == bot_uuid)

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

@@ -218,19 +218,6 @@ class PipelineService:
 
         pipeline = await self.get_pipeline(pipeline_uuid)
 
-        if 'name' in pipeline_data:
-            from ....entity.persistence import bot as persistence_bot
-
-            result = await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.select(persistence_bot.Bot).where(persistence_bot.Bot.use_pipeline_uuid == pipeline_uuid)
-            )
-
-            bots = result.all()
-
-            for bot in bots:
-                bot_data = {'use_pipeline_name': pipeline_data['name']}
-                await self.ap.bot_service.update_bot(bot.uuid, bot_data)
-
         await self.ap.pipeline_mgr.remove_pipeline(pipeline_uuid)
         await self.ap.pipeline_mgr.load_pipeline(pipeline)
 

src/langbot/pkg/api/mcp/server.py

@@ -142,7 +142,7 @@ class LangBotMCPServer:
             return _dump({'ok': True})
 
         # ----- Agents -------------------------------------------------- #
-        @mcp.tool(description='List product-level Agents, including Agent orchestrations and Pipelines.')
+        @mcp.tool(description='List product-level processors, including Agents and Pipelines.')
         async def list_agents() -> str:
             return _dump(await ap.agent_service.get_agents())
 
@@ -152,19 +152,19 @@ class LangBotMCPServer:
 
         @mcp.tool(
             description=(
-                'Create an Agent orchestration or Pipeline. `agent_data` matches '
+                'Create an Agent processor or Pipeline. `agent_data` matches '
                 'POST /api/v1/agents; set kind to `agent` or `pipeline`. Returns the new UUID and kind.'
             )
         )
         async def create_agent(agent_data: dict) -> str:
             return _dump(await ap.agent_service.create_agent(agent_data))
 
-        @mcp.tool(description='Update an Agent orchestration or Pipeline by UUID.')
+        @mcp.tool(description='Update an Agent processor or Pipeline by UUID.')
         async def update_agent(agent_uuid: str, agent_data: dict) -> str:
             await ap.agent_service.update_agent(agent_uuid, agent_data)
             return _dump({'ok': True})
 
-        @mcp.tool(description='Delete an Agent orchestration or Pipeline by UUID.')
+        @mcp.tool(description='Delete an Agent processor or Pipeline by UUID.')
         async def delete_agent(agent_uuid: str) -> str:
             await ap.agent_service.delete_agent(agent_uuid)
             return _dump({'ok': True})

src/langbot/pkg/entity/persistence/agent.py

@@ -4,7 +4,7 @@ from .base import Base
 
 
 class Agent(Base):
-    """Product-level Agent orchestration instance."""
+    """Product-level Agent processor."""
 
     __tablename__ = 'agents'
 

src/langbot/pkg/entity/persistence/bot.py

@@ -28,9 +28,6 @@ class Bot(Base):
     adapter = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
     adapter_config = sqlalchemy.Column(sqlalchemy.JSON, nullable=False)
     enable = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False, default=False)
-    use_pipeline_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
-    use_pipeline_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
-    pipeline_routing_rules = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, server_default='[]')
     event_bindings = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, server_default='[]')
     created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
     updated_at = sqlalchemy.Column(

src/langbot/pkg/persistence/alembic/versions/0009_migrate_routing_to_event_bindings.py

@@ -16,6 +16,16 @@ down_revision = '0008_agent_product_surface'
 depends_on = None
 
 
+def _table_exists(table_name: str) -> bool:
+    return table_name in sa.inspect(op.get_bind()).get_table_names()
+
+
+def _column_exists(table_name: str, column_name: str) -> bool:
+    if not _table_exists(table_name):
+        return False
+    return column_name in {column['name'] for column in sa.inspect(op.get_bind()).get_columns(table_name)}
+
+
 def _rule_to_filters(rule: dict) -> list[dict] | None:
     """Convert a pipeline_routing_rule to event_binding filters (best effort).
 
@@ -38,6 +48,11 @@ def _rule_to_filters(rule: dict) -> list[dict] | None:
 
 
 def upgrade() -> None:
+    if not _table_exists('bots') or not _column_exists('bots', 'event_bindings'):
+        return
+    if not _column_exists('bots', 'use_pipeline_uuid') or not _column_exists('bots', 'pipeline_routing_rules'):
+        return
+
     bind = op.get_bind()
     rows = bind.execute(
         sa.text('SELECT uuid, use_pipeline_uuid, pipeline_routing_rules, event_bindings FROM bots')

src/langbot/pkg/persistence/alembic/versions/0011_drop_legacy_bot_routing.py

@@ -0,0 +1,55 @@
+"""drop legacy bot pipeline routing columns
+
+Revision ID: 0011_drop_legacy_bot_routing
+Revises: 0010_merge_mcp_resource_agent_heads
+Create Date: 2026-07-01
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = '0011_drop_legacy_bot_routing'
+down_revision = '0010_merge_mcp_resource_agent_heads'
+branch_labels = None
+depends_on = None
+
+
+def _table_exists(table_name: str) -> bool:
+    return table_name in sa.inspect(op.get_bind()).get_table_names()
+
+
+def _column_exists(table_name: str, column_name: str) -> bool:
+    if not _table_exists(table_name):
+        return False
+    return column_name in {column['name'] for column in sa.inspect(op.get_bind()).get_columns(table_name)}
+
+
+def upgrade() -> None:
+    if not _table_exists('bots'):
+        return
+    existing_columns = {column['name'] for column in sa.inspect(op.get_bind()).get_columns('bots')}
+    columns_to_drop = [
+        column_name
+        for column_name in ('use_pipeline_name', 'use_pipeline_uuid', 'pipeline_routing_rules')
+        if column_name in existing_columns
+    ]
+    if not columns_to_drop:
+        return
+
+    with op.batch_alter_table('bots', schema=None) as batch_op:
+        for column_name in columns_to_drop:
+            batch_op.drop_column(column_name)
+
+
+def downgrade() -> None:
+    if not _table_exists('bots'):
+        return
+
+    with op.batch_alter_table('bots', schema=None) as batch_op:
+        if not _column_exists('bots', 'use_pipeline_name'):
+            batch_op.add_column(sa.Column('use_pipeline_name', sa.String(255), nullable=True))
+        if not _column_exists('bots', 'use_pipeline_uuid'):
+            batch_op.add_column(sa.Column('use_pipeline_uuid', sa.String(255), nullable=True))
+        if not _column_exists('bots', 'pipeline_routing_rules'):
+            batch_op.add_column(sa.Column('pipeline_routing_rules', sa.JSON(), nullable=False, server_default='[]'))

src/langbot/pkg/platform/botmgr.py

@@ -14,8 +14,6 @@ from ..core import app, entities as core_entities, taskmgr
 from ..discover import engine
 
 from ..entity.persistence import bot as persistence_bot
-from ..entity.persistence import pipeline as persistence_pipeline
-
 from ..entity.errors import platform as platform_errors
 from ..agent.runner.host_models import (
     AgentBinding,
@@ -32,6 +30,7 @@ import langbot_plugin.api.entities.builtin.provider.session as provider_session
 import langbot_plugin.api.entities.builtin.provider.message as provider_message
 import langbot_plugin.api.entities.events as plugin_events
 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.entities.builtin.platform.message as platform_message
 import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
 from langbot_plugin.api.entities.builtin.agent_runner.event import (
@@ -74,26 +73,6 @@ class RuntimeBot:
         self.task_context = taskmgr.TaskContext()
         self.logger = logger
 
-    @staticmethod
-    def _match_operator(actual: str, operator: str, expected: str) -> bool:
-        """Evaluate a single operator condition."""
-        if operator == 'eq':
-            return actual == expected
-        elif operator == 'neq':
-            return actual != expected
-        elif operator == 'contains':
-            return expected in actual
-        elif operator == 'not_contains':
-            return expected not in actual
-        elif operator == 'starts_with':
-            return actual.startswith(expected)
-        elif operator == 'regex':
-            try:
-                return bool(re.search(expected, actual))
-            except re.error:
-                return False
-        return False
-
     PIPELINE_DISCARD = '__discard__'
     PIPELINE_DISCARD_DISPLAY_NAME = 'Discarded'
     EVENT_DATA_MAX_STRING_BYTES = 512
@@ -229,12 +208,7 @@ class RuntimeBot:
             if isinstance(event_filter, dict)
         )
 
-    def _resolve_eba_event_binding(
-        self,
-        event: platform_events.EBAEvent,
-        event_type: str,
-    ) -> dict[str, typing.Any] | None:
-        """Resolve the highest priority Bot event binding for a platform event."""
+    def _get_event_bindings(self) -> list[dict[str, typing.Any]]:
         raw_bindings = self.bot_entity.event_bindings or []
         if isinstance(raw_bindings, str):
             try:
@@ -242,11 +216,18 @@ class RuntimeBot:
             except json.JSONDecodeError:
                 raw_bindings = []
         if not isinstance(raw_bindings, list):
-            return None
+            return []
+        return [binding for binding in raw_bindings if isinstance(binding, dict)]
 
+    def _resolve_eba_event_binding(
+        self,
+        event: platform_events.EBAEvent,
+        event_type: str,
+    ) -> dict[str, typing.Any] | None:
+        """Resolve the highest priority Bot event binding for a platform event."""
         matched: list[tuple[int, int, dict[str, typing.Any]]] = []
-        for index, binding in enumerate(raw_bindings):
-            if not isinstance(binding, dict) or not binding.get('enabled', True):
+        for index, binding in enumerate(self._get_event_bindings()):
+            if not binding.get('enabled', True):
                 continue
 
             event_pattern = str(binding.get('event_pattern') or '')
@@ -265,6 +246,29 @@ class RuntimeBot:
         matched.sort(key=lambda item: (item[0], item[1]), reverse=True)
         return matched[0][2]
 
+    def get_pipeline_target_for_event_type(self, event_type: str = 'message.received') -> str | None:
+        """Return the first Pipeline target configured for an event type."""
+        matched: list[tuple[int, int, str]] = []
+        for index, binding in enumerate(self._get_event_bindings()):
+            if not binding.get('enabled', True):
+                continue
+            if binding.get('target_type') != 'pipeline':
+                continue
+            target_uuid = str(binding.get('target_uuid') or '')
+            if not target_uuid:
+                continue
+            if not self._match_event_pattern(event_type, str(binding.get('event_pattern') or '')):
+                continue
+            priority = int(binding.get('priority') or 0)
+            order = int(binding.get('order', index))
+            matched.append((priority, -order, target_uuid))
+
+        if not matched:
+            return None
+
+        matched.sort(key=lambda item: (item[0], item[1]), reverse=True)
+        return matched[0][2]
+
     @staticmethod
     def _safe_model_dump(model: typing.Any) -> dict[str, typing.Any]:
         if model is None:
@@ -451,6 +455,56 @@ class RuntimeBot:
 
         return None, None, metadata
 
+    @staticmethod
+    def _extract_message_id(message_chain: platform_message.MessageChain) -> str:
+        for component in message_chain:
+            if isinstance(component, platform_message.Source):
+                value = getattr(component, 'id', '')
+                return str(value) if value is not None else ''
+        return ''
+
+    def _legacy_message_to_eba_event(
+        self,
+        event: platform_events.FriendMessage | platform_events.GroupMessage,
+        adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
+    ) -> platform_events.MessageReceivedEvent:
+        if isinstance(event, platform_events.GroupMessage):
+            group = platform_entities.UserGroup(
+                id=event.group.id,
+                name=event.group.name,
+            )
+            return platform_events.MessageReceivedEvent(
+                message_id=self._extract_message_id(event.message_chain),
+                message_chain=event.message_chain,
+                sender=platform_entities.User(
+                    id=event.sender.id,
+                    nickname=event.sender.member_name,
+                ),
+                chat_type=platform_entities.ChatType.GROUP,
+                chat_id=event.group.id,
+                group=group,
+                timestamp=event.time or time.time(),
+                bot_uuid=self.bot_entity.uuid,
+                adapter_name=adapter.__class__.__name__,
+                source_platform_object=event.source_platform_object,
+            )
+
+        return platform_events.MessageReceivedEvent(
+            message_id=self._extract_message_id(event.message_chain),
+            message_chain=event.message_chain,
+            sender=platform_entities.User(
+                id=event.sender.id,
+                nickname=event.sender.nickname,
+                remark=event.sender.remark,
+            ),
+            chat_type=platform_entities.ChatType.PRIVATE,
+            chat_id=event.sender.id,
+            timestamp=event.time or time.time(),
+            bot_uuid=self.bot_entity.uuid,
+            adapter_name=adapter.__class__.__name__,
+            source_platform_object=event.source_platform_object,
+        )
+
     @classmethod
     def _build_agent_input(cls, event: platform_events.EBAEvent) -> AgentInput:
         text = None
@@ -635,6 +689,22 @@ class RuntimeBot:
             platform_message.MessageChain([platform_message.Plain(text=final_text)]),
         )
 
+    async def _handle_platform_event(
+        self,
+        event: platform_events.EBAEvent,
+        adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
+    ) -> None:
+        event.bot_uuid = self.bot_entity.uuid
+        plugin_event = self._eba_event_to_plugin_event(event)
+
+        if plugin_event is not None:
+            try:
+                await self.ap.plugin_connector.emit_event(plugin_event)
+            except Exception:
+                await self.logger.error(f'Failed to dispatch platform event to plugins: {traceback.format_exc()}')
+
+        await self._dispatch_eba_event_to_agent(event, adapter)
+
     async def _dispatch_eba_event_to_agent(
         self,
         event: platform_events.EBAEvent,
@@ -644,8 +714,7 @@ class RuntimeBot:
 
         event_binding = self._resolve_eba_event_binding(event, event_type)
         if event_binding is None:
-            if isinstance(event, platform_events.MessageReceivedEvent):
-                await self._dispatch_eba_message_to_pipeline(event, adapter)
+            await self.logger.info(f'Platform event {event_type} ignored: no event route matched')
             return
 
         target_type = event_binding.get('target_type')
@@ -708,64 +777,6 @@ class RuntimeBot:
                 f'Failed to deliver Agent output for EBA event {event_type}: {traceback.format_exc()}'
             )
 
-    def resolve_pipeline_uuid(
-        self,
-        launcher_type: str,
-        launcher_id: str,
-        message_text: str,
-        message_element_types: list[str] | None = None,
-    ) -> tuple[str | None, bool]:
-        """Resolve pipeline UUID based on routing rules.
-
-        Rules are evaluated in order; first match wins.
-        Falls back to use_pipeline_uuid if no rule matches.
-
-        Rule types:
-          - launcher_type: session type ("person" / "group")
-          - launcher_id: session / group id
-          - message_content: message text content
-          - message_has_element: message contains element of given type
-            (Image, Voice, File, Forward, Face, At, AtAll, Quote)
-            Operators: eq (has), neq (doesn't have)
-
-        Operators: eq, neq, contains, not_contains, starts_with, regex
-
-        When pipeline_uuid is ``__discard__``, the message should be
-        silently dropped by the caller.
-
-        Returns:
-            tuple: (pipeline_uuid, routed_by_rule) - routed_by_rule is True
-            when a routing rule matched, False when falling back to default.
-        """
-        rules = self.bot_entity.pipeline_routing_rules or []
-        element_type_set = set(message_element_types or [])
-
-        for rule in rules:
-            rule_type = rule.get('type')
-            operator = rule.get('operator', 'eq')
-            rule_value = rule.get('value', '')
-            target_uuid = rule.get('pipeline_uuid')
-            if not rule_type or not target_uuid:
-                continue
-
-            if rule_type == 'launcher_type':
-                if self._match_operator(launcher_type, operator, rule_value):
-                    return target_uuid, True
-            elif rule_type == 'launcher_id':
-                if self._match_operator(str(launcher_id), operator, str(rule_value)):
-                    return target_uuid, True
-            elif rule_type == 'message_content':
-                if self._match_operator(message_text, operator, rule_value):
-                    return target_uuid, True
-            elif rule_type == 'message_has_element':
-                has_element = rule_value in element_type_set
-                if operator == 'eq' and has_element:
-                    return target_uuid, True
-                elif operator == 'neq' and not has_element:
-                    return target_uuid, True
-
-        return self.bot_entity.use_pipeline_uuid, False
-
     async def _record_discarded_message(
         self,
         launcher_type: provider_session.LauncherTypes,
@@ -932,21 +943,21 @@ class RuntimeBot:
             event: platform_events.FriendMessage,
             adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
         ):
-            await self._handle_legacy_message_event(
-                event,
-                adapter,
-                pipeline_uuid_override=websocket_pipeline_uuid(event, adapter),
-            )
+            pipeline_uuid = websocket_pipeline_uuid(event, adapter)
+            if pipeline_uuid:
+                await self._handle_legacy_message_event(event, adapter, pipeline_uuid_override=pipeline_uuid)
+                return
+            await self._handle_platform_event(self._legacy_message_to_eba_event(event, adapter), adapter)
 
         async def on_group_message(
             event: platform_events.GroupMessage,
             adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
         ):
-            await self._handle_legacy_message_event(
-                event,
-                adapter,
-                pipeline_uuid_override=websocket_pipeline_uuid(event, adapter),
-            )
+            pipeline_uuid = websocket_pipeline_uuid(event, adapter)
+            if pipeline_uuid:
+                await self._handle_legacy_message_event(event, adapter, pipeline_uuid_override=pipeline_uuid)
+                return
+            await self._handle_platform_event(self._legacy_message_to_eba_event(event, adapter), adapter)
 
         self.adapter.register_listener(platform_events.FriendMessage, on_friend_message)
         self.adapter.register_listener(platform_events.GroupMessage, on_group_message)
@@ -957,20 +968,11 @@ class RuntimeBot:
             adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
         ):
             try:
-                # Resolve pipeline name
+                pipeline_id = self.get_pipeline_target_for_event_type('message.received') or ''
                 pipeline_name = ''
-                if self.bot_entity.use_pipeline_uuid:
-                    try:
-                        pipeline_result = await self.ap.persistence_mgr.execute_async(
-                            sqlalchemy.select(persistence_pipeline.LegacyPipeline.name).where(
-                                persistence_pipeline.LegacyPipeline.uuid == self.bot_entity.use_pipeline_uuid
-                            )
-                        )
-                        pipeline_row = pipeline_result.first()
-                        if pipeline_row:
-                            pipeline_name = pipeline_row[0]
-                    except Exception:
-                        pass
+                if pipeline_id:
+                    pipeline = await self.ap.pipeline_service.get_pipeline(pipeline_id)
+                    pipeline_name = pipeline.get('name', '') if pipeline else ''
 
                 await self.ap.monitoring_service.record_feedback(
                     feedback_id=event.feedback_id,
@@ -979,7 +981,7 @@ class RuntimeBot:
                     inaccurate_reasons=event.inaccurate_reasons,
                     bot_id=self.bot_entity.uuid,
                     bot_name=self.bot_entity.name,
-                    pipeline_id=self.bot_entity.use_pipeline_uuid or '',
+                    pipeline_id=pipeline_id,
                     pipeline_name=pipeline_name,
                     session_id=event.session_id,
                     message_id=event.message_id,
@@ -999,16 +1001,7 @@ class RuntimeBot:
             event: platform_events.EBAEvent,
             adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
         ):
-            event.bot_uuid = self.bot_entity.uuid
-            plugin_event = self._eba_event_to_plugin_event(event)
-
-            if plugin_event is not None:
-                try:
-                    await self.ap.plugin_connector.emit_event(plugin_event)
-                except Exception:
-                    await self.logger.error(f'Failed to dispatch EBA event to plugins: {traceback.format_exc()}')
-
-            await self._dispatch_eba_event_to_agent(event, adapter)
+            await self._handle_platform_event(event, adapter)
 
         self.adapter.register_listener(platform_events.EBAEvent, on_eba_event)
 

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

@@ -77,6 +77,9 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
     stream_enabled: bool = pydantic.Field(default=True, exclude=True)
     """是否启用流式输出"""
 
+    current_pipeline_uuid: str = pydantic.Field(default='', exclude=True)
+    """Pipeline currently associated with the active WebSocket exchange."""
+
     def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger, **kwargs):
         super().__init__(
             config=config,
@@ -90,6 +93,18 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
         self.bot_account_id = 'websocketbot'
         self.outbound_message_queue = asyncio.Queue()
         self.stream_enabled = True
+        self.current_pipeline_uuid = ''
+
+    def _resolve_pipeline_uuid(
+        self,
+        message_source: platform_events.MessageEvent | None = None,
+        target_id: str = '',
+    ) -> str:
+        if message_source is not None:
+            event_pipeline_uuid = getattr(message_source, '_langbot_pipeline_uuid', '')
+            if isinstance(event_pipeline_uuid, str) and event_pipeline_uuid:
+                return event_pipeline_uuid
+        return self.current_pipeline_uuid or target_id
 
     async def send_message(
         self,
@@ -103,8 +118,7 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
         target_id 可能是 launcher_id（如 websocket_xxx）或 pipeline_uuid。
         我们需要尝试两种方式来确保消息能够送达。
         """
-        # 获取当前的 pipeline_uuid
-        pipeline_uuid = self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid
+        pipeline_uuid = self._resolve_pipeline_uuid(target_id=str(target_id))
         session_type = 'group' if target_type == 'group' else 'person'
 
         # 选择会话
@@ -152,8 +166,7 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
             else self.websocket_person_session
         )
 
-        # 从message_source获取pipeline_uuid和connection_id
-        pipeline_uuid = self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid
+        pipeline_uuid = self._resolve_pipeline_uuid(message_source)
         session_type = 'group' if isinstance(message_source, platform_events.GroupMessage) else 'person'
 
         # 生成新的消息ID
@@ -200,7 +213,7 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
             else self.websocket_person_session
         )
 
-        pipeline_uuid = self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid
+        pipeline_uuid = self._resolve_pipeline_uuid(message_source)
         session_type = 'group' if isinstance(message_source, platform_events.GroupMessage) else 'person'
         message_list = session.get_message_list(pipeline_uuid)
         stream_message_indexes = session.get_stream_message_indexes(pipeline_uuid)
@@ -380,6 +393,7 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
                        When provided, its identity is used for logging and session tracking.
         """
         pipeline_uuid = connection.pipeline_uuid
+        self.current_pipeline_uuid = pipeline_uuid
         session_type = connection.session_type
 
         # 获取stream参数，默认为True
@@ -449,11 +463,6 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
 
         object.__setattr__(event, '_langbot_pipeline_uuid', pipeline_uuid)
 
-        # 设置流水线UUID (proxy bot always needs it for reply_message routing)
-        self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid = pipeline_uuid
-        if owner_bot is not None:
-            owner_bot.bot_entity.use_pipeline_uuid = pipeline_uuid
-
         # 异步触发事件处理
         # Use owner_bot's listeners if available, otherwise fall back to proxy bot
         listeners = (