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' ? (