feat(bots): bind plugin processor configurations independently

src/langbot/pkg/api/http/service/bot.py

@@ -16,6 +16,7 @@ from .tenant import TenantContext, require_workspace_uuid, scope_statement
 from ....utils import httpclient
 from ....platform.sources import http_bot_signing
 from ....platform.adapter_names import canonical_adapter_name
+from ....agent.runner.errors import RunnerNotFoundError
 
 
 class BotService:
@@ -36,6 +37,7 @@ class BotService:
         'adapter_config',
         'enable',
         'event_bindings',
+        'plugin_processors',
     }
 
     def __init__(self, ap: app.Application) -> None:
@@ -517,7 +519,7 @@ class BotService:
                 )
                 if result.first() is None:
                     raise ValueError('Pipeline not found')
-            elif target_type in {'agent', 'event_processor'}:
+            elif target_type == 'agent':
                 result = await self.ap.persistence_mgr.execute_async(
                     scope_statement(
                         sqlalchemy.select(persistence_agent.Agent).where(persistence_agent.Agent.uuid == target_uuid),
@@ -551,6 +553,37 @@ class BotService:
 
         return normalized
 
+    async def _normalize_plugin_processors(self, context: TenantContext, subscriptions: typing.Any) -> list[dict]:
+        """Validate explicit subscriptions within this Workspace; events come from the Runner."""
+        if not isinstance(subscriptions, list):
+            raise ValueError('plugin_processors must be an array')
+        normalized = []
+        seen = set()
+        for subscription in subscriptions:
+            if not isinstance(subscription, dict):
+                raise ValueError('Each plugin processor binding must be an object')
+            processor_uuid = subscription.get('processor_uuid')
+            if not isinstance(processor_uuid, str) or not processor_uuid.strip():
+                raise ValueError('Plugin processor UUID is required')
+            if processor_uuid in seen:
+                raise ValueError('A plugin processor can only be bound once per bot')
+            enabled = subscription.get('enabled', True)
+            if not isinstance(enabled, bool):
+                raise ValueError('Plugin processor enabled must be a boolean')
+            agent = await self._get_agent_entity(context, processor_uuid)
+            if agent is None or agent.kind != 'event_processor':
+                raise ValueError('Plugin processor not found')
+            if enabled:
+                try:
+                    descriptor = await self.ap.runner_registry.get(context, agent.component_ref)
+                except RunnerNotFoundError as exc:
+                    raise ValueError('Runner component is unavailable') from exc
+                if 'event' not in descriptor.usages or not descriptor.supported_event_patterns:
+                    raise ValueError('Select an available Runner that declares event usage')
+            seen.add(processor_uuid)
+            normalized.append({'processor_uuid': processor_uuid, 'enabled': enabled})
+        return normalized
+
     async def _prepare_bot_data(self, context: TenantContext, bot_data: dict, *, include_uuid: bool) -> dict:
         """Normalize Bot write payloads to the current event-routing model."""
         update_data = bot_data.copy()
@@ -564,6 +597,10 @@ class BotService:
             update_data['event_bindings'] = await self._normalize_event_bindings(
                 context, update_data.get('event_bindings')
             )
+        if 'plugin_processors' in update_data:
+            update_data['plugin_processors'] = await self._normalize_plugin_processors(
+                context, update_data['plugin_processors']
+            )
         return update_data
 
     async def get_bots(self, context: TenantContext, include_secret: bool = False) -> list[dict]:
@@ -660,6 +697,7 @@ class BotService:
         bot_data['uuid'] = str(uuid.uuid4())
         bot_data['workspace_uuid'] = workspace_uuid
         bot_data.setdefault('event_bindings', [])
+        bot_data.setdefault('plugin_processors', [])
 
         await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_bot.Bot).values(bot_data))
 
@@ -688,7 +726,7 @@ class BotService:
         if getattr(result, 'rowcount', None) == 0:
             raise WorkspaceNotFoundError('Bot not found')
 
-        runtime_fields = {'adapter', 'adapter_config', 'enable', 'event_bindings'}
+        runtime_fields = {'adapter', 'adapter_config', 'enable', 'event_bindings', 'plugin_processors'}
         if not runtime_fields.intersection(update_data):
             runtime_bot = await self.ap.platform_mgr.get_bot_by_uuid(context, bot_uuid)
             if runtime_bot is not None:

src/langbot/pkg/api/mcp/server.py

@@ -109,14 +109,22 @@ class LangBotMCPServer:
             description=(
                 'Create a bot. `bot_data` is a JSON object matching the LangBot '
                 'POST /api/v1/platform/bots body (e.g. name, adapter, config). '
-                'Returns the new bot UUID.'
+                'Use event_bindings for exclusive Agent/Pipeline routes; plugin_processors is an array '
+                'of {processor_uuid, enabled} subscriptions that automatically match Runner-declared events. '
+                'Subscriptions run independently of each other and of event_bindings. Returns the new bot UUID.'
             )
         )
         async def create_bot(bot_data: dict) -> str:
             context = _authorized(Permission.RESOURCE_MANAGE)
             return _dump({'uuid': await ap.bot_service.create_bot(context, bot_data)})
 
-        @mcp.tool(description='Update a bot by UUID. `bot_data` matches the PUT bot body.')
+        @mcp.tool(
+            description=(
+                'Update a bot by UUID. `bot_data` matches the PUT bot body. '
+                'plugin_processors replaces the bot subscriptions with [{processor_uuid, enabled}]. '
+                'Do not put event_processor targets in event_bindings; those are exclusive Agent/Pipeline routes.'
+            )
+        )
         async def update_bot(bot_uuid: str, bot_data: dict) -> str:
             context = _authorized(Permission.RESOURCE_MANAGE)
             await ap.bot_service.update_bot(context, bot_uuid, bot_data)

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

@@ -53,6 +53,7 @@ class Bot(Base):
     adapter_config = sqlalchemy.Column(sqlalchemy.JSON, nullable=False)
     enable = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False, default=False)
     event_bindings = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, server_default='[]')
+    plugin_processors = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, server_default='[]')
     created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
     updated_at = sqlalchemy.Column(
         sqlalchemy.DateTime,

src/langbot/pkg/persistence/alembic/versions/0025_bot_plugin_processors.py

@@ -0,0 +1,79 @@
+"""Separate plugin processor subscriptions from exclusive bot event routes."""
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = '0025_bot_plugin_processors'
+down_revision = '0024_unify_runner_state'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if not inspector.has_table('bots'):
+        return
+    if 'plugin_processors' not in {column['name'] for column in inspector.get_columns('bots')}:
+        op.add_column('bots', sa.Column('plugin_processors', sa.JSON(), nullable=False, server_default='[]'))
+    bots = sa.table(
+        'bots',
+        sa.column('uuid', sa.String()),
+        sa.column('event_bindings', sa.JSON()),
+        sa.column('plugin_processors', sa.JSON()),
+    )
+    for row in bind.execute(sa.select(bots)).mappings():
+        routes = []
+        subscriptions = {item['processor_uuid']: dict(item) for item in (row['plugin_processors'] or [])}
+        for route in row['event_bindings'] or []:
+            if route.get('target_type') != 'event_processor':
+                routes.append(route)
+                continue
+            processor_uuid = route.get('target_uuid')
+            if not processor_uuid:
+                continue
+            if processor_uuid not in subscriptions:
+                subscriptions[processor_uuid] = {'processor_uuid': processor_uuid, 'enabled': False}
+            subscriptions[processor_uuid]['enabled'] |= bool(route.get('enabled', True))
+        bind.execute(
+            bots.update()
+            .where(bots.c.uuid == row['uuid'])
+            .values(
+                event_bindings=routes,
+                plugin_processors=list(subscriptions.values()),
+            )
+        )
+
+
+def downgrade():
+    # A subscription may match several events. Represent it as a wildcard route
+    # when reverting, while retaining the target and enabled state.
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if not inspector.has_table('bots') or 'plugin_processors' not in {
+        column['name'] for column in inspector.get_columns('bots')
+    }:
+        return
+    bots = sa.table(
+        'bots',
+        sa.column('uuid', sa.String()),
+        sa.column('event_bindings', sa.JSON()),
+        sa.column('plugin_processors', sa.JSON()),
+    )
+    for row in bind.execute(sa.select(bots)).mappings():
+        routes = list(row['event_bindings'] or [])
+        for item in row['plugin_processors'] or []:
+            routes.append(
+                {
+                    'id': f'plugin_processor:{item["processor_uuid"]}',
+                    'event_pattern': '*',
+                    'target_type': 'event_processor',
+                    'target_uuid': item['processor_uuid'],
+                    'enabled': item['enabled'],
+                    'filters': [],
+                    'priority': 0,
+                    'order': len(routes),
+                }
+            )
+        bind.execute(bots.update().where(bots.c.uuid == row['uuid']).values(event_bindings=routes))
+    op.drop_column('bots', 'plugin_processors')

src/langbot/pkg/platform/botmgr.py

@@ -280,6 +280,8 @@ class RuntimeBot:
         diagnostic_steps: list[dict[str, typing.Any]] = []
 
         for index, binding in enumerate(bindings):
+            if binding.get('target_type') == 'event_processor':
+                continue
             event_pattern = str(binding.get('event_pattern') or '')
             priority = int(binding.get('priority') or 0)
             order = int(binding.get('order', index))
@@ -841,22 +843,72 @@ class RuntimeBot:
         event.bot_uuid = self.bot_entity.uuid
         await self._record_adapter_event(event, adapter)
 
-        if isinstance(event, platform_events.PlatformSpecificEvent) and event.action == 'interaction.submitted':
-            await self._handle_interaction_submission(event, adapter)
-            return
+        primary = (
+            self._handle_interaction_submission(event, adapter)
+            if isinstance(event, platform_events.PlatformSpecificEvent) and event.action == 'interaction.submitted'
+            else self._dispatch_eba_event_to_processor(event, adapter)
+        )
+        subscriptions = self._get_event_bindings_from_value(getattr(self.bot_entity, 'plugin_processors', []))
+        seen = set()
+        tasks = [primary]
+        for subscription in subscriptions:
+            processor_uuid = subscription.get('processor_uuid')
+            if not processor_uuid or processor_uuid in seen or not subscription.get('enabled', True):
+                continue
+            seen.add(processor_uuid)
+            tasks.append(self._dispatch_plugin_subscription(event, adapter, processor_uuid))
+        # Start all deliveries together. A slow or failed subscriber cannot block
+        # another subscriber or the primary route from receiving the event.
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        for result in results:
+            if isinstance(result, BaseException):
+                await self.logger.error(f'Event delivery failed: {result}')
 
-        # Legacy listeners run inside Pipeline stages. EBA handlers require an
-        # explicitly created and routed plugin processor instance.
-        await self._dispatch_eba_event_to_processor(event, adapter)
+    async def _dispatch_plugin_subscription(self, event, adapter, processor_uuid):
+        event_type = event.type
+        event_binding = {
+            'id': f'plugin_processor:{processor_uuid}',
+            'target_type': 'event_processor',
+            'target_uuid': processor_uuid,
+            'event_pattern': event_type,
+        }
+        try:
+            agent = await self.ap.agent_service.get_agent(self.execution_context, processor_uuid)
+            if not agent or agent.get('kind') != 'event_processor':
+                raise ValueError('Plugin processor not found')
+            descriptor = await self.ap.runner_registry.get(self.execution_context, agent.get('component_ref'))
+            if 'event' not in descriptor.usages:
+                raise ValueError('Runner does not support event processing')
+            # Resolve the installed declaration each time, including after plugin updates.
+            patterns = descriptor.supported_event_patterns
+            if not patterns or not self._agent_supports_event_type(patterns, event_type):
+                return
+            agent = {**agent, 'supported_event_patterns': patterns}
+            return await self._dispatch_eba_event_to_processor(event, adapter, event_binding, agent)
+        except Exception as exc:
+            return await self._record_event_route_trace(
+                event_type=event_type,
+                status='failed',
+                level='error',
+                binding=event_binding,
+                target_type='event_processor',
+                target_uuid=processor_uuid,
+                failure_code='runner_failed',
+                reason=str(exc),
+                text=f'Plugin processor {processor_uuid} failed: {exc}',
+            )
 
     async def _dispatch_eba_event_to_processor(
         self,
         event: platform_events.EBAEvent,
         adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
+        event_binding: dict | None = None,
+        agent: dict | None = None,
     ) -> dict[str, typing.Any]:
         event_type = getattr(event, 'type', None) or event.__class__.__name__
 
-        event_binding = self._resolve_eba_event_binding(event, event_type)
+        if event_binding is None:
+            event_binding = self._resolve_eba_event_binding(event, event_type)
         if event_binding is None:
             return await self._record_event_route_trace(
                 event_type=event_type,
@@ -938,7 +990,8 @@ class RuntimeBot:
             )
 
         target_uuid = event_binding.get('target_uuid')
-        agent = await self.ap.agent_service.get_agent(self.execution_context, target_uuid)
+        if agent is None:
+            agent = await self.ap.agent_service.get_agent(self.execution_context, target_uuid)
         if not agent or agent.get('kind') != target_type:
             return await self._record_event_route_trace(
                 event_type=event_type,