Feat: bot message routing (#2100)

* refactor: pipeline routing rules - add routed_by_rule bypass and diagnostic logging

- Add routing rules editor (RoutingRulesEditor component)
- Add routed_by_rule bypass logic in response rules
- Add diagnostic logging for pipeline routing
- Database migration for bot pipeline routing rules
- Extract RoutingRulesEditor component from BotForm
- Revert log levels to debug

* feat: add message_has_element routing rule type

Support routing by message element type (Image, Voice, File, Forward,
Face, At, AtAll, Quote) with eq/neq operators.

* test: add unit tests for pipeline routing rules

20 tests covering _match_operator (eq/neq/contains/not_contains/
starts_with/regex/invalid) and resolve_pipeline_uuid (launcher_type/
launcher_id/message_content/message_has_element/first-match-wins/
skip-invalid/default-operator).

* fix(web): add missing 'message_has_element' to routing rule type validation

The Zod schema and TypeScript type for PipelineRoutingRule.type were
missing the 'message_has_element' variant, causing silent form validation
failure when saving routing rules with this type.

* feat: add pipeline discard functionality and localization support

* feat(web): improve drag-and-drop with DragOverlay, add discard monitoring and pipeline icons

- Add DragOverlay for smooth cursor-following drag in routing rules editor
- Remove transition to eliminate redundant swap animation on drop
- Record discarded messages in monitoring system via _record_discarded_message
- Display pipeline name (Workflow icon) and runner name (Play icon) on session monitor messages
- Show discard badge on discarded messages in session monitor
- Add i18n translations for discarded/userMessage/botMessage

* fix: ensure discarded messages appear in session monitor and improve icons

- Create/update monitoring session for discarded messages so they show in
  the bot session monitor (was only inserting message rows, not sessions)
- Use human-readable 'Discarded' as pipeline_name instead of '__discard__'
- Change runner icon from Play to Bot for better AI Agent semantics

* fix: merge discarded messages into same session and remove session-level pipeline name

- Use LauncherTypes enum for session_id in discarded messages to match
  the format used by monitoring_helper (fixes duplicate sessions)
- Don't overwrite session pipeline info on discard — a session can have
  messages from multiple pipelines
- Remove pipeline_name from session list and chat header since it's
  now shown per-message and a session is no longer single-pipeline

* fix(web): only show save button on config tab in bot detail page

* fix(web): scroll to bottom after messages render in session monitor

---------

Co-authored-by: RockChinQ <rockchinq@gmail.com>

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

@@ -16,6 +16,7 @@ class Bot(Base):
     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='[]')
     created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
     updated_at = sqlalchemy.Column(
         sqlalchemy.DateTime,

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

@@ -0,0 +1,15 @@
+import sqlalchemy
+from .. import migration
+
+
+@migration.migration_class(25)
+class DBMigrateBotPipelineRoutingRules(migration.DBMigration):
+    """Add pipeline_routing_rules column to bots table"""
+
+    async def upgrade(self):
+        sql_text = sqlalchemy.text("ALTER TABLE bots ADD COLUMN pipeline_routing_rules JSON NOT NULL DEFAULT '[]'")
+        await self.ap.persistence_mgr.execute_async(sql_text)
+
+    async def downgrade(self):
+        sql_text = sqlalchemy.text('ALTER TABLE bots DROP COLUMN pipeline_routing_rules')
+        await self.ap.persistence_mgr.execute_async(sql_text)

src/langbot/pkg/pipeline/aggregator.py

@@ -37,6 +37,7 @@ class PendingMessage:
     message_chain: platform_message.MessageChain
     adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter
     pipeline_uuid: typing.Optional[str]
+    routed_by_rule: bool = False
     timestamp: float = field(default_factory=time.time)
 
 
@@ -125,6 +126,7 @@ class MessageAggregator:
         message_chain: platform_message.MessageChain,
         adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
         pipeline_uuid: typing.Optional[str] = None,
+        routed_by_rule: bool = False,
     ) -> None:
         """Add a message to the aggregation buffer
 
@@ -145,6 +147,7 @@ class MessageAggregator:
                 message_chain=message_chain,
                 adapter=adapter,
                 pipeline_uuid=pipeline_uuid,
+                routed_by_rule=routed_by_rule,
             )
             return
 
@@ -159,6 +162,7 @@ class MessageAggregator:
             message_chain=message_chain,
             adapter=adapter,
             pipeline_uuid=pipeline_uuid,
+            routed_by_rule=routed_by_rule,
         )
 
         force_flush = False
@@ -217,6 +221,7 @@ class MessageAggregator:
                 message_chain=msg.message_chain,
                 adapter=msg.adapter,
                 pipeline_uuid=msg.pipeline_uuid,
+                routed_by_rule=msg.routed_by_rule,
             )
             return
 
@@ -231,6 +236,7 @@ class MessageAggregator:
             message_chain=merged_msg.message_chain,
             adapter=merged_msg.adapter,
             pipeline_uuid=merged_msg.pipeline_uuid,
+            routed_by_rule=merged_msg.routed_by_rule,
         )
 
     def _merge_messages(self, messages: list[PendingMessage]) -> PendingMessage:

src/langbot/pkg/pipeline/controller.py

@@ -63,6 +63,14 @@ class Controller:
                                 pipeline = await self.ap.pipeline_mgr.get_pipeline_by_uuid(pipeline_uuid)
                                 if pipeline:
                                     await pipeline.run(selected_query)
+                                else:
+                                    self.ap.logger.warning(
+                                        f'Pipeline {pipeline_uuid} not found for query {selected_query.query_id}, query dropped'
+                                    )
+                            else:
+                                self.ap.logger.warning(
+                                    f'No pipeline_uuid for query {selected_query.query_id}, query dropped'
+                                )
 
                         async with self.ap.query_pool:
                             (await self.ap.sess_mgr.get_session(selected_query))._semaphore.release()

src/langbot/pkg/pipeline/pipelinemgr.py

@@ -323,6 +323,9 @@ class RuntimePipeline:
             event_ctx = await self.ap.plugin_connector.emit_event(event_obj, bound_plugins)
 
             if event_ctx.is_prevented_default():
+                self.ap.logger.debug(
+                    f'MessageReceived event prevented default for query {query.query_id}, pipeline={pipeline_name}'
+                )
                 return
 
             self.ap.logger.debug(f'Processing query {query.query_id}')

src/langbot/pkg/pipeline/pool.py

@@ -41,6 +41,7 @@ class QueryPool:
         message_chain: platform_message.MessageChain,
         adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
         pipeline_uuid: typing.Optional[str] = None,
+        routed_by_rule: bool = False,
     ) -> pipeline_query.Query:
         async with self.condition:
             query_id = self.query_id_counter
@@ -52,7 +53,7 @@ class QueryPool:
                 sender_id=sender_id,
                 message_event=message_event,
                 message_chain=message_chain,
-                variables={},
+                variables={'_routed_by_rule': routed_by_rule},
                 resp_messages=[],
                 resp_message_chain=[],
                 adapter=adapter,

src/langbot/pkg/pipeline/process/handlers/chat.py

@@ -61,6 +61,9 @@ class ChatMessageHandler(handler.MessageHandler):
 
                 yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
             else:
+                self.ap.logger.debug(
+                    f'NormalMessageReceived event prevented default for query {query.query_id} without reply'
+                )
                 yield entities.StageProcessResult(result_type=entities.ResultType.INTERRUPT, new_query=query)
         else:
             if event_ctx.event.user_message_alter is not None:

src/langbot/pkg/pipeline/resprule/resprule.py

@@ -37,6 +37,10 @@ class GroupRespondRuleCheckStage(stage.PipelineStage):
         if query.launcher_type.value != 'group':  # 只处理群消息
             return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
 
+        # 通过路由规则明确指定的流水线，跳过群响应规则检查
+        if query.variables and query.variables.get('_routed_by_rule', False):
+            return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
+
         rules = query.pipeline_config['trigger']['group-respond-rules']
 
         use_rule = rules

src/langbot/pkg/platform/botmgr.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import asyncio
+import json
+import re
 import traceback
 import sqlalchemy
 
@@ -52,6 +54,148 @@ 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'
+
+    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,
+        launcher_id: str | int,
+        sender_id: str | int,
+        message_event: platform_events.MessageEvent,
+        message_chain: platform_message.MessageChain,
+    ) -> None:
+        """Record a discarded message in the monitoring system."""
+        try:
+            if hasattr(message_chain, 'model_dump'):
+                message_content = json.dumps(message_chain.model_dump(), ensure_ascii=False)
+            else:
+                message_content = str(message_chain)
+
+            sender_name = None
+            if hasattr(message_event, 'sender'):
+                if hasattr(message_event.sender, 'nickname'):
+                    sender_name = message_event.sender.nickname
+                elif hasattr(message_event.sender, 'member_name'):
+                    sender_name = message_event.sender.member_name
+
+            # Use the same session_id format as monitoring_helper.py
+            session_id = f'{launcher_type}_{launcher_id}'
+            platform = launcher_type.value if hasattr(launcher_type, 'value') else str(launcher_type)
+
+            await self.ap.monitoring_service.record_message(
+                bot_id=self.bot_entity.uuid,
+                bot_name=self.bot_entity.name or self.bot_entity.uuid,
+                pipeline_id=self.PIPELINE_DISCARD,
+                pipeline_name=self.PIPELINE_DISCARD_DISPLAY_NAME,
+                message_content=message_content,
+                session_id=session_id,
+                status='discarded',
+                level='info',
+                platform=platform,
+                user_id=str(sender_id),
+                user_name=sender_name,
+            )
+
+            # Ensure the session exists so the message appears in the session monitor.
+            # Don't overwrite pipeline info — a session may have messages from
+            # multiple pipelines; discarding shouldn't change the displayed pipeline.
+            session_updated = await self.ap.monitoring_service.update_session_activity(
+                session_id,
+            )
+            if not session_updated:
+                # No session yet (first message for this launcher was discarded).
+                await self.ap.monitoring_service.record_session_start(
+                    session_id=session_id,
+                    bot_id=self.bot_entity.uuid,
+                    bot_name=self.bot_entity.name or self.bot_entity.uuid,
+                    pipeline_id=self.PIPELINE_DISCARD,
+                    pipeline_name=self.PIPELINE_DISCARD_DISPLAY_NAME,
+                    platform=platform,
+                    user_id=str(sender_id),
+                    user_name=sender_name,
+                )
+        except Exception as e:
+            await self.logger.error(f'Failed to record discarded message: {e}')
+
     async def initialize(self):
         async def on_friend_message(
             event: platform_events.FriendMessage,
@@ -83,6 +227,23 @@ class RuntimeBot:
                     if custom_launcher_id:
                         launcher_id = custom_launcher_id
 
+                message_text = str(event.message_chain)
+                element_types = [comp.type for comp in event.message_chain]
+                pipeline_uuid, routed_by_rule = self.resolve_pipeline_uuid(
+                    'person', launcher_id, message_text, element_types
+                )
+
+                if pipeline_uuid == self.PIPELINE_DISCARD:
+                    await self.logger.info('Person message discarded by routing rule')
+                    await self._record_discarded_message(
+                        provider_session.LauncherTypes.PERSON,
+                        launcher_id,
+                        event.sender.id,
+                        event,
+                        event.message_chain,
+                    )
+                    return
+
                 await self.ap.msg_aggregator.add_message(
                     bot_uuid=self.bot_entity.uuid,
                     launcher_type=provider_session.LauncherTypes.PERSON,
@@ -91,7 +252,8 @@ class RuntimeBot:
                     message_event=event,
                     message_chain=event.message_chain,
                     adapter=adapter,
-                    pipeline_uuid=self.bot_entity.use_pipeline_uuid,
+                    pipeline_uuid=pipeline_uuid,
+                    routed_by_rule=routed_by_rule,
                 )
             else:
                 await self.logger.info('Pipeline skipped for person message due to webhook response')
@@ -126,6 +288,23 @@ class RuntimeBot:
                     if custom_launcher_id:
                         launcher_id = custom_launcher_id
 
+                message_text = str(event.message_chain)
+                element_types = [comp.type for comp in event.message_chain]
+                pipeline_uuid, routed_by_rule = self.resolve_pipeline_uuid(
+                    'group', launcher_id, message_text, element_types
+                )
+
+                if pipeline_uuid == self.PIPELINE_DISCARD:
+                    await self.logger.info('Group message discarded by routing rule')
+                    await self._record_discarded_message(
+                        provider_session.LauncherTypes.GROUP,
+                        launcher_id,
+                        event.sender.id,
+                        event,
+                        event.message_chain,
+                    )
+                    return
+
                 await self.ap.msg_aggregator.add_message(
                     bot_uuid=self.bot_entity.uuid,
                     launcher_type=provider_session.LauncherTypes.GROUP,
@@ -134,7 +313,8 @@ class RuntimeBot:
                     message_event=event,
                     message_chain=event.message_chain,
                     adapter=adapter,
-                    pipeline_uuid=self.bot_entity.use_pipeline_uuid,
+                    pipeline_uuid=pipeline_uuid,
+                    routed_by_rule=routed_by_rule,
                 )
             else:
                 await self.logger.info('Pipeline skipped for group message due to webhook response')

src/langbot/pkg/utils/constants.py

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