fix(pipeline): apply returned plugin event contexts

docs/event-based-agents/00-overview.md

@@ -1,5 +1,7 @@
 # Event Based Agents 架构设计总览
 
+> Product revision (2026-09-07): [Event processors and Pipeline plugin compatibility](./09-event-processors.md) defines an explicitly bound EventProcessor alongside Pipeline and Agent. It supersedes the automatic EBA observer product model below; the new component and UI are planned, not yet implemented.
+
 > 当前状态（2026-09-05）：平台事件、Bot `event_bindings`、独立 Agent、Pipeline / Agent 平级路由及 WebUI 已集成到 `dev/4.11.x`。实现入口为 `pkg/platform/botmgr.py::RuntimeBot` 与 `pkg/agent/runner/`。下文“当前架构的局限性”“现有架构”描述改造前背景；EventBus / EventRouter 图表示职责划分，不表示存在同名独立服务。当前实现和验收以 [STATUS.md](../agent-runner-pluginization/STATUS.md) 为准，平台动作使用[授权工具](../agent-runner-pluginization/PLATFORM_ACTION_TOOLS.md)。
 
 ## 1. 背景与动机

docs/event-based-agents/09-event-processors.md

@@ -0,0 +1,151 @@
+# Event processors and Pipeline plugin compatibility
+
+Status: product and implementation design, 2026-09-07. The EventProcessor
+component, routing target, and UI described below are not implemented yet.
+This design supersedes the automatic EBA EventListener observer broadcast in
+the earlier EBA documents. Existing Pipeline plugin behavior remains supported.
+
+## Product boundary
+
+Three processor types appear together in the Processors area:
+
+| Product | Implementation | Event entry | Flow ownership |
+| --- | --- | --- | --- |
+| Pipeline | Existing Pipeline stages and configuration | Received messages | Pipeline stages, including legacy plugin hooks |
+| Agent | A configured plugin AgentRunner | Supported EBA events | The selected runner |
+| Event processor | A configured plugin EventProcessor | Supported EBA events | Plugin Python handlers |
+
+Use **Event processor** as the product label and **EventProcessor** as the SDK
+component name. The localized description should explain that the plugin defines
+the processing logic. It must not suggest an LLM, prompt, or visual workflow is
+required.
+
+An installed component is a reusable implementation. A processor instance is a
+user-created configuration of that component. A Bot event binding selects an
+instance, not an installed plugin package directly.
+
+## Legacy EventListener contract
+
+EventListener remains a Pipeline extension. Existing plugins retain their import
+paths, handler registration syntax, event classes, Query-based APIs, and Pipeline
+plugin selection behavior. No conversion of installed listeners into standalone
+processor instances takes place.
+
+Compatibility must cover execution behavior, not just successful deserialization:
+
+| Hook | Required behavior |
+| --- | --- |
+| PersonMessageReceived / GroupMessageReceived | Read the returned EventContext before later stages; retain message edits and default prevention |
+| PromptPreProcessing | Preserve timing and apply returned default_prompt and prompt |
+| PersonNormalMessageReceived / GroupNormalMessageReceived | Preserve user_message_alter, default prevention, and replacement replies |
+| PersonCommandSent / GroupCommandSent | Preserve command stage timing, default prevention, and replacement replies |
+| NormalMessageResponded | Preserve response-stage timing, default prevention, and replacement message chains, including streaming behavior |
+| All hooks | Preserve plugin ordering, prevent_postorder across installations, bound-plugin filtering, Query identity, and Workspace scope |
+
+RPC responses are new Python objects. Host code must consume returned values
+rather than assume mutations reached the original Query by object identity.
+Host-only references, including the active Query and raw adapter message, must
+remain available for legacy reply APIs without being exposed in serialized
+plugin events.
+
+Preserve source event fields across EBA-to-legacy conversion, including group
+member permissions, bot group permissions, and member titles. Missing platform
+information must be distinguished from fields that were dropped during conversion.
+
+Direct Agent and Event processor execution must not synthesize Pipeline lifecycle
+hooks. Those hooks describe actual Pipeline stages.
+
+## EventProcessor SDK contract
+
+Introduce a distinct component kind instead of changing what an existing
+EventListener manifest means. One package may contain both kinds; only the legacy
+EventListener participates in Pipeline hook dispatch.
+
+Retain the familiar authoring shape:
+
+```python
+# Illustrative API contract; these classes are not available yet.
+class WelcomeProcessor(EventProcessor):
+    async def initialize(self):
+        await super().initialize()
+
+        @self.handler(MemberJoinedEvent)
+        async def on_join(ctx: EventProcessorContext):
+            await ctx.reply(f"Hello, {ctx.event.member.nickname}")
+```
+
+Handlers receive typed EBA events directly. Do not maintain a second, incomplete
+mapping into plugin-only EBA wrapper classes. Preserve complete public event
+fields; compact log previews must not become the execution payload. Include the
+generic platform-specific event contract for adapter-specific events.
+
+The context belongs to one invocation and exposes the event, processor/run
+identifiers, instance configuration, logging, and authorized Host APIs. It has no
+fabricated Pipeline Query. Reuse Host run tracking, deadlines, installation
+authority, platform capabilities, and delivery records where appropriate.
+
+A handler returning normally completes its invocation. There is no implicit LLM
+loop, automatic second processor, or hidden retry of side effects. An exception
+marks the run failed and retains the associated log. New processing handlers do
+not use prevent_default to control another processor; routing has already chosen
+the current processor. The legacy methods keep their existing Pipeline meaning.
+
+## Activation and routing
+
+The activation sequence is explicit:
+
+1. Install a plugin containing an EventProcessor component.
+2. Create an Event processor in the Processors area.
+3. Select its plugin component and enter any component-defined configuration.
+4. Bind a Bot event to that processor instance in the existing event routing UI.
+
+Installation and processor creation alone do not subscribe to Bot events.
+The component declares the event types it handles; Bot bindings select the subset
+of supported events to deliver. One package may supply multiple components, and
+multiple instances may use the same component with independent configuration.
+
+Extend the existing single-target route arbitration with `event_processor`.
+Remove automatic EBA broadcasts to installed EventListeners when this route is
+ready. Keep Pipeline hook dispatch inside the Pipeline path. Existing observer
+plugins must explicitly adopt the new component and be bound by the user; do not
+create subscriptions during migration.
+
+Validate component availability, event compatibility, Workspace ownership, and
+instance identity at creation/update and again at invocation. A disabled or
+unavailable plugin leaves the instance visible with an actionable unavailable
+status. It must not silently fall back to Agent or Pipeline.
+
+## Compact UI
+
+Creation adds a third type next to Agent and Pipeline, followed by a component
+selector and basic instance information. Show configuration fields only when the
+component declares them. If no component is installed, show a relevant plugin
+installation entry point; installing still does not create a binding.
+
+The detail page prioritizes a single run list. Selecting a run shows a chronological
+trace of the incoming event, handler logs, outgoing actions/messages, and outcome.
+Keep payloads and error details collapsed until expanded. Distinguish attempted
+delivery from confirmed delivery and display the actual destination.
+
+Place component identity, availability, bindings, and configuration in a compact
+secondary area. Do not add a prompt editor, model selector, or flow designer.
+The plugin implements the processing flow in code.
+
+## Delivery sequence and acceptance
+
+1. Repair and regression-test Pipeline EventContext handling and legacy payload
+   conversion independently of the new processor feature.
+2. Add the SDK component, context, manifest/scaffolding, and explicit invocation
+   contract; verify registration, event coverage, and process isolation.
+3. Add Host instance management, event routing, execution tracking, and matching
+   HTTP/MCP/skill surfaces. Turn off automatic EBA observer dispatch in this step.
+4. Add creation, binding, availability, logs, and delivery trace UI with i18n.
+5. Exercise a real packaged plugin through installation, explicit instance
+   creation, Bot binding, invocation, logging, and reply delivery.
+
+Acceptance must prove that installation alone invokes no handlers; one matching
+binding invokes exactly the chosen component; instance configuration and run
+history remain separate; all declared EBA events retain their fields; unavailable
+components fail visibly; and legacy plugins keep the documented Pipeline hook
+order and behavior. Unit tests alone do not establish a successful live plugin
+installation or platform delivery.

src/langbot/pkg/pipeline/pipelinemgr.py

@@ -122,20 +122,14 @@ class RuntimePipeline:
         self.placement_generation = self.execution_context.placement_generation
 
         # Extract bound plugins and MCP servers from extensions_preferences
-        extensions_prefs = normalize_extension_preferences(
+        extensions_prefs = normalize_extension_preferences(pipeline_entity.extensions_preferences)
-            pipeline_entity.extensions_preferences
-        )
         self.enable_all_plugins = extensions_prefs['enable_all_plugins'] is True
-        self.enable_all_mcp_servers = (
+        self.enable_all_mcp_servers = extensions_prefs['enable_all_mcp_servers'] is True
-            extensions_prefs['enable_all_mcp_servers'] is True
-        )
         pipeline_config = pipeline_entity.config or {}
         runner_config: dict[str, typing.Any] = {}
         runner_id = RunnerConfigResolver.resolve_runner_id(pipeline_config)
         if runner_id:
-            resolved = RunnerConfigResolver.resolve_runner_config(
+            resolved = RunnerConfigResolver.resolve_runner_config(pipeline_config, runner_id)
-                pipeline_config, runner_id
-            )
             if isinstance(resolved, dict):
                 runner_config = resolved
         self.mcp_resource_attachments = runner_config.get(
@@ -154,10 +148,7 @@ class RuntimePipeline:
             # None indicates to use all available plugins
             self.bound_plugins = None
         else:
-            self.bound_plugins = [
+            self.bound_plugins = [f'{plugin["author"]}/{plugin["name"]}' for plugin in extensions_prefs['plugins']]
-                f'{plugin["author"]}/{plugin["name"]}'
-                for plugin in extensions_prefs['plugins']
-            ]
 
         if self.enable_all_mcp_servers:
             # None indicates to use all available MCP servers
@@ -385,9 +376,7 @@ class RuntimePipeline:
         # Get runner name from pipeline config
         runner_name = None
         if query.pipeline_config:
-            runner_name = RunnerConfigResolver.resolve_runner_id(
+            runner_name = RunnerConfigResolver.resolve_runner_id(query.pipeline_config)
-                query.pipeline_config
-            )
 
         # Record query start and store message_id
         message_id = ''
@@ -442,6 +431,11 @@ class RuntimePipeline:
                 )
                 return
 
+            # The Runtime returns a deserialized event, not the original Query.
+            # Carry plugin message edits into the following Pipeline stages.
+            query.message_chain = event_ctx.event.message_chain
+            query.message_event.message_chain = query.message_chain
+
             self.ap.logger.debug(f'Processing query {query.query_id}')
 
             await self._execute_from_stage(0, query)
@@ -669,9 +663,7 @@ class PipelineManager:
         stage_containers: list[StageInstContainer] = []
         for stage_name in pipeline_entity.stages:
             if stage_name not in self.stage_dict:
-                self.ap.logger.warning(
+                self.ap.logger.warning(f'Pipeline stage {stage_name} is not registered; skipping')
-                    f'Pipeline stage {stage_name} is not registered; skipping'
-                )
                 continue
             stage_containers.append(StageInstContainer(inst_name=stage_name, inst=self.stage_dict[stage_name](self.ap)))
 

src/langbot/pkg/plugin/connector.py

@@ -2114,6 +2114,10 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             event_ctx = context.EventContext.model_validate(result['event_context'])
             emitted_plugins.extend(result.get('emitted_plugins', []))
             response_sources.extend(result.get('response_sources', []))
+            if event_ctx.is_prevented_postorder():
+                break
+        if query is not None:
+            event_ctx.event.query = query
         event_ctx._emitted_plugins = emitted_plugins
         event_ctx._response_sources = response_sources
 

tests/unit_tests/pipeline/test_pipelinemgr.py

@@ -229,6 +229,7 @@ async def test_remove_pipeline(mock_app):
 @pytest.mark.asyncio
 async def test_runtime_pipeline_execute(mock_app, sample_query):
     """Test runtime pipeline execution with real Pydantic models."""
+    sample_query.query_id = 1
     pipelinemgr = get_pipelinemgr_module()
     stage = get_stage_module()
     persistence_pipeline = get_persistence_pipeline_module()
@@ -266,18 +267,59 @@ async def test_runtime_pipeline_execute(mock_app, sample_query):
     )
 
     # Mock plugin connector
-    event_ctx = Mock()
+    from langbot_plugin.api.entities.context import EventContext
-    event_ctx.is_prevented_default = Mock(return_value=False)
+
-    mock_app.plugin_connector.emit_event = AsyncMock(return_value=event_ctx)
+    async def return_event_context(event, bound_plugins):
+        return EventContext.model_validate(EventContext.from_event(event).model_dump())
+
+    mock_app.plugin_connector.emit_event = AsyncMock(side_effect=return_event_context)
 
     # Execute pipeline
     await runtime_pipeline.run(sample_query)
 
     # Verify stage was called
-    mock_stage.process.assert_called_once()
+    assert mock_stage.process.call_count == 1, mock_app.logger.error.call_args_list
     mock_app.query_pool.remove_query.assert_awaited_once_with(sample_query)
 
 
+@pytest.mark.asyncio
+async def test_received_event_edits_reach_pipeline_stages(mock_app, sample_query):
+    """Read edits from the returned RPC context before running message stages."""
+    from langbot_plugin.api.entities.context import EventContext
+    from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain
+
+    sample_query.query_id = 1
+    pipeline_entity = SimpleNamespace(
+        name='Compatibility test',
+        uuid='test-pipeline-uuid',
+        workspace_uuid='test-workspace',
+        config=sample_query.pipeline_config,
+        extensions_preferences={'plugins': []},
+    )
+    runtime_pipeline = get_pipelinemgr_module().RuntimePipeline(
+        mock_app,
+        pipeline_entity,
+        [],
+        _context('test-pipeline-uuid'),
+    )
+    observed = []
+
+    async def plugin_edit(event, bound_plugins):
+        ctx = EventContext.model_validate(EventContext.from_event(event).model_dump())
+        ctx.event.message_chain = MessageChain([Plain(text='edited by plugin')])
+        return ctx
+
+    async def capture_stage(index, query):
+        observed.append((str(query.message_chain), str(query.message_event.message_chain)))
+
+    mock_app.plugin_connector.emit_event = AsyncMock(side_effect=plugin_edit)
+    runtime_pipeline._execute_from_stage = AsyncMock(side_effect=capture_stage)
+
+    await runtime_pipeline.run(sample_query)
+
+    assert observed == [('edited by plugin', 'edited by plugin')], mock_app.logger.error.call_args_list
+
+
 @pytest.mark.asyncio
 async def test_runtime_pipeline_rejects_stale_generation_before_side_effects(
     mock_app,

tests/unit_tests/pipeline/test_preproc.py

@@ -76,6 +76,41 @@ def make_session(
 class TestPreProcessorNormalText:
     """Tests for normal text message preprocessing."""
 
+    @pytest.mark.asyncio
+    async def test_returned_plugin_prompt_edits_are_applied(self):
+        """Prompt hooks retain both edits across the serialized Runtime boundary."""
+        from langbot_plugin.api.entities.context import EventContext
+        from langbot_plugin.api.entities.builtin.provider.message import Message
+
+        app = FakeApp()
+        app.sess_mgr.get_session = AsyncMock(return_value=make_session())
+        conversation = Mock()
+        conversation.prompt = Mock(messages=[])
+        conversation.prompt.copy = Mock(return_value=Mock(messages=[]))
+        conversation.messages = []
+        conversation.uuid = None
+        app.sess_mgr.get_conversation = AsyncMock(return_value=conversation)
+        model = Mock()
+        model.model_entity = Mock(uuid='test-model', abilities=['func_call'])
+        app.model_mgr.get_model_by_uuid = AsyncMock(return_value=model)
+        observed_hooks = []
+
+        async def edit_prompts(event, bound_plugins):
+            observed_hooks.append(event.event_name)
+            ctx = EventContext.model_validate(EventContext.from_event(event).model_dump())
+            ctx.event.default_prompt = [Message(role='system', content='plugin system prompt')]
+            ctx.event.prompt = [Message(role='assistant', content='plugin history')]
+            return ctx
+
+        app.plugin_connector.emit_event = AsyncMock(side_effect=edit_prompts)
+        query = text_query('hello')
+
+        await get_preproc_module().PreProcessor(app).process(query, 'PreProcessor')
+
+        assert observed_hooks == ['PromptPreProcessing']
+        assert query.prompt.messages[0].content == 'plugin system prompt'
+        assert query.messages[0].content == 'plugin history'
+
     @pytest.mark.asyncio
     async def test_normal_text_continues(self):
         """Normal text message should continue pipeline."""

tests/unit_tests/pipeline/test_wrapper.py

@@ -228,6 +228,49 @@ class TestResponseWrapperPlugin:
 class TestResponseWrapperAssistant:
     """Tests for assistant response wrapping."""
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize('streaming', [False, True])
+    @pytest.mark.parametrize('prevent_default', [False, True])
+    async def test_returned_response_context_controls_delivery(self, streaming, prevent_default):
+        """Legacy response hooks can replace or block the actual outgoing chain."""
+        from langbot_plugin.api.entities.context import EventContext
+        from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk
+
+        app = FakeApp()
+        app.sess_mgr.get_session = AsyncMock(return_value=make_session())
+        observed_hooks = []
+
+        async def edit_response(event, bound_plugins):
+            observed_hooks.append(event.event_name)
+            ctx = EventContext.model_validate(EventContext.from_event(event).model_dump())
+            ctx.event.reply_message_chain = platform_message.MessageChain(
+                [
+                    platform_message.Plain(text='plugin reply'),
+                ]
+            )
+            if prevent_default:
+                ctx.prevent_default()
+            return ctx
+
+        app.plugin_connector.emit_event = AsyncMock(side_effect=edit_response)
+        stage = get_wrapper_module().ResponseWrapper(app)
+        query = text_query('hello')
+        query.pipeline_config = make_wrapper_config()
+        message_class = MessageChunk if streaming else Message
+        query.resp_messages = [message_class(role='assistant', content='model reply')]
+        query.resp_message_chain = []
+        await stage.initialize(query.pipeline_config)
+
+        results = [result async for result in stage.process(query, 'ResponseWrapper')]
+
+        assert observed_hooks == ['NormalMessageResponded']
+        if prevent_default:
+            assert results[0].result_type == get_entities_module().ResultType.INTERRUPT
+            assert query.resp_message_chain == []
+        else:
+            assert results[0].result_type == get_entities_module().ResultType.CONTINUE
+            assert [str(chain) for chain in query.resp_message_chain] == ['plugin reply']
+
     @pytest.mark.asyncio
     async def test_assistant_content_response(self):
         """Assistant with content should emit event and wrap."""

tests/unit_tests/plugin/test_connector_methods.py

@@ -239,6 +239,43 @@ class TestListPlugins:
 
 
 class TestPluginDiagnostics:
+    @pytest.mark.asyncio
+    async def test_prevent_postorder_stops_later_installations_and_restores_query(self):
+        from langbot_plugin.api.entities.events import PersonMessageReceived
+
+        connector = create_mock_connector()
+        query = text_query('hello')
+        event = PersonMessageReceived(
+            query=query,
+            launcher_type=query.launcher_type.value,
+            launcher_id=query.launcher_id,
+            sender_id=query.sender_id,
+            message_event=query.message_event,
+            message_chain=query.message_chain,
+        )
+        second_binding = TEST_INSTALLATION_BINDING.model_copy(
+            update={
+                'installation_uuid': '00000000-0000-4000-8000-000000000002',
+            }
+        )
+        connector._operation_bindings = AsyncMock(return_value=[TEST_INSTALLATION_BINDING, second_binding])
+
+        async def stop_following_plugins(event_context, include_plugins=None):
+            event_context['is_prevent_postorder'] = True
+            return {'event_context': event_context, 'emitted_plugins': ['first']}
+
+        runtime_handler = configure_handler(connector, Mock())
+        runtime_handler.emit_event = AsyncMock(side_effect=stop_following_plugins)
+
+        returned = await connector.emit_event(event)
+
+        runtime_handler.emit_event.assert_awaited_once()
+        assert returned.is_prevented_postorder()
+        assert not returned.is_prevented_default()
+        assert returned.event.query is query
+        assert 'query' not in returned.event.model_dump()
+        assert returned._emitted_plugins == ['first']
+
     @pytest.mark.asyncio
     async def test_emit_event_preserves_response_sources(self):
         connector = create_mock_connector()