From d13ae56ea59eb1966b71e6462d6d7dba1157d3b0 Mon Sep 17 00:00:00 2001 From: RockChinQ Date: Tue, 20 Oct 2026 21:22:53 +0800 Subject: [PATCH] fix: bot routing setup and busy-session response checks (#2728) Apply group response rules before busy-session steering. Initialize usable private-message and group-mention routes for new bots, including Wizard-created pipelines, and preserve member-configured rules across saves and retries. Fix empty pipeline configuration, refresh the processor sidebar after automatic creation, and show creation feedback. Align routing regressions and use Node 24 for the browser CI job. Correct the task's erroneous agent attribution while preserving every code tree and the original human contribution. Co-authored-by: multica-agent --- .github/workflows/frontend-tests.yml | 2 +- docs/qa-message-routing-acceptance.md | 82 ++++++ skills/skills/langbot-mcp-ops/SKILL.md | 16 ++ .../http/controller/groups/platform/bots.py | 32 +++ src/langbot/pkg/api/http/service/agent.py | 1 - src/langbot/pkg/api/http/service/bot.py | 242 +++++++++++++++++- src/langbot/pkg/api/http/service/pipeline.py | 2 +- src/langbot/pkg/api/mcp/server.py | 29 +++ src/langbot/pkg/pipeline/controller.py | 17 ++ .../pkg/pipeline/resprule/rules/random.py | 5 +- src/langbot/pkg/platform/botmgr.py | 24 +- .../templates/metadata/pipeline/trigger.yaml | 12 +- .../api/service/test_agent_service.py | 1 - .../api/service/test_bot_default_routes.py | 84 ++++++ .../api/service/test_bot_service.py | 8 +- .../api/service/test_pipeline_service.py | 5 +- tests/unit_tests/pipeline/test_controller.py | 33 +++ tests/unit_tests/pipeline/test_resprule.py | 14 + .../home/bots/components/bot-form/BotForm.tsx | 26 +- .../bot-form/EventBindingsEditor.tsx | 50 +++- .../bot-form/message-routing-template.ts | 53 ++++ web/src/app/wizard/bot-setup.ts | 64 +++++ web/src/app/wizard/page.tsx | 67 +++-- web/src/i18n/locales/en-US.ts | 8 +- web/src/i18n/locales/es-ES.ts | 7 +- web/src/i18n/locales/ja-JP.ts | 7 +- web/src/i18n/locales/ru-RU.ts | 7 +- web/src/i18n/locales/th-TH.ts | 7 +- web/src/i18n/locales/vi-VN.ts | 8 +- web/src/i18n/locales/zh-Hans.ts | 7 +- web/src/i18n/locales/zh-Hant.ts | 7 +- web/tests/e2e/crud-smoke.spec.ts | 44 ++-- .../e2e/wizard-platform-regressions.spec.ts | 20 +- .../unit/message-routing-template.test.mjs | 53 ++++ web/tests/unit/wizard-bot-setup.test.mjs | 101 ++++++++ 35 files changed, 1064 insertions(+), 81 deletions(-) create mode 100644 docs/qa-message-routing-acceptance.md create mode 100644 tests/unit_tests/api/service/test_bot_default_routes.py create mode 100644 web/src/app/home/bots/components/bot-form/message-routing-template.ts create mode 100644 web/src/app/wizard/bot-setup.ts create mode 100644 web/tests/unit/message-routing-template.test.mjs create mode 100644 web/tests/unit/wizard-bot-setup.test.mjs diff --git a/.github/workflows/frontend-tests.yml b/.github/workflows/frontend-tests.yml index 7bdeba5e5..7e9980846 100644 --- a/.github/workflows/frontend-tests.yml +++ b/.github/workflows/frontend-tests.yml @@ -26,7 +26,7 @@ jobs: - name: Setup Node.js uses: actions/setup-node@v4 with: - node-version: '25' + node-version: '24' - name: Install pnpm uses: pnpm/action-setup@v4 diff --git a/docs/qa-message-routing-acceptance.md b/docs/qa-message-routing-acceptance.md new file mode 100644 index 000000000..b54203c47 --- /dev/null +++ b/docs/qa-message-routing-acceptance.md @@ -0,0 +1,82 @@ +# Message routing acceptance + +Use an isolated worktree, data directory, ports and Plugin Runtime. Preserve the +member's database and bot configuration. Back up SQLite with its backup API before +repairing data; do not copy a live WAL database with a plain file copy. + +## Reproduction and repair + +The product processor form sends `POST /api/v1/agents` with `kind=pipeline`. +Previously `AgentService` forwarded `config={}`, bypassing the Pipeline service's +default-config branch. The row was inserted, then stage initialization raised +`KeyError: safety`. Repeated failed creates could leave empty-config rows. + +Pipeline creation now omits this empty override and the Pipeline service also +handles explicit `{}` as a request for defaults. Existing non-empty configs are +not replaced. For affected historical rows, back up first, inspect their config, +and repair only rows whose entire config is exactly `{}` with the version's +default Pipeline template. Preserve UUID, name and all other columns. Do not +silently reset partial or non-empty configurations. + +New bots previously had no routes if there was no Pipeline. First-bot setup now +creates one default Pipeline and reuses it, serialized within the instance. +Existing bots are never rewritten. The old message preset added one unfiltered +route; it now adds separate private and group-mention routes, with an explicit +boolean mention condition in the editor. Reapplying preserves manual targets and +disabled routes. + +## Acceptance sequence + +1. Use the real WebUI to create a Pipeline, verify success, refresh its page, and + inspect persisted `trigger`, `safety`, `ai`, `output` sections. Correlate the + creation request with the HTTP access log. +2. Create a separate test bot. Confirm that the two message routes already + exist. Expand conditions and verify the private/group split and boolean + `is_at_bot=true`. +3. On the test bot only, remove the generated routes, use the message template, + save, wait for success, then refresh. Click the template again and check that + it neither duplicates conditions nor replaces manual targets. +4. Connect a local OneBot Universal protocol simulator using its own token and + fake bot/group/user IDs. Never attach the simulator to a real QQ account. +5. Send a bounded sequence: unmentioned group, mention of another user, private, + mention of the bot. Capture per-case task counters, Core logs, and the WebUI + bot log/matching records. The first two must create no Pipeline tasks; the + latter two must route to the selected Pipeline. +6. Verify actual replies separately with an authorized model. A matched route + followed by a missing-model error proves routing, not successful AI replies. + Label local protocol simulation separately from real-platform acceptance. +7. Confirm original member rows/configs are unchanged, capture service ownership + and health, and keep the isolated dev services available for manual review. + +Use existing `skills/.env` conventions for machine-specific URLs. Apply resource +stop thresholds before bounded tests; do not treat a short idle sample as a soak +test or a health endpoint as feature acceptance. + +## Wizard regression coverage + +The Wizard creates a dedicated Pipeline before creating its disabled Bot, and +passes the same explicit private/group-mention template used by manual setup. +Returning to the same platform reuses the Bot and retains unsaved form values. +A successful Pipeline remains in the in-page draft if Bot creation fails, so a +retry reuses it. Create/save operations have a synchronous reentry guard. +Saving adapter settings reads the current Bot and only initializes routes when +none exist; it never replaces nonempty custom rules or re-enables disabled ones. + +For real UI acceptance, use a separate empty data directory and ports; never +reset the member's database or Wizard progress. Complete registration, platform +selection, save/enable, receipt of an isolated protocol message, AI choice and +finish. Refresh the completion page and compare persisted Bot/Pipeline UUIDs. +Return to the platform step and submit again; assert one Bot, one Pipeline and +unchanged route IDs. Remove/reapply the message template through the Bot UI and +repeat the private/unmentioned-group/other-mention/bot-mention protocol probes. +Disable a rule through the UI, restore a QA-only step-1 Wizard draft pointing to +the existing Bot/Pipeline, save through the UI, and compare every binding field. +Verify the disabled rule also stays inactive at runtime. + +The 2026-10-20 Mac run completed this flow against real Core/RT using native +Codex Computer Use. Both default and manual templates created tasks only for +private messages and mentions of this bot. After custom group-route disablement +and Wizard resave, even a mention created no task. Model execution failed with +missing credentials; successful AI replies and real long-running busy model +sessions are not certified by this run. Busy-session controller/response-rule +unit tests are a separate regression result, not a substitute for that evidence. diff --git a/skills/skills/langbot-mcp-ops/SKILL.md b/skills/skills/langbot-mcp-ops/SKILL.md index ecdc59759..b2009a7fd 100644 --- a/skills/skills/langbot-mcp-ops/SKILL.md +++ b/skills/skills/langbot-mcp-ops/SKILL.md @@ -180,6 +180,22 @@ state as well; it is not the same operation as disconnecting an account. - The global key is plaintext in config.yaml — only enable it on trusted/internal deployments and serve over HTTPS. +## Default message routes + +`create_bot` without non-empty `event_bindings` adds two explicit message routes: +private chat (`chat_type eq person`) and group chat (`chat_type eq group` plus +`is_at_bot eq true`). Both use the Workspace's default Pipeline, or the first +available Pipeline. If none exists, first-bot setup creates one with the full +default config. Creating a Pipeline via `create_agent(kind="pipeline")` also uses +that full default config. Configure an authorized model before testing replies. + +Explicit routes on create are preserved; `update_bot` never automatically +replaces manual routes, including an intentionally empty list. The WebUI message +template uses the same private/group conditions and preserves existing routes +when applied repeatedly. Unmentioned group messages do not enter the Pipeline. +An existing broad route can still match first; inspect route order and diagnostics +before changing a member's configuration. + ## Event processors Create a processor with `kind: "event_processor"` and basic information. Without diff --git a/src/langbot/pkg/api/http/controller/groups/platform/bots.py b/src/langbot/pkg/api/http/controller/groups/platform/bots.py index ad00d7888..f7824dc93 100644 --- a/src/langbot/pkg/api/http/controller/groups/platform/bots.py +++ b/src/langbot/pkg/api/http/controller/groups/platform/bots.py @@ -232,3 +232,35 @@ class BotsRouterGroup(group.RouterGroup): async def _(bot_uuid: str, admin_id: int, request_context: RequestContext) -> str: await self.ap.bot_service.delete_bot_admin(request_context, bot_uuid, admin_id) return self.success() + + @self.route( + '//group-filter-migration', + methods=['GET'], + auth_type=group.AuthType.USER_TOKEN_OR_API_KEY, + permission=Permission.RESOURCE_VIEW, + ) + async def _(bot_uuid: str, request_context: RequestContext) -> str: + try: + result = await self.ap.bot_service.preview_group_filter_migration(request_context, bot_uuid) + except ValueError as exc: + return self.http_status(400, -1, str(exc)) + return self.success(data=result) + + @self.route( + '//group-filter-migration', + methods=['POST'], + auth_type=group.AuthType.USER_TOKEN_OR_API_KEY, + permission=Permission.RESOURCE_MANAGE, + ) + async def _(bot_uuid: str, request_context: RequestContext) -> str: + json_data = await quart.request.get_json(silent=True) or {} + disable_pipeline_at_bot = bool(json_data.get('disable_pipeline_at_bot', True)) + try: + result = await self.ap.bot_service.apply_group_filter_migration( + request_context, bot_uuid, disable_pipeline_at_bot=disable_pipeline_at_bot + ) + except BotApplyError as exc: + return self._apply_error_response(exc) + except ValueError as exc: + return self.http_status(400, -1, str(exc)) + return self.success(data=result) diff --git a/src/langbot/pkg/api/http/service/agent.py b/src/langbot/pkg/api/http/service/agent.py index c0f0b34bf..442e1dcf7 100644 --- a/src/langbot/pkg/api/http/service/agent.py +++ b/src/langbot/pkg/api/http/service/agent.py @@ -437,7 +437,6 @@ class AgentService: 'name': agent_data.get('name') or t('New Pipeline'), 'description': agent_data.get('description') or '', 'emoji': agent_data.get('emoji') or '⚙️', - 'config': {}, }, ) return {'uuid': pipeline_uuid, 'kind': AGENT_KIND_PIPELINE} diff --git a/src/langbot/pkg/api/http/service/bot.py b/src/langbot/pkg/api/http/service/bot.py index 766adbc87..b9de1c1ba 100644 --- a/src/langbot/pkg/api/http/service/bot.py +++ b/src/langbot/pkg/api/http/service/bot.py @@ -3,6 +3,7 @@ from __future__ import annotations import uuid import typing import json +import asyncio import sqlalchemy from ....core import app @@ -18,6 +19,7 @@ from ....platform.sources import http_bot_signing from ....platform.adapter_names import canonical_adapter_name from ....cloud.plan_policy import require_workspace_platform from ....agent.runner.errors import RunnerNotFoundError +from ....i18n import t class BotService: @@ -43,6 +45,7 @@ class BotService: def __init__(self, ap: app.Application) -> None: self.ap = ap + self._default_pipeline_lock = asyncio.Lock() def _get_adapter_component(self, adapter_name: str) -> engine.Component | None: """Return the discovered platform adapter component for an adapter name.""" @@ -202,6 +205,22 @@ class BotService: 'diagnostic_details': diagnostic_steps or [], } + @staticmethod + def _extract_dry_run_bot_account_id( + context: typing.Any, + event_data: typing.Any, + ) -> str | None: + """Extract bot_account_id from dry-run context for is_at_bot filter support.""" + if isinstance(context, dict): + val = context.get('bot_account_id') + if val is not None: + return str(val) + if isinstance(event_data, dict): + val = event_data.get('bot_account_id') + if val is not None: + return str(val) + return None + @staticmethod def _build_dry_run_event(event_type: str, event_data: typing.Any, context: typing.Any) -> dict[str, typing.Any]: event: dict[str, typing.Any] = {} @@ -354,6 +373,7 @@ class BotService: bindings, event, event_type, + bot_account_id=self._extract_dry_run_bot_account_id(context, event_data), ) if selected_binding is None: return self._diagnostic_result( @@ -490,6 +510,60 @@ class BotService: ], ) + async def _build_default_event_bindings(self, context: TenantContext) -> list[dict]: + """Build the default event bindings for a new bot. + + Two bindings route message.received to the first available pipeline: + - private chat: always respond + - group chat: require an @mention of the bot (is_at_bot=true) + + The group @mention check runs at the routing layer, so messages that + do not match never create a pipeline task. Existing bots keep their + current bindings; this only applies to bots created without any. + """ + # Serialize first-bot setup without retaining an unbounded lock per Workspace. + async with self._default_pipeline_lock: + pipelines = await self.ap.pipeline_service.get_pipelines(context) + if pipelines: + selected = next((item for item in pipelines if item.get('is_default')), pipelines[0]) + pipeline_uuid = selected.get('uuid') + else: + pipeline_uuid = await self.ap.pipeline_service.create_pipeline( + context, {'name': t('New Pipeline'), 'description': '', 'emoji': '⚙️'}, default=True + ) + if not pipeline_uuid: + return [] + + return [ + { + 'id': str(uuid.uuid4()), + 'event_pattern': 'message.received', + 'target_type': 'pipeline', + 'target_uuid': pipeline_uuid, + 'filters': [ + {'field': 'chat_type', 'operator': 'eq', 'value': 'person'}, + ], + 'priority': 0, + 'enabled': True, + 'description': 'Private chat: respond directly', + 'order': 0, + }, + { + 'id': str(uuid.uuid4()), + 'event_pattern': 'message.received', + 'target_type': 'pipeline', + 'target_uuid': pipeline_uuid, + 'filters': [ + {'field': 'chat_type', 'operator': 'eq', 'value': 'group'}, + {'field': 'is_at_bot', 'operator': 'eq', 'value': True}, + ], + 'priority': 0, + 'enabled': True, + 'description': 'Group chat: respond only when the bot is @mentioned', + 'order': 1, + }, + ] + async def _normalize_event_bindings(self, context: TenantContext, bindings: list[dict] | None) -> list[dict]: """Validate and normalize Bot event bindings.""" if not bindings: @@ -700,7 +774,10 @@ class BotService: await require_workspace_platform(self.ap, workspace_uuid, bot_data['adapter']) bot_data['uuid'] = str(uuid.uuid4()) bot_data['workspace_uuid'] = workspace_uuid - bot_data.setdefault('event_bindings', []) + if not bot_data.get('event_bindings'): + bot_data['event_bindings'] = await self._build_default_event_bindings(context) + else: + bot_data['event_bindings'] = await self._normalize_event_bindings(context, bot_data['event_bindings']) bot_data.setdefault('plugin_processors', []) await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_bot.Bot).values(bot_data)) @@ -860,6 +937,169 @@ class BotService: 'stale_routes': stale_routes, } + async def preview_group_filter_migration(self, context: TenantContext, bot_uuid: str) -> dict[str, typing.Any]: + """Preview the effect of promoting the pipeline at-bot rule to a routing-layer filter. + + Detects bots whose pipeline trigger still requires at-bot, and whose event + bindings do not already filter group messages by is_at_bot. Returns the + proposed binding changes plus a per-pipeline summary. Does not persist. + """ + from ....platform.botmgr import RuntimeBot + + bot = await self.get_bot(context, bot_uuid, include_secret=False) + if bot is None: + raise WorkspaceNotFoundError('Bot not found') + + bindings = RuntimeBot._get_event_bindings_from_value(bot.get('event_bindings') or []) + pipeline_uuids = { + str(b.get('target_uuid')) for b in bindings if b.get('target_type') == 'pipeline' and b.get('target_uuid') + } + + # Load each pipeline's trigger config to detect at-bot still enabled. + pipeline_flags: dict[str, dict[str, typing.Any]] = {} + for pipeline_uuid in pipeline_uuids: + pipeline = await self.ap.pipeline_service.get_pipeline(context, pipeline_uuid) + trigger = (pipeline or {}).get('config', {}).get('trigger', {}) if pipeline else {} + group_rules = trigger.get('group-respond-rules', {}) if isinstance(trigger, dict) else {} + pipeline_flags[pipeline_uuid] = { + 'at_bot_enabled': bool(group_rules.get('at')) if isinstance(group_rules, dict) else False, + 'prefix': group_rules.get('prefix', []) if isinstance(group_rules, dict) else [], + 'regexp': group_rules.get('regexp', []) if isinstance(group_rules, dict) else [], + 'random': group_rules.get('random', 0.0) if isinstance(group_rules, dict) else 0.0, + } + + # A binding "already has the at-bot filter" if it filters chat_type=group + # AND requires is_at_bot=true (or non-false). + def _has_group_at_filter(binding: dict) -> bool: + filters = binding.get('filters') or [] + if not isinstance(filters, list): + return False + has_group = any( + isinstance(f, dict) + and f.get('field') == 'chat_type' + and f.get('operator', 'eq') == 'eq' + and f.get('value') == 'group' + for f in filters + ) + has_at = any( + isinstance(f, dict) + and f.get('field') == 'is_at_bot' + and f.get('operator', 'eq') == 'eq' + and f.get('value') is True + for f in filters + ) + return has_group and has_at + + needs_migration = [pipeline_uuid for pipeline_uuid, flags in pipeline_flags.items() if flags['at_bot_enabled']] + bindings_with_at_filter = [b for b in bindings if _has_group_at_filter(b)] + bindings_missing_at_filter = [ + b + for b in bindings + if b.get('target_type') == 'pipeline' + and str(b.get('target_uuid') or '') in needs_migration + and not _has_group_at_filter(b) + ] + + proposed_bindings: list[dict[str, typing.Any]] = [] + for binding in bindings_missing_at_filter: + new_binding = json.loads(json.dumps(binding)) + filters = list(new_binding.get('filters') or []) + if not any( + isinstance(f, dict) and f.get('field') == 'chat_type' and f.get('value') == 'group' for f in filters + ): + filters.append({'field': 'chat_type', 'operator': 'eq', 'value': 'group'}) + if not any(isinstance(f, dict) and f.get('field') == 'is_at_bot' for f in filters): + filters.append({'field': 'is_at_bot', 'operator': 'eq', 'value': True}) + new_binding['filters'] = filters + proposed_bindings.append(new_binding) + + return { + 'bot_uuid': bot_uuid, + 'needs_migration': bool(bindings_missing_at_filter), + 'pipelines_with_at_bot': sorted(needs_migration), + 'pipeline_flags': pipeline_flags, + 'bindings_already_filtered': [b.get('id') for b in bindings_with_at_filter], + 'bindings_to_update': [ + { + 'binding_id': b.get('id'), + 'event_pattern': b.get('event_pattern'), + 'target_type': b.get('target_type'), + 'target_uuid': b.get('target_uuid'), + 'current_filters': b.get('filters') or [], + 'proposed_filters': next( + (p.get('filters') for p in proposed_bindings if p.get('id') == b.get('id')), + b.get('filters') or [], + ), + } + for b in bindings_missing_at_filter + ], + } + + async def apply_group_filter_migration( + self, + context: TenantContext, + bot_uuid: str, + *, + disable_pipeline_at_bot: bool = True, + ) -> dict[str, typing.Any]: + """Apply the group filter migration previewed by preview_group_filter_migration. + + Updates the bot's event_bindings to add is_at_bot=true filter on group + bindings whose pipeline still has at-bot enabled. Optionally disables the + pipeline's at-bot trigger rule so the filter is not checked twice. Existing + prefix/regexp/random rules remain untouched. Idempotent. + """ + from ....platform.botmgr import RuntimeBot + + preview = await self.preview_group_filter_migration(context, bot_uuid) + if not preview['needs_migration']: + return {'bot_uuid': bot_uuid, 'applied': False, 'reason': 'no_migration_needed'} + + bot = await self.get_bot(context, bot_uuid, include_secret=False) + if bot is None: + raise WorkspaceNotFoundError('Bot not found') + + bindings = RuntimeBot._get_event_bindings_from_value(bot.get('event_bindings') or []) + bindings_by_id = {b.get('id'): b for b in bindings if b.get('id')} + for item in preview['bindings_to_update']: + binding_id = item['binding_id'] + existing = bindings_by_id.get(binding_id) + if existing is None: + continue + existing['filters'] = item['proposed_filters'] + + update_payload = { + 'event_bindings': await self._normalize_event_bindings(context, bindings), + } + await self.update_bot(context, bot_uuid, update_payload) + + updated_pipelines: list[str] = [] + if disable_pipeline_at_bot: + for pipeline_uuid in preview['pipelines_with_at_bot']: + pipeline = await self.ap.pipeline_service.get_pipeline(context, pipeline_uuid, include_secret=True) + if not pipeline: + continue + config = pipeline.get('config') or {} + trigger = config.get('trigger') or {} + group_rules = trigger.get('group-respond-rules') or {} + if not isinstance(group_rules, dict) or not group_rules.get('at'): + continue + new_config = json.loads(json.dumps(config)) + new_config.setdefault('trigger', {}).setdefault('group-respond-rules', {})['at'] = False + await self.ap.pipeline_service.update_pipeline( + context, + pipeline_uuid, + {'config': new_config}, + ) + updated_pipelines.append(pipeline_uuid) + + return { + 'bot_uuid': bot_uuid, + 'applied': True, + 'updated_bindings': [item['binding_id'] for item in preview['bindings_to_update']], + 'disabled_pipeline_at_bot': updated_pipelines, + } + async def send_http_bot_test_message( self, context: TenantContext, diff --git a/src/langbot/pkg/api/http/service/pipeline.py b/src/langbot/pkg/api/http/service/pipeline.py index 4316ee864..6df59edbf 100644 --- a/src/langbot/pkg/api/http/service/pipeline.py +++ b/src/langbot/pkg/api/http/service/pipeline.py @@ -189,7 +189,7 @@ class PipelineService: pipeline_data['stages'] = default_stage_order.copy() pipeline_data['is_default'] = default - if 'config' not in pipeline_data: + if 'config' not in pipeline_data or pipeline_data['config'] == {}: pipeline_data['config'] = await self.get_default_pipeline_config(context) RunnerConfigResolver.validate_pipeline_config(pipeline_data['config']) diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py index 2e3df5519..27fcfca19 100644 --- a/src/langbot/pkg/api/mcp/server.py +++ b/src/langbot/pkg/api/mcp/server.py @@ -387,6 +387,35 @@ class LangBotMCPServer: async def list_bot_event_route_statuses(bot_uuid: str) -> str: return _dump(await ap.bot_service.list_event_route_statuses(bot_uuid)) + @mcp.tool( + description=( + 'Preview promoting the pipeline at-bot trigger rule to a routing-layer ' + 'is_at_bot filter. Reports which bindings and pipelines need migration ' + 'without persisting changes.' + ) + ) + async def preview_bot_group_filter_migration(bot_uuid: str) -> str: + context = _authorized(Permission.RESOURCE_VIEW) + return _dump(await ap.bot_service.preview_group_filter_migration(context, bot_uuid)) + + @mcp.tool( + description=( + 'Apply the group filter migration previewed by ' + 'preview_bot_group_filter_migration. Adds is_at_bot=true filter to group ' + 'message bindings and, when disable_pipeline_at_bot is true (default), ' + 'turns off the pipeline trigger at-bot rule. Idempotent.' + ) + ) + async def apply_bot_group_filter_migration(bot_uuid: str, disable_pipeline_at_bot: bool = True) -> str: + context = _authorized(Permission.RESOURCE_MANAGE) + return _dump( + await ap.bot_service.apply_group_filter_migration( + context, + bot_uuid, + disable_pipeline_at_bot=disable_pipeline_at_bot, + ) + ) + # ----- Pipelines ----------------------------------------------- # @mcp.tool(description='List all pipelines.') async def list_pipelines() -> str: diff --git a/src/langbot/pkg/pipeline/controller.py b/src/langbot/pkg/pipeline/controller.py index 11ebeae09..bb042b591 100644 --- a/src/langbot/pkg/pipeline/controller.py +++ b/src/langbot/pkg/pipeline/controller.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import copy import traceback from ..core import app @@ -59,6 +60,22 @@ class Controller: query.pipeline_config = pipeline.pipeline_entity.config query.variables['_pipeline_bound_plugins'] = pipeline.bound_plugins query.variables['_pipeline_bound_mcp_servers'] = pipeline.bound_mcp_servers + if getattr(getattr(query, 'launcher_type', None), 'value', None) == 'group': + from .resprule.resprule import GroupRespondRuleCheckStage + from .entities import ResultType + + candidate = copy.copy(query) + candidate.message_chain = query.message_chain.model_copy(deep=True) + candidate.variables = dict(query.variables) + gate = GroupRespondRuleCheckStage(self.ap) + await gate.initialize(query.pipeline_config) + result = await gate.process(candidate, 'GroupRespondRuleCheckStage') + random_key = '_group_response_random_draw' + if random_key in candidate.variables: + query.variables[random_key] = candidate.variables[random_key] + if result.result_type == ResultType.INTERRUPT: + return False + query = candidate return await self.ap.agent_run_orchestrator.try_claim_steering_from_query(query) except Exception as exc: self.ap.logger.warning( diff --git a/src/langbot/pkg/pipeline/resprule/rules/random.py b/src/langbot/pkg/pipeline/resprule/rules/random.py index 2bfe8b71c..29d825b0b 100644 --- a/src/langbot/pkg/pipeline/resprule/rules/random.py +++ b/src/langbot/pkg/pipeline/resprule/rules/random.py @@ -18,4 +18,7 @@ class RandomRespRule(rule_model.GroupRespondRule): ) -> entities.RuleJudgeResult: random_rate = rule_dict['random'] - return entities.RuleJudgeResult(matching=random.random() < random_rate, replacement=message_chain) + key = '_group_response_random_draw' + if key not in query.variables: + query.variables[key] = random.random() + return entities.RuleJudgeResult(matching=query.variables[key] < random_rate, replacement=message_chain) diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index e89adacb2..cd3e150fe 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -229,6 +229,7 @@ class RuntimeBot: def _augment_event_data( cls, event_data: dict[str, typing.Any], + bot_account_id: str | None = None, ) -> dict[str, typing.Any]: """Inject virtual computed fields to simplify common filter patterns.""" message_chain = event_data.get('message_chain') @@ -240,6 +241,13 @@ class RuntimeBot: event_data['message_element_types'] = [ comp.get('type', '') for comp in message_chain if isinstance(comp, dict) ] + if bot_account_id: + event_data['is_at_bot'] = any( + isinstance(comp, dict) + and comp.get('type') == 'At' + and str(comp.get('target', '')) == str(bot_account_id) + for comp in message_chain + ) if 'group' in event_data: event_data['chat_type'] = 'group' if event_data.get('group') is not None else 'person' return event_data @@ -249,13 +257,14 @@ class RuntimeBot: cls, event: platform_events.EBAEvent, filters: typing.Any, + bot_account_id: str | None = None, ) -> bool: if not filters: return True if not isinstance(filters, list): return False - event_data = cls._augment_event_data(cls._safe_model_dump(event)) + event_data = cls._augment_event_data(cls._safe_model_dump(event), bot_account_id=bot_account_id) return all( cls._match_event_filter(event_data, event_filter) for event_filter in filters @@ -283,6 +292,7 @@ class RuntimeBot: bindings: list[dict[str, typing.Any]], event: platform_events.EBAEvent, event_type: str, + bot_account_id: str | None = None, ) -> tuple[dict[str, typing.Any] | None, list[dict[str, typing.Any]]]: """Evaluate Bot event bindings with the same precedence used at runtime.""" matched: list[tuple[int, int, dict[str, typing.Any], dict[str, typing.Any]]] = [] @@ -325,7 +335,7 @@ class RuntimeBot: diagnostic_steps.append(step) continue - if not cls._match_event_filters(event, binding.get('filters')): + if not cls._match_event_filters(event, binding.get('filters'), bot_account_id=bot_account_id): step['failure_code'] = 'filters_mismatch' step['reason'] = 'Event data does not satisfy binding filters' diagnostic_steps.append(step) @@ -356,7 +366,10 @@ class RuntimeBot: event_type: str, ) -> dict[str, typing.Any] | None: """Resolve the highest priority Bot event binding for a platform event.""" - selected, _ = self._evaluate_eba_event_bindings(self._get_event_bindings(), event, event_type) + bot_account_id = getattr(getattr(self, 'adapter', None), 'bot_account_id', None) or None + selected, _ = self._evaluate_eba_event_bindings( + self._get_event_bindings(), event, event_type, bot_account_id=bot_account_id + ) return selected def diagnose_eba_event_binding( @@ -365,7 +378,10 @@ class RuntimeBot: event_type: str, ) -> tuple[dict[str, typing.Any] | None, list[dict[str, typing.Any]]]: """Return the selected event binding plus per-binding diagnostic steps.""" - return self._evaluate_eba_event_bindings(self._get_event_bindings(), event, event_type) + bot_account_id = getattr(getattr(self, 'adapter', None), 'bot_account_id', None) or None + return self._evaluate_eba_event_bindings( + self._get_event_bindings(), event, event_type, bot_account_id=bot_account_id + ) @staticmethod def _route_ref( diff --git a/src/langbot/templates/metadata/pipeline/trigger.yaml b/src/langbot/templates/metadata/pipeline/trigger.yaml index 12a1e3dd2..6082a7a29 100644 --- a/src/langbot/templates/metadata/pipeline/trigger.yaml +++ b/src/langbot/templates/metadata/pipeline/trigger.yaml @@ -16,8 +16,16 @@ stages: en_US: At zh_Hans: '@' description: - en_US: Whether to trigger when the message mentions the bot - zh_Hans: 是否在消息@机器人时触发 + en_US: >- + Whether to trigger when the message @-mentions the bot (fallback filter + inside the pipeline). For new bots this check is handled earlier at the + routing layer via the is_at_bot filter — keeping it enabled here applies + it a second time. Use the group-filter migration assistant to move the + check to the routing layer and disable this option. + zh_Hans: >- + 是否仅响应 @机器人 的群消息（流水线内部过滤，兜底用）。新建机器人的路由层已通过 + is_at_bot 条件提前过滤，再启用此项会重复检查。建议使用群消息过滤迁移助手将过滤 + 前移到路由层后，再关闭此开关。 type: boolean required: true default: false diff --git a/tests/unit_tests/api/service/test_agent_service.py b/tests/unit_tests/api/service/test_agent_service.py index f3916d829..bca8b3a9c 100644 --- a/tests/unit_tests/api/service/test_agent_service.py +++ b/tests/unit_tests/api/service/test_agent_service.py @@ -778,7 +778,6 @@ class TestAgentServiceCreateUpdateDelete: 'name': 'Pipeline Agent', 'description': 'Legacy pipeline', 'emoji': 'P', - 'config': {}, }, ) app.pipeline_service.update_pipeline.assert_awaited_once_with( diff --git a/tests/unit_tests/api/service/test_bot_default_routes.py b/tests/unit_tests/api/service/test_bot_default_routes.py new file mode 100644 index 000000000..5aaaacc28 --- /dev/null +++ b/tests/unit_tests/api/service/test_bot_default_routes.py @@ -0,0 +1,84 @@ +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from langbot.pkg.api.http.service.bot import BotService + + +async def test_concurrent_first_bots_share_one_default_pipeline(): + pipelines = [] + + async def create(context, data, default=False): + await asyncio.sleep(0) + assert default + pipelines.append({'uuid': 'p', 'is_default': True}) + return 'p' + + ap = SimpleNamespace( + pipeline_service=SimpleNamespace( + get_pipelines=AsyncMock(side_effect=lambda context: list(pipelines)), + create_pipeline=AsyncMock(side_effect=create), + ) + ) + service = BotService(ap) + first, second = await asyncio.gather( + service._build_default_event_bindings('workspace'), + service._build_default_event_bindings('workspace'), + ) + ap.pipeline_service.create_pipeline.assert_awaited_once() + for bindings in (first, second): + assert len(bindings) == 2 + assert all(binding['target_uuid'] == 'p' for binding in bindings) + assert bindings[0]['filters'] == [{'field': 'chat_type', 'operator': 'eq', 'value': 'person'}] + assert bindings[1]['filters'][-1] == {'field': 'is_at_bot', 'operator': 'eq', 'value': True} + + +async def test_existing_default_pipeline_preferred_without_creating_another(): + ap = SimpleNamespace( + pipeline_service=SimpleNamespace( + get_pipelines=AsyncMock( + return_value=[ + {'uuid': 'recent'}, + {'uuid': 'default', 'is_default': True}, + ] + ), + create_pipeline=AsyncMock(), + ) + ) + routes = await BotService(ap)._build_default_event_bindings('workspace') + assert all(route['target_uuid'] == 'default' for route in routes) + ap.pipeline_service.create_pipeline.assert_not_awaited() + + +async def test_explicit_manual_route_does_not_create_or_replace_defaults(): + ap = SimpleNamespace( + instance_config=SimpleNamespace(data={}), + persistence_mgr=SimpleNamespace(execute_async=AsyncMock()), + pipeline_service=SimpleNamespace(get_pipelines=AsyncMock(), create_pipeline=AsyncMock()), + ) + service = BotService(ap) + service.get_bot = AsyncMock(return_value={'enable': False}) + route = { + 'id': 'manual-route', + 'event_pattern': 'message.received', + 'target_type': 'discard', + 'target_uuid': '', + 'filters': [{'field': 'chat_id', 'operator': 'eq', 'value': 'manual-group'}], + 'priority': 10, + 'enabled': False, + 'description': 'Keep this manual rule', + 'order': 0, + } + await service.create_bot( + 'workspace', + { + 'name': 'Manual', + 'adapter': 'onebot-v11', + 'adapter_config': {}, + 'event_bindings': [route], + }, + ) + inserted = ap.persistence_mgr.execute_async.await_args_list[0].args[0].compile().params + assert inserted['event_bindings'] == [route] + ap.pipeline_service.get_pipelines.assert_not_awaited() + ap.pipeline_service.create_pipeline.assert_not_awaited() diff --git a/tests/unit_tests/api/service/test_bot_service.py b/tests/unit_tests/api/service/test_bot_service.py index 574f28d7b..187635475 100644 --- a/tests/unit_tests/api/service/test_bot_service.py +++ b/tests/unit_tests/api/service/test_bot_service.py @@ -364,6 +364,7 @@ class TestBotServiceCreateBot: return_value={'uuid': 'uuid-1', 'name': 'Bot 1', 'adapter': 'telegram'} ) + ap.pipeline_service = SimpleNamespace(get_pipelines=AsyncMock(return_value=[{'uuid': 'default-pipeline'}])) service = BotService(ap) # Execute & Verify @@ -409,6 +410,7 @@ class TestBotServiceCreateBot: return_value={'uuid': 'new-uuid', 'name': 'New Bot', 'adapter': 'telegram', 'enable': False} ) + ap.pipeline_service = SimpleNamespace(get_pipelines=AsyncMock(return_value=[{'uuid': 'default-pipeline'}])) service = BotService(ap) # Execute @@ -429,10 +431,12 @@ class TestBotServiceCreateBot: 'name': 'New Bot', 'adapter': 'telegram', 'adapter_config': {}, - 'event_bindings': [], 'plugin_processors': [], } assert {key: values.get(key) for key in expected} == expected + assert len(values['event_bindings']) == 2 + assert all(b['target_uuid'] == 'default-pipeline' for b in values['event_bindings']) + assert values['event_bindings'][1]['filters'][-1] == {'field': 'is_at_bot', 'operator': 'eq', 'value': True} ap.platform_mgr.load_bot.assert_not_awaited() async def test_create_bot_scopes_insert_without_legacy_pipeline_fields(self): @@ -467,6 +471,7 @@ class TestBotServiceCreateBot: } ) + ap.pipeline_service = SimpleNamespace(get_pipelines=AsyncMock(return_value=[{'uuid': 'default-pipeline'}])) service = BotService(ap) # Execute @@ -509,6 +514,7 @@ class TestBotServiceCreateBot: ) ap.persistence_mgr = PersistenceManager(ap) ap.persistence_mgr.db = SimpleNamespace(get_engine=lambda: engine) + ap.pipeline_service = SimpleNamespace(get_pipelines=AsyncMock(return_value=[{'uuid': 'default-pipeline'}])) service = BotService(ap) try: async with engine.begin() as connection: diff --git a/tests/unit_tests/api/service/test_pipeline_service.py b/tests/unit_tests/api/service/test_pipeline_service.py index c1d65ede5..6f324c9b8 100644 --- a/tests/unit_tests/api/service/test_pipeline_service.py +++ b/tests/unit_tests/api/service/test_pipeline_service.py @@ -249,7 +249,8 @@ class TestPipelineServiceCreatePipeline: with pytest.raises(ValueError, match='Maximum number of pipelines'): await service.create_pipeline(WORKSPACE_UUID, {'name': 'New Pipeline'}) - async def test_create_pipeline_no_limit(self): + @pytest.mark.parametrize('config_fields', [{}, {'config': {}}]) + async def test_create_pipeline_no_limit(self, config_fields): """Creates pipeline without limit when max_pipelines=-1.""" # Setup ap = SimpleNamespace() @@ -276,7 +277,7 @@ class TestPipelineServiceCreatePipeline: with patch( 'langbot.pkg.utils.paths.get_resource_path', return_value='templates/default-pipeline-config.json' ): - bot_uuid = await service.create_pipeline(WORKSPACE_UUID, {'name': 'New Pipeline'}) + bot_uuid = await service.create_pipeline(WORKSPACE_UUID, {'name': 'New Pipeline', **config_fields}) # Verify assert bot_uuid is not None diff --git a/tests/unit_tests/pipeline/test_controller.py b/tests/unit_tests/pipeline/test_controller.py index b0b10a4d6..19f4e7f7a 100644 --- a/tests/unit_tests/pipeline/test_controller.py +++ b/tests/unit_tests/pipeline/test_controller.py @@ -81,6 +81,39 @@ async def test_try_claim_steering_sets_pipeline_context_before_claiming(): app.agent_run_orchestrator.try_claim_steering_from_query.assert_awaited_once_with(query) +@pytest.mark.asyncio +@pytest.mark.parametrize('text,expected', [('ordinary chat', False), ('qa: hello', True)]) +@pytest.mark.parametrize('claim_result', [True, False]) +async def test_busy_group_steering_obeys_response_rules(sample_query, text, expected, claim_result): + from langbot_plugin.api.entities.builtin.provider.session import LauncherTypes, Session + from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain + + app = make_app() + app.sess_mgr.get_session.return_value = Session( + instance_uuid='test-instance', + workspace_uuid='test-workspace', + placement_generation=1, + launcher_type=LauncherTypes.GROUP, + launcher_id=sample_query.launcher_id, + ) + pipeline = make_pipeline() + pipeline.pipeline_entity.config['trigger'] = { + 'group-respond-rules': {'at': True, 'prefix': ['qa:'], 'regexp': [], 'random': 0} + } + app.pipeline_mgr.get_pipeline_by_uuid.return_value = pipeline + app.agent_run_orchestrator.try_claim_steering_from_query.return_value = claim_result + sample_query.launcher_type = LauncherTypes.GROUP + sample_query.message_chain = MessageChain([Plain(text=text)]) + claimed = await Controller(app)._try_claim_steering_before_session_slot(sample_query) + assert claimed is (expected and claim_result), app.logger.warning.call_args + assert str(sample_query.message_chain) == text + if expected: + forwarded = app.agent_run_orchestrator.try_claim_steering_from_query.await_args.args[0] + assert str(forwarded.message_chain) == ' hello' + else: + app.agent_run_orchestrator.try_claim_steering_from_query.assert_not_awaited() + + @pytest.mark.asyncio async def test_consumer_transfers_query_from_queue_to_running_task(): app = make_app() diff --git a/tests/unit_tests/pipeline/test_resprule.py b/tests/unit_tests/pipeline/test_resprule.py index 4032ad6ff..b66814e31 100644 --- a/tests/unit_tests/pipeline/test_resprule.py +++ b/tests/unit_tests/pipeline/test_resprule.py @@ -65,6 +65,20 @@ async def test_group_message_no_match(mock_app, sample_query): mock_rule.match.assert_called_once() +@pytest.mark.asyncio +async def test_random_response_draw_is_reused_across_queue_rechecks(sample_query, mock_app, monkeypatch): + from langbot.pkg.pipeline.resprule.rules.random import RandomRespRule + from unittest.mock import Mock + + draw = Mock(side_effect=[0.9, 0.1]) + monkeypatch.setattr('langbot.pkg.pipeline.resprule.rules.random.random.random', draw) + matcher = RandomRespRule(mock_app) + for _ in range(3): + result = await matcher.match('hello', sample_query.message_chain, {'random': 0.5}, sample_query) + assert result.matching is False + draw.assert_called_once() + + @pytest.mark.asyncio async def test_group_message_match(mock_app, sample_query): """Test group message with matching rule""" diff --git a/web/src/app/home/bots/components/bot-form/BotForm.tsx b/web/src/app/home/bots/components/bot-form/BotForm.tsx index 0ae7d6dff..1e142da11 100644 --- a/web/src/app/home/bots/components/bot-form/BotForm.tsx +++ b/web/src/app/home/bots/components/bot-form/BotForm.tsx @@ -23,6 +23,7 @@ import { UUID } from 'uuidjs'; import DynamicFormComponent from '@/app/home/components/dynamic-form/DynamicFormComponent'; import { httpClient } from '@/app/infra/http/HttpClient'; import { systemInfo } from '@/app/infra/http'; +import { useSidebarData } from '@/app/home/components/home-sidebar/SidebarDataContext'; import { Agent, Bot } from '@/app/infra/entities/api'; import { getAdapterDocUrl } from '@/app/infra/entities/adapter-docs'; import { @@ -220,6 +221,7 @@ const BotForm = forwardRef(function BotForm( ref, ) { const { t } = useTranslation(); + const { refreshPipelines } = useSidebarData(); const formSchema = getFormSchema(t); const form = useForm>({ @@ -668,8 +670,30 @@ const BotForm = forwardRef(function BotForm( }; try { + // A failed metadata refresh must not prevent creating the bot. + const before = await httpClient.getPipelines().catch(() => null); const res = await httpClient.createBot(newBot); - toast.success(t('bots.createSuccess')); + const [created] = await Promise.all([ + httpClient.getBot(res.uuid).catch(() => null), + refreshPipelines(), + ]); + const automaticallyCreatedPipeline = + before !== null && + created?.bot.event_bindings?.some( + (binding) => + binding.event_pattern === 'message.received' && + binding.target_type === 'pipeline' && + !before.pipelines.some( + (pipeline) => pipeline.uuid === binding.target_uuid, + ), + ); + toast.success( + t( + automaticallyCreatedPipeline + ? 'bots.autoCreatedPipeline' + : 'bots.createSuccess', + ), + ); onNewBotCreated(res.uuid); } catch (error) { const err = error as CustomApiError & { diff --git a/web/src/app/home/bots/components/bot-form/EventBindingsEditor.tsx b/web/src/app/home/bots/components/bot-form/EventBindingsEditor.tsx index 06fcae3d7..178e5324d 100644 --- a/web/src/app/home/bots/components/bot-form/EventBindingsEditor.tsx +++ b/web/src/app/home/bots/components/bot-form/EventBindingsEditor.tsx @@ -12,6 +12,7 @@ import { useTranslation } from 'react-i18next'; import { Link } from 'react-router-dom'; import { cn } from '@/lib/utils'; import styles from './EventBindingsEditor.module.css'; +import { appendMessageRoutingTemplate } from './message-routing-template'; import type { TFunction } from 'i18next'; import { UseFormReturn } from 'react-hook-form'; import { @@ -144,17 +145,22 @@ interface EventBindingsEditorProps { } type FilterField = - 'chat_type' | 'chat_id' | 'message_text' | 'message_element_types'; + | 'chat_type' + | 'chat_id' + | 'message_text' + | 'message_element_types' + | 'is_at_bot'; type FilterOperator = 'eq' | 'neq' | 'contains' | 'not_contains' | 'starts_with' | 'regex'; interface FilterRow { field: FilterField; operator: FilterOperator; - value: string; + value: string | boolean; } const OPERATORS_BY_FIELD: Record = { + is_at_bot: ['eq', 'neq'], chat_type: ['eq', 'neq'], chat_id: ['eq', 'neq', 'contains', 'not_contains', 'regex'], message_text: [ @@ -443,7 +449,7 @@ function samplePayloadForEvent(eventType: string): Record { if (eventType === 'message.received') { return { message_text: 'Hello', - chat_type: 'private', + chat_type: 'person', chat_id: 'test-user', user_id: 'test-user', }; @@ -727,7 +733,7 @@ function FilterConditionsPanel({ next[i] = { ...next[i], ...patch }; if ('field' in patch) { next[i].operator = OPERATORS_BY_FIELD[patch.field as FilterField][0]; - next[i].value = ''; + next[i].value = patch.field === 'is_at_bot' ? true : ''; } onChange(next); } @@ -755,6 +761,9 @@ function FilterConditionsPanel({ + + {t('bots.filterIsAtBot')} + {t('bots.filterChatType')} @@ -784,9 +793,22 @@ function FilterConditionsPanel({ - {f.field === 'chat_type' ? ( + {f.field === 'is_at_bot' ? ( + ) : f.field === 'chat_type' ? ( + ) : f.field === 'message_element_types' ? (