From 812eb09ee42755061d3e47a2231675b419e8be8a Mon Sep 17 00:00:00 2001 From: RockChinQ Date: Mon, 7 Sep 2026 23:51:34 +0800 Subject: [PATCH] fix(pipeline): apply returned plugin event contexts --- docs/event-based-agents/00-overview.md | 2 + .../event-based-agents/09-event-processors.md | 151 ++++++++++++++++++ src/langbot/pkg/pipeline/pipelinemgr.py | 30 ++-- src/langbot/pkg/plugin/connector.py | 4 + tests/unit_tests/pipeline/test_pipelinemgr.py | 50 +++++- tests/unit_tests/pipeline/test_preproc.py | 35 ++++ tests/unit_tests/pipeline/test_wrapper.py | 43 +++++ .../plugin/test_connector_methods.py | 37 +++++ 8 files changed, 329 insertions(+), 23 deletions(-) create mode 100644 docs/event-based-agents/09-event-processors.md diff --git a/docs/event-based-agents/00-overview.md b/docs/event-based-agents/00-overview.md index 1e4708dad..a3bb6f660 100644 --- a/docs/event-based-agents/00-overview.md +++ b/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. 背景与动机 diff --git a/docs/event-based-agents/09-event-processors.md b/docs/event-based-agents/09-event-processors.md new file mode 100644 index 000000000..775614222 --- /dev/null +++ b/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. diff --git a/src/langbot/pkg/pipeline/pipelinemgr.py b/src/langbot/pkg/pipeline/pipelinemgr.py index 34284f142..0f7b4d956 100644 --- a/src/langbot/pkg/pipeline/pipelinemgr.py +++ b/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( - pipeline_entity.extensions_preferences - ) + extensions_prefs = normalize_extension_preferences(pipeline_entity.extensions_preferences) self.enable_all_plugins = extensions_prefs['enable_all_plugins'] is True - self.enable_all_mcp_servers = ( - extensions_prefs['enable_all_mcp_servers'] is True - ) + self.enable_all_mcp_servers = 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( - pipeline_config, runner_id - ) + resolved = RunnerConfigResolver.resolve_runner_config(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 = [ - f'{plugin["author"]}/{plugin["name"]}' - for plugin in extensions_prefs['plugins'] - ] + self.bound_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( - query.pipeline_config - ) + runner_name = RunnerConfigResolver.resolve_runner_id(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( - f'Pipeline stage {stage_name} is not registered; skipping' - ) + self.ap.logger.warning(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))) diff --git a/src/langbot/pkg/plugin/connector.py b/src/langbot/pkg/plugin/connector.py index 3feb37ac6..8c2b499da 100644 --- a/src/langbot/pkg/plugin/connector.py +++ b/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 diff --git a/tests/unit_tests/pipeline/test_pipelinemgr.py b/tests/unit_tests/pipeline/test_pipelinemgr.py index 2265d917b..ceabf845b 100644 --- a/tests/unit_tests/pipeline/test_pipelinemgr.py +++ b/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() - event_ctx.is_prevented_default = Mock(return_value=False) - mock_app.plugin_connector.emit_event = AsyncMock(return_value=event_ctx) + from langbot_plugin.api.entities.context import EventContext + + 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, diff --git a/tests/unit_tests/pipeline/test_preproc.py b/tests/unit_tests/pipeline/test_preproc.py index 60e4c0b8e..defdc10f0 100644 --- a/tests/unit_tests/pipeline/test_preproc.py +++ b/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.""" diff --git a/tests/unit_tests/pipeline/test_wrapper.py b/tests/unit_tests/pipeline/test_wrapper.py index 034c14115..48823e840 100644 --- a/tests/unit_tests/pipeline/test_wrapper.py +++ b/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.""" diff --git a/tests/unit_tests/plugin/test_connector_methods.py b/tests/unit_tests/plugin/test_connector_methods.py index 3c94587e2..7b753f48e 100644 --- a/tests/unit_tests/plugin/test_connector_methods.py +++ b/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()