From 96280b3bcce213c7ab0d85bfaef4d809469c68d0 Mon Sep 17 00:00:00 2001 From: Hyu Date: Fri, 9 Oct 2026 21:22:34 +0800 Subject: [PATCH] fix(events): restrict pipeline routes and load agent prompts --- src/langbot/pkg/agent/runner/orchestrator.py | 9 +++++ src/langbot/pkg/api/http/service/agent.py | 2 +- src/langbot/pkg/api/http/service/bot.py | 4 +- .../versions/0035_pipeline_received_only.py | 39 +++++++++++++++++++ src/langbot/pkg/platform/botmgr.py | 10 ++++- .../bot-form/EventBindingsEditor.tsx | 2 +- web/src/i18n/locales/en-US.ts | 2 +- web/src/i18n/locales/es-ES.ts | 2 +- web/src/i18n/locales/ja-JP.ts | 2 +- web/src/i18n/locales/ru-RU.ts | 2 +- web/src/i18n/locales/th-TH.ts | 2 +- web/src/i18n/locales/vi-VN.ts | 2 +- web/src/i18n/locales/zh-Hans.ts | 2 +- web/src/i18n/locales/zh-Hant.ts | 2 +- 14 files changed, 69 insertions(+), 13 deletions(-) create mode 100644 src/langbot/pkg/persistence/alembic/versions/0035_pipeline_received_only.py diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index 74f862d79..5818902eb 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -5,9 +5,11 @@ from __future__ import annotations import time import asyncio import contextlib +import copy import typing from langbot_plugin.api.entities.builtin.provider import message as provider_message +from langbot_plugin.api.entities.builtin.provider import prompt as provider_prompt from langbot_plugin.api.entities.builtin.pipeline import query as pipeline_query from langbot_plugin.entities.io.actions.enums import PluginToRuntimeAction @@ -18,6 +20,7 @@ from ...pipeline.pool import get_query_execution_context from .binding_resolver import AgentBindingResolver from .context_builder import RunnerContextBuilder, RunnerContextPayload from .descriptor import RunnerDescriptor +from .config_schema import extract_prompt_config from .execution_context import ( append_mcp_resource_context_to_event, build_mcp_resource_context_addition, @@ -115,6 +118,12 @@ class AgentRunOrchestrator: if execution_query is None: execution_query = build_execution_query(event, []) + # Event runs bypass pipeline prompt preparation, just like debug runs. + prompt_config = extract_prompt_config(descriptor, binding.runner_config, []) + execution_query.prompt = provider_prompt.Prompt( + name='default', + messages=[provider_message.Message(**item) for item in copy.deepcopy(prompt_config)], + ) # Synthetic events must expose the same trusted scope as pipeline queries. for field_name in ('instance_uuid', 'workspace_uuid', 'placement_generation', 'query_uuid'): object.__setattr__(execution_query, field_name, getattr(execution_context, field_name)) diff --git a/src/langbot/pkg/api/http/service/agent.py b/src/langbot/pkg/api/http/service/agent.py index 2d626117a..20298dd12 100644 --- a/src/langbot/pkg/api/http/service/agent.py +++ b/src/langbot/pkg/api/http/service/agent.py @@ -47,7 +47,7 @@ from .tenant import TenantContext, require_workspace_uuid, scope_statement AGENT_KIND_AGENT = 'agent' AGENT_KIND_PIPELINE = 'pipeline' AGENT_KIND_EVENT_PROCESSOR = 'event_processor' -PIPELINE_EVENT_PATTERNS = ['message.*'] +PIPELINE_EVENT_PATTERNS = ['message.received'] AGENT_DEFAULT_EVENT_PATTERNS = ['*'] diff --git a/src/langbot/pkg/api/http/service/bot.py b/src/langbot/pkg/api/http/service/bot.py index 3d2ef6b13..1074c9d92 100644 --- a/src/langbot/pkg/api/http/service/bot.py +++ b/src/langbot/pkg/api/http/service/bot.py @@ -64,7 +64,7 @@ class BotService: @staticmethod def _is_message_event_pattern(event_pattern: str) -> bool: - return event_pattern == 'message.*' or event_pattern.startswith('message.') + return event_pattern == 'message.received' @staticmethod def _event_pattern_covers(supported_pattern: str, binding_pattern: str) -> bool: @@ -507,7 +507,7 @@ class BotService: if target_type == 'pipeline': if not self._is_message_event_pattern(event_pattern): - raise ValueError('Pipeline can only be bound to message events') + raise ValueError('Pipeline can only be bound to message.received') result = await self.ap.persistence_mgr.execute_async( scope_statement( sqlalchemy.select(persistence_pipeline.LegacyPipeline.uuid).where( diff --git a/src/langbot/pkg/persistence/alembic/versions/0035_pipeline_received_only.py b/src/langbot/pkg/persistence/alembic/versions/0035_pipeline_received_only.py new file mode 100644 index 000000000..146d0c754 --- /dev/null +++ b/src/langbot/pkg/persistence/alembic/versions/0035_pipeline_received_only.py @@ -0,0 +1,39 @@ +"""Narrow legacy Pipeline wildcard routes to incoming messages.""" + +import json + +from alembic import op +import sqlalchemy as sa + +revision = '0035_pipeline_received_only' +down_revision = '0034_monitoring_execution_links' +branch_labels = None +depends_on = None + + +def upgrade(): + connection = op.get_bind() + if not sa.inspect(connection).has_table('bots'): + return + bots = sa.table('bots', sa.column('uuid', sa.String()), sa.column('event_bindings', sa.JSON())) + for bot_uuid, bindings in connection.execute(sa.select(bots.c.uuid, bots.c.event_bindings)).all(): + if isinstance(bindings, str): + bindings = json.loads(bindings) + if not isinstance(bindings, list): + continue + changed = False + for binding in bindings: + if ( + isinstance(binding, dict) + and binding.get('target_type') == 'pipeline' + and binding.get('event_pattern') == 'message.*' + ): + binding['event_pattern'] = 'message.received' + changed = True + if changed: + connection.execute(bots.update().where(bots.c.uuid == bot_uuid).values(event_bindings=bindings)) + + +def downgrade(): + # An exact incoming-message route remains valid with older versions. + pass diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index 7b0bc8e08..8698a7399 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -158,7 +158,7 @@ class RuntimeBot: @classmethod def _is_message_event_type(cls, event_type: str) -> bool: - return cls._match_event_pattern(event_type, 'message.*') + return event_type == 'message.received' @classmethod def _agent_supports_event_type( @@ -310,6 +310,12 @@ class RuntimeBot: step['reason'] = 'Event type does not match binding event_pattern' diagnostic_steps.append(step) continue + if binding.get('target_type') == 'pipeline' and not cls._is_message_event_type(event_type): + step['failure_code'] = 'processor_incompatible' + step['reason'] = 'Pipeline targets only support message.received' + diagnostic_steps.append(step) + continue + if not cls._match_event_filters(event, binding.get('filters')): step['failure_code'] = 'filters_mismatch' step['reason'] = 'Event data does not satisfy binding filters' @@ -423,6 +429,8 @@ class RuntimeBot: def get_pipeline_target_for_event_type(self, event_type: str = 'message.received') -> str | None: """Return the first Pipeline target configured for an event type.""" + if not self._is_message_event_type(event_type): + return None matched: list[tuple[int, int, str]] = [] for index, binding in enumerate(self._get_event_bindings()): if not binding.get('enabled', True): 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 3c5618fea..56fbcba89 100644 --- a/web/src/app/home/bots/components/bot-form/EventBindingsEditor.tsx +++ b/web/src/app/home/bots/components/bot-form/EventBindingsEditor.tsx @@ -218,7 +218,7 @@ const BEHAVIOR_PRESETS = [ // ── helpers ─────────────────────────────────────────────────────────────────── function isMessageEventPattern(p: string) { - return p === 'message.*' || p.startsWith('message.'); + return p === 'message.received'; } interface RouteConflict { diff --git a/web/src/i18n/locales/en-US.ts b/web/src/i18n/locales/en-US.ts index d4d4409d6..43f3e93df 100644 --- a/web/src/i18n/locales/en-US.ts +++ b/web/src/i18n/locales/en-US.ts @@ -886,7 +886,7 @@ const enUS = { enabled: 'Enabled', eventBindingDescriptionPlaceholder: 'Rule description', noEventBindings: 'No event routes', - unsupportedPipelineEvent: 'Pipelines can only be used for message.* events', + unsupportedPipelineEvent: 'Pipelines can only be used for message.received events', disable: 'Disable', enable: 'Enable', disabledBindings: 'Disabled', diff --git a/web/src/i18n/locales/es-ES.ts b/web/src/i18n/locales/es-ES.ts index 3cdb9e435..6544f7d30 100644 --- a/web/src/i18n/locales/es-ES.ts +++ b/web/src/i18n/locales/es-ES.ts @@ -904,7 +904,7 @@ const esES = { enabled: 'Activado', eventBindingDescriptionPlaceholder: 'Descripción de la regla', noEventBindings: 'No hay rutas de eventos', - unsupportedPipelineEvent: 'Los flujos solo admiten eventos message.*', + unsupportedPipelineEvent: 'Los flujos solo admiten eventos message.received', disable: 'Desactivar', enable: 'Activar', disabledBindings: 'Desactivado', diff --git a/web/src/i18n/locales/ja-JP.ts b/web/src/i18n/locales/ja-JP.ts index bf771cdd5..62674fa9b 100644 --- a/web/src/i18n/locales/ja-JP.ts +++ b/web/src/i18n/locales/ja-JP.ts @@ -891,7 +891,7 @@ const jaJP = { eventBindingDescriptionPlaceholder: 'ルール説明', noEventBindings: 'イベントルートはありません', unsupportedPipelineEvent: - 'Pipeline は message.* イベントにのみ使用できます', + 'Pipeline は message.received イベントにのみ使用できます', disable: '無効化', enable: '有効化', disabledBindings: '無効', diff --git a/web/src/i18n/locales/ru-RU.ts b/web/src/i18n/locales/ru-RU.ts index fb1073449..27bcd17ec 100644 --- a/web/src/i18n/locales/ru-RU.ts +++ b/web/src/i18n/locales/ru-RU.ts @@ -899,7 +899,7 @@ const ruRU = { enabled: 'Включено', eventBindingDescriptionPlaceholder: 'Описание правила', noEventBindings: 'Нет маршрутов событий', - unsupportedPipelineEvent: 'Конвейеры поддерживают только события message.*', + unsupportedPipelineEvent: 'Конвейеры поддерживают только события message.received', disable: 'Отключить', enable: 'Включить', disabledBindings: 'Отключено', diff --git a/web/src/i18n/locales/th-TH.ts b/web/src/i18n/locales/th-TH.ts index 4e23a853c..a9b67fa94 100644 --- a/web/src/i18n/locales/th-TH.ts +++ b/web/src/i18n/locales/th-TH.ts @@ -873,7 +873,7 @@ const thTH = { enabled: 'เปิดใช้งาน', eventBindingDescriptionPlaceholder: 'คำอธิบายกฎ', noEventBindings: 'ไม่มีเส้นทางเหตุการณ์', - unsupportedPipelineEvent: 'ไปป์ไลน์ใช้ได้กับเหตุการณ์ message.* เท่านั้น', + unsupportedPipelineEvent: 'ไปป์ไลน์ใช้ได้กับเหตุการณ์ message.received เท่านั้น', disable: 'ปิดใช้งาน', enable: 'เปิดใช้งาน', disabledBindings: 'ปิดใช้งาน', diff --git a/web/src/i18n/locales/vi-VN.ts b/web/src/i18n/locales/vi-VN.ts index 13ffd9d9e..8c7f0358a 100644 --- a/web/src/i18n/locales/vi-VN.ts +++ b/web/src/i18n/locales/vi-VN.ts @@ -888,7 +888,7 @@ const viVN = { enabled: 'Đã bật', eventBindingDescriptionPlaceholder: 'Mô tả quy tắc', noEventBindings: 'Chưa có tuyến sự kiện', - unsupportedPipelineEvent: 'Pipeline chỉ hỗ trợ sự kiện message.*', + unsupportedPipelineEvent: 'Pipeline chỉ hỗ trợ sự kiện message.received', disable: 'Tắt', enable: 'Bật', disabledBindings: 'Đã tắt', diff --git a/web/src/i18n/locales/zh-Hans.ts b/web/src/i18n/locales/zh-Hans.ts index 6e2469f27..9a85ea073 100644 --- a/web/src/i18n/locales/zh-Hans.ts +++ b/web/src/i18n/locales/zh-Hans.ts @@ -833,7 +833,7 @@ const zhHans = { enabled: '启用', eventBindingDescriptionPlaceholder: '规则说明', noEventBindings: '暂无事件路由', - unsupportedPipelineEvent: '流水线 仅可用于 message.* 事件', + unsupportedPipelineEvent: '流水线 仅可用于 message.received 事件', disable: '禁用', enable: '启用', disabledBindings: '已禁用', diff --git a/web/src/i18n/locales/zh-Hant.ts b/web/src/i18n/locales/zh-Hant.ts index 2e6026f96..7e06ea8d4 100644 --- a/web/src/i18n/locales/zh-Hant.ts +++ b/web/src/i18n/locales/zh-Hant.ts @@ -832,7 +832,7 @@ const zhHant = { enabled: '啟用', eventBindingDescriptionPlaceholder: '規則說明', noEventBindings: '暫無事件路由', - unsupportedPipelineEvent: '流水線 僅可用於 message.* 事件', + unsupportedPipelineEvent: '流水線 僅可用於 message.received 事件', disable: '禁用', enable: '啟用', disabledBindings: '已禁用',