fix(events): restrict pipeline routes and load agent prompts

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))

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 = ['*']
 
 

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(

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

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):