diff --git a/docs/event-based-agents/04-event-routing.md b/docs/event-based-agents/04-event-routing.md index d45a9199d..b4670b70f 100644 --- a/docs/event-based-agents/04-event-routing.md +++ b/docs/event-based-agents/04-event-routing.md @@ -1,5 +1,7 @@ # 事件路由与编排 +> Implementation update (2026-09-08): the EventListener observer-broadcast proposal below is superseded by [Event processors](09-event-processors.md). Legacy EventListener hooks run only inside Pipeline. New EBA handlers use explicitly created and bound EventProcessor instances, a third peer processor type alongside Agent and Pipeline. + > 状态：当前实施模型（2026-07-12）。本文以 Pipeline / Agent 平级并存为准，不再保留早期 `pipeline / agent / webhook / plugin` 四种 Handler 草案。 ## 1. 路由边界 diff --git a/docs/event-based-agents/05-plugin-sdk.md b/docs/event-based-agents/05-plugin-sdk.md index eabed1055..c4201e901 100644 --- a/docs/event-based-agents/05-plugin-sdk.md +++ b/docs/event-based-agents/05-plugin-sdk.md @@ -1,5 +1,7 @@ # 插件 SDK 改造 +> Implementation update (2026-09-08): the EventListener observer-broadcast proposal below is superseded by [Event processors](09-event-processors.md). Legacy EventListener hooks run only inside Pipeline. New EBA handlers use explicitly created and bound EventProcessor instances, a third peer processor type alongside Agent and Pipeline. + ## 1. 概述 插件 SDK 需要配合 EBA 架构进行以下改造： diff --git a/docs/event-based-agents/06-migration-plan.md b/docs/event-based-agents/06-migration-plan.md index 9d7262b65..332921119 100644 --- a/docs/event-based-agents/06-migration-plan.md +++ b/docs/event-based-agents/06-migration-plan.md @@ -1,5 +1,7 @@ # EBA 分阶段实施计划 +> Implementation update (2026-09-08): the EventListener observer-broadcast proposal below is superseded by [Event processors](09-event-processors.md). Legacy EventListener hooks run only inside Pipeline. New EBA handlers use explicitly created and bound EventProcessor instances, a third peer processor type alongside Agent and Pipeline. + > 更新：2026-09-05。P0–P4 的主要实现已落入 `dev/4.11.x`，P5 仍需按当前版本验收；下文工作项用于维护实现边界，不表示全部待开发。文件名沿用早期设计，但这里的“迁移”仅指代码架构逐步接入 EBA，不代表 LangBot 3.x 数据库或配置升级。当前提交、定向测试及发布缺口见 [STATUS.md](../agent-runner-pluginization/STATUS.md)。 ## 1. 发布边界 diff --git a/docs/event-based-agents/07-agent-orchestration.md b/docs/event-based-agents/07-agent-orchestration.md index 4f49c13b2..a16f4f8e1 100644 --- a/docs/event-based-agents/07-agent-orchestration.md +++ b/docs/event-based-agents/07-agent-orchestration.md @@ -1,5 +1,7 @@ # Agent 与 Pipeline 统一编排（产品最终形态） +> Implementation update (2026-09-08): the EventListener observer-broadcast proposal below is superseded by [Event processors](09-event-processors.md). Legacy EventListener hooks run only inside Pipeline. New EBA handlers use explicitly created and bound EventProcessor instances, a third peer processor type alongside Agent and Pipeline. + > **状态**：历史方向稿（2026-06-12）；2026-09-05 标记归档用途。本文的示意 schema、5.0 发布火车、SDK 0.5.0aX 配套与多租户“预留”描述不再作为实施合同。当前 4.11 产品形态见 [08-agent-page-and-event-orchestration.md](./08-agent-page-and-event-orchestration.md)，协议见 [PROTOCOL_V1.md](../agent-runner-pluginization/PROTOCOL_V1.md)，已完成与剩余事项见 [STATUS.md](../agent-runner-pluginization/STATUS.md)。保留正文仅用于解释早期设计取舍。 > > 本文档修订 [00-overview.md](./00-overview.md) §3.4 与 [04-event-routing.md](./04-event-routing.md) 中"四种 Handler"的编排模型：**所有编排目标统一进入处理器选择与事件绑定界面，但独立 Agent 与现有 Pipeline 保持不同类型**。事件路由的匹配机制、数据迁移策略、WebUI 交互骨架等内容仍以 04 为准，仅 handler 分类法被本文档取代。 diff --git a/docs/event-based-agents/08-agent-page-and-event-orchestration.md b/docs/event-based-agents/08-agent-page-and-event-orchestration.md index e531cc400..9fb53a55a 100644 --- a/docs/event-based-agents/08-agent-page-and-event-orchestration.md +++ b/docs/event-based-agents/08-agent-page-and-event-orchestration.md @@ -1,5 +1,7 @@ # 处理器页面与事件编排产品设计 +> Implementation update (2026-09-08): the EventListener observer-broadcast proposal below is superseded by [Event processors](09-event-processors.md). Legacy EventListener hooks run only inside Pipeline. New EBA handlers use explicitly created and bound EventProcessor instances, a third peer processor type alongside Agent and Pipeline. + > 状态：当前实现说明（2026-09-05），对应 `dev/4.11.x`。P0–P3 已集成；发布验收见 [STATUS.md](../agent-runner-pluginization/STATUS.md)。 > > 本文档修订 [07-agent-orchestration.md](./07-agent-orchestration.md) 中“Agent 替代 Pipeline”的表述。当前产品形态保留两种长期并存的同级处理器：**Agent** 与 **Pipeline**。处理器页面只是共享入口，不改变二者各自的持久化模型和执行语义。 diff --git a/docs/event-based-agents/09-event-processors.md b/docs/event-based-agents/09-event-processors.md index 775614222..5c4155f98 100644 --- a/docs/event-based-agents/09-event-processors.md +++ b/docs/event-based-agents/09-event-processors.md @@ -1,7 +1,9 @@ # 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. +Status: implemented in the 4.11 development branches of LangBot and the Plugin SDK, +2026-09-08. The Host uv configuration pins the matching SDK commit. Deploy both +revisions together; older SDK releases do not contain this component. Switch the +development source pin to a published SDK release before a stable PyPI release. This design supersedes the automatic EBA EventListener observer broadcast in the earlier EBA documents. Existing Pipeline plugin behavior remains supported. @@ -64,7 +66,8 @@ EventListener participates in Pipeline hook dispatch. Retain the familiar authoring shape: ```python -# Illustrative API contract; these classes are not available yet. +from langbot_plugin.api.definition.components.event_processor import EventProcessor, EventProcessorContext +from langbot_plugin.api.entities.builtin.platform.events import MemberJoinedEvent class WelcomeProcessor(EventProcessor): async def initialize(self): await super().initialize() @@ -105,8 +108,7 @@ 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 +There is no automatic EBA broadcast to installed EventListeners. 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. @@ -149,3 +151,66 @@ history remain separate; all declared EBA events retain their fields; unavailabl 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. + + +## Implemented transport and APIs + +`lbp comp EventProcessor` scaffolds a component in `components/event_processor`. +Its manifest uses `kind: EventProcessor` and `spec.events`, for example +`[group.member_joined]`. `spec.config` defines instance parameters. A component +that calls `ctx.reply()` declares `spec.permissions.tools: [detail, call]`. + +References use `event_processor:author/plugin/component`, separate from +`plugin:author/plugin/runner`. Both kinds share the existing run transport, +installation authorization, deadlines and run ledger. The trusted Host selects +the component kind; the worker invokes only that exact kind and name. +There is no model invocation in the EventProcessor base class. + +`EventProcessorContext` provides `event`, `run_id`, `config`, `api`, `log()` and +`reply()`. `api` is the existing run-scoped Host proxy. Use `ctx.config` for instance +parameters; plugin installation configuration remains separate. Handlers may +register the `EBAEvent` base class as a catch-all. An exact typed handler takes +precedence over that fallback. Multiple handlers for the same type run in their +registration order, within one invocation. + +HTTP instance management uses `/api/v1/agents` with `kind: event_processor`. +Metadata at `/api/v1/agents/_/metadata` lists installed `event_processors`. +Creation accepts `component_ref` and `parameters`; the Host derives the supported +event patterns from the component. Bot bindings use `target_type: event_processor` +and the created instance UUID as `target_id`. + +- `GET /api/v1/agents/{id}/runs?before_id=...` lists this instance's runs. +- `GET /api/v1/agents/{id}/runs/{run_id}/events?after_sequence=...` pages its logs + and action results. A run from another instance or Workspace is rejected. +- The corresponding MCP tools are `get_processor_metadata`, `list_processor_runs` + and `get_processor_run_events`, alongside processor CRUD. +- `/api/v1/agents/{id}/debug` accepts a full typed EBA event in `data`. Platform + actions use Mock; other authorized tools retain their configured behavior. + +The detail page polls run updates, keeps payload details collapsed and separates +logs from platform delivery. Completed handlers produce no synthetic reply text. + + +## Verification (2026-09-08) + +- SDK API, scaffolding and Plugin Runtime suites: 684 passed. +- Host runner, service, controller, MCP and adapter regression suites: 971 passed. +- Pipeline and registry regression suites: 243 passed, one environment-dependent skip. +- Frontend unit suite: 74 passed; TypeScript and changed-file lint checks passed. +- Real packaged-plugin tests cover installation, component-kind separation, + invocation, mock platform delivery, instance isolation and persisted logs. +- Authenticated Edge testing created an instance and a loopback OneBot bot, saved + a member-join binding, injected one native notice and received exactly one + `send_group_msg` response. The detail page displayed the completed run, + localized action name, destination and returned message ID. No external IM + account or live model was involved. +- Browser regression covers expanding long payloads, reaching the last log and + pagination without duplication. Eight unrelated existing browser failures were + reproduced against the pre-change commit; the full suite is not green. +- Repository-wide lint also retains the pre-existing duplicate `send_image_msg` + in the WeCom customer-service library and existing formatting failures outside + this change. The i18n check has the same pre-existing diagnostics as its baseline. + +Native Agent interaction-resumption is not enabled for EventProcessor bindings; +its input contract is the platform EBA event collection, not a synthetic Agent +continuation. Plugins should handle platform events through their typed handlers. diff --git a/pyproject.toml b/pyproject.toml index 93f7d817b..09bd38e7f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -69,7 +69,7 @@ dependencies = [ "langchain-text-splitters>=1.1.2", "chromadb>=1.0.0,<2.0.0", "qdrant-client (>=1.15.1,<2.0.0)", - "langbot-plugin==0.5.3", + "langbot-plugin==0.5.5", "asyncpg>=0.30.0", "line-bot-sdk>=3.19.0", "matrix-nio>=0.25.2", @@ -229,3 +229,7 @@ skip-magic-trailing-comma = false # Like Black, automatically detect the appropriate line ending. line-ending = "auto" + +[tool.uv.sources] +# Development contract: update to the matching SDK release before publishing. +langbot-plugin = { git = "https://github.com/langbot-app/langbot-plugin-sdk", rev = "23011398160cedcd4ef090054692bc2a4b34ee9f" } diff --git a/skills/skills/langbot-mcp-ops/SKILL.md b/skills/skills/langbot-mcp-ops/SKILL.md index a7e360412..22cef1800 100644 --- a/skills/skills/langbot-mcp-ops/SKILL.md +++ b/skills/skills/langbot-mcp-ops/SKILL.md @@ -65,7 +65,9 @@ The tools wrap the LangBot service layer. Current tools (v1): | `get_system_info` | Version, edition, instance id | | `list_bots` / `get_bot` / `create_bot` / `update_bot` / `delete_bot` | Manage messaging-platform bots (secrets redacted on read) | | `list_bot_event_route_statuses` | Inspect bot event-route runtime status | -| `list_processors` / `get_processor` / `create_processor` / `update_processor` / `delete_processor` | Manage the peer Agent and Pipeline processor types | +| `list_processors` / `get_processor` / `create_processor` / `update_processor` / `delete_processor` | Manage the peer Agent, Pipeline and Event processor types | +| `get_processor_metadata` | Discover installed EventProcessor components, schemas and supported event patterns. | +| `list_processor_runs` / `get_processor_run_events` | Read one Event processor instance run history and logs; paginate with `before_id` / `after_sequence`. | | `debug_agent` | Execute a synthetic Agent event (`processor_uuid`, `payload`); requires `runtime.operate`. Returns final text and up to 1000 execution events (thinking, text, tool arguments/results). Platform tools use Mock; other configured tools execute normally. Optional `payload.mock`: `errors`/`results` keyed by platform tool name, `unsupported_apis` lists unavailable platform APIs. | | `list_pipelines` / `get_pipeline` / `create_pipeline` / `update_pipeline` / `delete_pipeline` | Manage pipelines | | `list_llm_models` / `get_llm_model` / `list_embedding_models` / `list_model_providers` | Inspect models & providers | @@ -111,3 +113,12 @@ already have a default pipeline. - A `403` means the key is valid but lacks the permission required by the tool. - The global key is plaintext in config.yaml — only enable it on trusted/internal deployments and serve over HTTPS. + +## Event processors + +Install the plugin, discover its component with `get_processor_metadata`, then +create a processor with `kind: "event_processor"`, `component_ref` and optional +`parameters`. Bind bot events to this instance with `target_type: "event_processor"` +and `target_id` equal to its UUID. Installation alone never activates a handler. +`debug_agent` accepts the complete typed EBA event in `payload.data` for this kind. +Legacy EventListener plugins remain in the Pipeline lifecycle. diff --git a/src/langbot/pkg/agent/runner/descriptor.py b/src/langbot/pkg/agent/runner/descriptor.py index 2397f169e..69c27c838 100644 --- a/src/langbot/pkg/agent/runner/descriptor.py +++ b/src/langbot/pkg/agent/runner/descriptor.py @@ -1,4 +1,5 @@ """Agent runner descriptor.""" + from __future__ import annotations import typing @@ -44,19 +45,18 @@ class AgentRunnerDescriptor(pydantic.BaseModel): config_schema: list[dict[str, typing.Any]] = pydantic.Field(default_factory=list) """Configuration schema using DynamicForm format""" - capabilities: AgentRunnerCapabilities = pydantic.Field( - default_factory=AgentRunnerCapabilities - ) + capabilities: AgentRunnerCapabilities = pydantic.Field(default_factory=AgentRunnerCapabilities) """Runner capabilities: streaming, tool_calling, knowledge_retrieval, etc.""" - permissions: AgentRunnerPermissions = pydantic.Field( - default_factory=AgentRunnerPermissions - ) + permissions: AgentRunnerPermissions = pydantic.Field(default_factory=AgentRunnerPermissions) """Requested LangBot resource permissions.""" raw_manifest: dict[str, typing.Any] = pydantic.Field(default_factory=dict) """Original manifest for reference""" + component_kind: typing.Literal['AgentRunner', 'EventProcessor'] = 'AgentRunner' + supported_event_patterns: list[str] = pydantic.Field(default_factory=lambda: ['*']) + model_config = pydantic.ConfigDict( extra='allow', ) diff --git a/src/langbot/pkg/agent/runner/host_models.py b/src/langbot/pkg/agent/runner/host_models.py index 71d2e9259..d3a1b0fac 100644 --- a/src/langbot/pkg/agent/runner/host_models.py +++ b/src/langbot/pkg/agent/runner/host_models.py @@ -160,7 +160,7 @@ class AgentConfig(pydantic.BaseModel): agent_id: str | None = None """Host-side Agent/config identifier.""" - processor_type: typing.Literal['agent', 'pipeline'] = 'agent' + processor_type: typing.Literal['agent', 'pipeline', 'event_processor'] = 'agent' """Product processor kind represented by this runtime config.""" processor_id: str | None = None @@ -222,7 +222,7 @@ class AgentBinding(pydantic.BaseModel): agent_id: str | None = None """Host-side Agent/config identifier for this binding.""" - processor_type: typing.Literal['agent', 'pipeline'] = 'agent' + processor_type: typing.Literal['agent', 'pipeline', 'event_processor'] = 'agent' """Product processor kind selected for this binding.""" processor_id: str | None = None diff --git a/src/langbot/pkg/agent/runner/id.py b/src/langbot/pkg/agent/runner/id.py index e01099041..736ff87c4 100644 --- a/src/langbot/pkg/agent/runner/id.py +++ b/src/langbot/pkg/agent/runner/id.py @@ -1,4 +1,5 @@ """Agent runner ID parsing and formatting.""" + from __future__ import annotations import dataclasses @@ -7,6 +8,7 @@ import dataclasses @dataclasses.dataclass(frozen=True) class RunnerIdParts: """Parsed runner ID components.""" + source: str # 'plugin' (future: 'builtin') plugin_author: str plugin_name: str @@ -29,31 +31,27 @@ def parse_runner_id(runner_id: str) -> RunnerIdParts: Raises: ValueError: If runner_id format is invalid """ - if runner_id.startswith('plugin:'): - parts = runner_id[7:].split('/') + if runner_id.startswith(('plugin:', 'event_processor:')): + source, value = runner_id.split(':', 1) + parts = value.split('/') if len(parts) != 3: raise ValueError( - f'Invalid plugin runner ID format: {runner_id}. ' - f'Expected: plugin:author/plugin_name/runner_name' + f'Invalid plugin runner ID format: {runner_id}. Expected: plugin:author/plugin_name/runner_name' ) plugin_author, plugin_name, runner_name = parts if not plugin_author or not plugin_name or not runner_name: raise ValueError( - f'Invalid plugin runner ID: {runner_id}. ' - f'author, plugin_name, and runner_name must be non-empty' + f'Invalid plugin runner ID: {runner_id}. author, plugin_name, and runner_name must be non-empty' ) return RunnerIdParts( - source='plugin', + source=source, plugin_author=plugin_author, plugin_name=plugin_name, runner_name=runner_name, ) else: # Only plugin runner IDs are valid at the protocol boundary. - raise ValueError( - f'Invalid runner ID format: {runner_id}. ' - f'Expected: plugin:author/plugin_name/runner_name' - ) + raise ValueError(f'Invalid runner ID format: {runner_id}. Expected: plugin:author/plugin_name/runner_name') def format_runner_id( @@ -73,8 +71,8 @@ def format_runner_id( Returns: Runner ID string """ - if source == 'plugin': - return f'plugin:{plugin_author}/{plugin_name}/{runner_name}' + if source in {'plugin', 'event_processor'}: + return f'{source}:{plugin_author}/{plugin_name}/{runner_name}' else: raise ValueError(f'Invalid runner source: {source}') @@ -88,4 +86,4 @@ def is_plugin_runner_id(runner_id: str) -> bool: Returns: True if runner ID starts with 'plugin:' """ - return runner_id.startswith('plugin:') + return runner_id.startswith(('plugin:', 'event_processor:')) diff --git a/src/langbot/pkg/agent/runner/invoker.py b/src/langbot/pkg/agent/runner/invoker.py index 89285bfb9..b65fbd594 100644 --- a/src/langbot/pkg/agent/runner/invoker.py +++ b/src/langbot/pkg/agent/runner/invoker.py @@ -41,6 +41,17 @@ class AgentRunnerInvoker: ) try: + if descriptor.component_kind == 'EventProcessor': + context = { + **context, + 'runtime': { + **context['runtime'], + 'metadata': { + **context['runtime'].get('metadata', {}), + 'component_kind': 'EventProcessor', + }, + }, + } gen = self.ap.plugin_connector.run_agent( plugin_author=descriptor.plugin_author, plugin_name=descriptor.plugin_name, diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index 50ba360e3..a4da83fe6 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -102,6 +102,10 @@ class AgentRunOrchestrator: bound_plugins, ) + expected_kind = 'EventProcessor' if binding.processor_type == 'event_processor' else 'AgentRunner' + if descriptor.component_kind != expected_kind: + raise ValueError('Processor kind does not match the selected plugin component') + if execution_query is None: execution_query = build_execution_query(event, []) # Synthetic events must expose the same trusted scope as pipeline queries. diff --git a/src/langbot/pkg/agent/runner/registry.py b/src/langbot/pkg/agent/runner/registry.py index 1a98187d3..4eb93a699 100644 --- a/src/langbot/pkg/agent/runner/registry.py +++ b/src/langbot/pkg/agent/runner/registry.py @@ -110,19 +110,19 @@ class AgentRunnerRegistry: manifest = runner_data.get('manifest', {}) runner_id = format_runner_id( - source='plugin', + source='event_processor' if manifest.get('component_kind') == 'EventProcessor' else 'plugin', plugin_author=plugin_author, plugin_name=plugin_name, runner_name=runner_name, ) typed_manifest = AgentRunnerManifest.model_validate(manifest) - config_schema = [ - item.model_dump(mode='json') for item in typed_manifest.config_schema - ] + config_schema = [item.model_dump(mode='json') for item in typed_manifest.config_schema] return AgentRunnerDescriptor( id=runner_id, + component_kind=typed_manifest.component_kind, + supported_event_patterns=typed_manifest.supported_event_patterns, source='plugin', label=typed_manifest.label, description=typed_manifest.description, @@ -152,6 +152,7 @@ class AgentRunnerRegistry: context: TenantContext, bound_plugins: list[str] | None = None, use_cache: bool = True, + component_kind: str = 'AgentRunner', ) -> list[AgentRunnerDescriptor]: """List available runners. @@ -169,7 +170,11 @@ class AgentRunnerRegistry: # Filter from cache. Do not treat an empty cache as final because the # plugin runtime may still be launching installed plugins when the # first metadata request arrives. - return self._filter_runners_by_bound_plugins(cached, bound_plugins) + return [ + r + for r in self._filter_runners_by_bound_plugins(cached, bound_plugins) + if r.component_kind == component_kind + ] # Discover fresh (always full list) runners = await self._discover_runners() @@ -179,7 +184,11 @@ class AgentRunnerRegistry: self._cache[cache_key] = runners # Filter locally - return self._filter_runners_by_bound_plugins(runners, bound_plugins) + return [ + r + for r in self._filter_runners_by_bound_plugins(runners, bound_plugins) + if r.component_kind == component_kind + ] def _filter_runners_by_bound_plugins( self, @@ -233,7 +242,8 @@ class AgentRunnerRegistry: except ValueError as e: raise RunnerNotFoundError(runner_id) from e - runners = await self.list_runners(context, bound_plugins=None) + component_kind = 'EventProcessor' if runner_id.startswith('event_processor:') else 'AgentRunner' + runners = await self.list_runners(context, bound_plugins=None, component_kind=component_kind) descriptor = next((item for item in runners if item.id == runner_id), None) if descriptor is None: # The runtime launches installed plugins asynchronously, so an @@ -242,6 +252,7 @@ class AgentRunnerRegistry: context, bound_plugins=None, use_cache=False, + component_kind=component_kind, ) descriptor = next((item for item in runners if item.id == runner_id), None) if descriptor is None: diff --git a/src/langbot/pkg/agent/runner/result_normalizer.py b/src/langbot/pkg/agent/runner/result_normalizer.py index d06cd1034..b80c213cd 100644 --- a/src/langbot/pkg/agent/runner/result_normalizer.py +++ b/src/langbot/pkg/agent/runner/result_normalizer.py @@ -10,6 +10,7 @@ from langbot_plugin.api.entities.builtin.agent_runner.result import ( MessageCompletedPayload, MessageDeltaPayload, RunCompletedPayload, + ProcessorLogPayload, RunFailedPayload, StateUpdatedPayload, ToolCallCompletedPayload, @@ -34,6 +35,7 @@ STRICT_RESULT_PAYLOADS: dict[str, type[pydantic.BaseModel]] = { 'action.requested': ActionRequestedPayload, 'run.completed': RunCompletedPayload, 'run.failed': RunFailedPayload, + 'processor.log': ProcessorLogPayload, } @@ -114,6 +116,9 @@ class AgentResultNormalizer: if not self.validate_payload(result_type, data, descriptor): return None + if result_type == 'processor.log': + return None + if result_type == 'message.delta': return self._normalize_message_delta(data, descriptor) diff --git a/src/langbot/pkg/agent/runner/run_journal.py b/src/langbot/pkg/agent/runner/run_journal.py index 2c478c84e..d4a42ea6b 100644 --- a/src/langbot/pkg/agent/runner/run_journal.py +++ b/src/langbot/pkg/agent/runner/run_journal.py @@ -93,6 +93,13 @@ class AgentRunJournal: metadata={ 'event_type': event.event_type, 'source': event.source, + 'processor_id': binding.processor_id, + 'processor_type': binding.processor_type, + **( + {'input_event': event.data, 'delivery': event.delivery.model_dump(mode='json')} + if binding.processor_type == 'event_processor' + else {} + ), }, ) diff --git a/src/langbot/pkg/agent/runner/run_ledger_store.py b/src/langbot/pkg/agent/runner/run_ledger_store.py index c5e53c9f7..41f835a01 100644 --- a/src/langbot/pkg/agent/runner/run_ledger_store.py +++ b/src/langbot/pkg/agent/runner/run_ledger_store.py @@ -188,8 +188,8 @@ class RunLedgerStore: query = query.where(AgentRun.conversation_id == conversation_id) query = self._apply_scope_filters(query, bot_id, workspace_id, thread_id, strict_thread) - query = query.order_by(AgentRun.priority.desc(), AgentRun.id.asc()).limit(1).with_for_update( - skip_locked=True + query = ( + query.order_by(AgentRun.priority.desc(), AgentRun.id.asc()).limit(1).with_for_update(skip_locked=True) ) result = await session.execute(query) run = result.scalars().first() @@ -571,8 +571,7 @@ class RunLedgerStore: # Filter by labels runtimes = [ - rt for rt in all_runtimes - if all(rt.get('labels', {}).get(k) == v for k, v in labels.items()) + rt for rt in all_runtimes if all(rt.get('labels', {}).get(k) == v for k, v in labels.items()) ] total_count = len(runtimes) @@ -636,6 +635,7 @@ class RunLedgerStore: thread_id: str | None = None, strict_thread: bool = False, runner_id: str | None = None, + binding_id: str | None = None, ) -> tuple[list[dict[str, typing.Any]], int | None, bool, int]: """Page runs by scope. @@ -652,6 +652,8 @@ class RunLedgerStore: count_query = count_query.where(AgentRun.status.in_(statuses)) if runner_id is not None: count_query = count_query.where(AgentRun.runner_id == runner_id) + if binding_id is not None: + count_query = count_query.where(AgentRun.binding_id == binding_id) count_query = self._apply_scope_filters(count_query, bot_id, workspace_id, thread_id, strict_thread) count_result = await session.execute(count_query) total_count = count_result.scalar() or 0 @@ -664,6 +666,8 @@ class RunLedgerStore: query = query.where(AgentRun.status.in_(statuses)) if runner_id is not None: query = query.where(AgentRun.runner_id == runner_id) + if binding_id is not None: + query = query.where(AgentRun.binding_id == binding_id) if before_id is not None: query = query.where(AgentRun.id < before_id) query = self._apply_scope_filters(query, bot_id, workspace_id, thread_id, strict_thread) @@ -848,10 +852,7 @@ class RunLedgerStore: # Count by status status_query = ( - sqlalchemy.select( - AgentRun.status, - func.count(AgentRun.id).label('count') - ) + sqlalchemy.select(AgentRun.status, func.count(AgentRun.id).label('count')) .where(*base_filter) .group_by(AgentRun.status) ) @@ -873,18 +874,15 @@ class RunLedgerStore: avg_queue_wait_seconds = None # Fetch completed runs with timing data - timing_query = ( - sqlalchemy.select( - AgentRun.started_at, - AgentRun.finished_at, - AgentRun.created_at, - ) - .where( - AgentRun.status == 'completed', - AgentRun.started_at.is_not(None), - AgentRun.finished_at.is_not(None), - *base_filter - ) + timing_query = sqlalchemy.select( + AgentRun.started_at, + AgentRun.finished_at, + AgentRun.created_at, + ).where( + AgentRun.status == 'completed', + AgentRun.started_at.is_not(None), + AgentRun.finished_at.is_not(None), + *base_filter, ) timing_result = await session.execute(timing_query) timing_rows = timing_result.all() @@ -899,16 +897,10 @@ class RunLedgerStore: avg_duration_seconds = round(sum(durations) / len(durations), 2) # Queue wait time - compute in Python - queue_query = ( - sqlalchemy.select( - AgentRun.created_at, - AgentRun.started_at, - ) - .where( - AgentRun.started_at.is_not(None), - *base_filter - ) - ) + queue_query = sqlalchemy.select( + AgentRun.created_at, + AgentRun.started_at, + ).where(AgentRun.started_at.is_not(None), *base_filter) queue_result = await session.execute(queue_query) queue_rows = queue_result.all() @@ -957,12 +949,8 @@ class RunLedgerStore: async with self._session_factory() as session: # Count by status - status_query = ( - sqlalchemy.select( - AgentRuntime.status, - func.count(AgentRuntime.id).label('count') - ) - .group_by(AgentRuntime.status) + status_query = sqlalchemy.select(AgentRuntime.status, func.count(AgentRuntime.id).label('count')).group_by( + AgentRuntime.status ) status_result = await session.execute(status_query) status_counts = {row.status: row.count for row in status_result} @@ -975,9 +963,8 @@ class RunLedgerStore: avg_heartbeat_age = None max_heartbeat_age = None - heartbeat_query = ( - sqlalchemy.select(AgentRuntime.last_heartbeat_at) - .where(AgentRuntime.last_heartbeat_at.is_not(None)) + heartbeat_query = sqlalchemy.select(AgentRuntime.last_heartbeat_at).where( + AgentRuntime.last_heartbeat_at.is_not(None) ) heartbeat_result = await session.execute(heartbeat_query) heartbeat_rows = heartbeat_result.all() @@ -995,16 +982,12 @@ class RunLedgerStore: avg_heartbeat_age = round(sum(ages) / len(ages), 2) max_heartbeat_age = round(max(ages), 2) - active_runs_query = ( - sqlalchemy.select(func.count(AgentRun.id)) - .where(AgentRun.status.in_(['running', 'claimed'])) + active_runs_query = sqlalchemy.select(func.count(AgentRun.id)).where( + AgentRun.status.in_(['running', 'claimed']) ) active_runs_result = await session.execute(active_runs_query) active_runs = active_runs_result.scalar() or 0 - claimed_runs_query = ( - sqlalchemy.select(func.count(AgentRun.id)) - .where(AgentRun.status == 'claimed') - ) + claimed_runs_query = sqlalchemy.select(func.count(AgentRun.id)).where(AgentRun.status == 'claimed') claimed_runs_result = await session.execute(claimed_runs_query) claimed_runs = claimed_runs_result.scalar() or 0 @@ -1048,23 +1031,10 @@ class RunLedgerStore: AgentRun.runner_id, func.count(AgentRun.id).label('total'), func.sum( - sqlalchemy.case( - (AgentRun.status.in_(['queued', 'claimed', 'running']), 1), - else_=0 - ) + sqlalchemy.case((AgentRun.status.in_(['queued', 'claimed', 'running']), 1), else_=0) ).label('active'), - func.sum( - sqlalchemy.case( - (AgentRun.status == 'completed', 1), - else_=0 - ) - ).label('completed'), - func.sum( - sqlalchemy.case( - (AgentRun.status.in_(['failed', 'timeout']), 1), - else_=0 - ) - ).label('failed'), + func.sum(sqlalchemy.case((AgentRun.status == 'completed', 1), else_=0)).label('completed'), + func.sum(sqlalchemy.case((AgentRun.status.in_(['failed', 'timeout']), 1), else_=0)).label('failed'), ) .where( AgentRun.created_at >= start_dt, @@ -1087,16 +1057,18 @@ class RunLedgerStore: failed = row.failed or 0 success_rate = completed / total if total > 0 else None - stats.append({ - 'runner_id': runner_id, - 'runner_label': None, # Would need to join with runner descriptors - 'plugin_identity': None, - 'total_runs': total, - 'active_runs': row.active or 0, - 'completed_runs': completed, - 'failed_runs': failed, - 'success_rate': round(success_rate, 4) if success_rate is not None else None, - 'avg_duration_seconds': None, # Would need more complex query - }) + stats.append( + { + 'runner_id': runner_id, + 'runner_label': None, # Would need to join with runner descriptors + 'plugin_identity': None, + 'total_runs': total, + 'active_runs': row.active or 0, + 'completed_runs': completed, + 'failed_runs': failed, + 'success_rate': round(success_rate, 4) if success_rate is not None else None, + 'avg_duration_seconds': None, # Would need more complex query + } + ) return stats diff --git a/src/langbot/pkg/api/http/controller/groups/agents.py b/src/langbot/pkg/api/http/controller/groups/agents.py index 696aee6e2..4b0edcf36 100644 --- a/src/langbot/pkg/api/http/controller/groups/agents.py +++ b/src/langbot/pkg/api/http/controller/groups/agents.py @@ -18,6 +18,43 @@ from .agent_debug_stream import debug_stream_response @group.group_class('agents', '/api/v1/agents') class AgentsRouterGroup(group.RouterGroup): async def initialize(self) -> None: + @self.route( + '//runs', + methods=['GET'], + auth_type=group.AuthType.USER_TOKEN_OR_API_KEY, + permission=Permission.RESOURCE_VIEW, + ) + async def processor_runs(agent_uuid: str, request_context: RequestContext): + try: + cursor = quart.request.args.get('before_id') + result = await self.ap.agent_service.get_processor_runs( + request_context, + agent_uuid, + before_id=int(cursor) if cursor else None, + ) + return self.success(data=result) + except ValueError as exc: + return self.http_status(400, -1, str(exc)) + + @self.route( + '//runs//events', + methods=['GET'], + auth_type=group.AuthType.USER_TOKEN_OR_API_KEY, + permission=Permission.RESOURCE_VIEW, + ) + async def processor_run_events(agent_uuid: str, run_id: str, request_context: RequestContext): + try: + cursor = quart.request.args.get('after_sequence') + result = await self.ap.agent_service.get_processor_run_events( + request_context, + agent_uuid, + run_id, + after_sequence=int(cursor) if cursor else None, + ) + return self.success(data=result) + except ValueError as exc: + return self.http_status(400, -1, str(exc)) + @self.route( '//debug/stream', methods=['POST'], diff --git a/src/langbot/pkg/api/http/service/agent.py b/src/langbot/pkg/api/http/service/agent.py index 1f52cf5a1..c2a0b3172 100644 --- a/src/langbot/pkg/api/http/service/agent.py +++ b/src/langbot/pkg/api/http/service/agent.py @@ -40,6 +40,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.*'] AGENT_DEFAULT_EVENT_PATTERNS = ['*'] @@ -67,7 +68,19 @@ class AgentService: ) except Exception as exc: self.ap.logger.warning(f'Failed to load Agent Host tool catalog: {exc}') + event_processors = [] + registry = getattr(self.ap, 'agent_runner_registry', None) + if registry is not None: + event_processors = [ + item.model_dump(mode='json') + for item in await registry.list_runners( + context, + component_kind='EventProcessor', + use_cache=False, + ) + ] return { + 'event_processors': event_processors, 'runner_config': ai_metadata, 'platform_tools': platform_tool_catalog(), 'host_tools': host_tools, @@ -82,6 +95,7 @@ class AgentService: 'supported_event_patterns': PIPELINE_EVENT_PATTERNS, 'message_only': True, }, + {'name': AGENT_KIND_EVENT_PROCESSOR, 'supported_event_patterns': ['*'], 'message_only': False}, ], } @@ -131,7 +145,7 @@ class AgentService: non-message event envelopes. """ agent = await self.get_agent(context, agent_uuid) - if agent is None or agent.get('kind') != AGENT_KIND_AGENT: + if agent is None or agent.get('kind') not in {AGENT_KIND_AGENT, AGENT_KIND_EVENT_PROCESSOR}: raise ValueError('Agent not found') event_type = str(payload.get('event_type', 'message.received')).strip() @@ -242,8 +256,28 @@ class AgentService: raw_ref=RawEventRef(ref_id=event_id, storage_key=None), data=event_data, ) + if agent.get('kind') == AGENT_KIND_EVENT_PROCESSOR: + from langbot_plugin.api.entities.builtin.platform.events import parse_eba_event + + typed_event = parse_eba_event({**event_data, 'type': event_type}) + from ....platform.botmgr import RuntimeBot + + event.actor = RuntimeBot._infer_actor_context(typed_event) + event.subject = RuntimeBot._infer_subject_context(typed_event) + target_type, target_id, target_metadata = RuntimeBot._infer_reply_target(typed_event) + if target_id is not None: + event.delivery.reply_target = { + 'target_type': target_type, + 'target_id': str(target_id), + **target_metadata, + } + event.data = typed_event.model_dump(mode='json', exclude={'source_platform_object', 'legacy_event'}) binding = AgentBinding( - binding_id=f'debug:{agent_uuid}:{runner_id}', + binding_id=( + f'event_processor:{agent_uuid}' + if agent.get('kind') == AGENT_KIND_EVENT_PROCESSOR + else f'debug:{agent_uuid}:{runner_id}' + ), scope=BindingScope(scope_type='agent', scope_id=agent_uuid), event_types=[event_type], runner_id=runner_id, @@ -263,7 +297,7 @@ class AgentService: enable_interactions=False, ), agent_id=agent_uuid, - processor_type='agent', + processor_type=agent.get('kind', 'agent'), processor_id=agent_uuid, ) execution_context = ExecutionContext.from_request( @@ -282,6 +316,7 @@ class AgentService: 'tool.call.completed', 'run.completed', 'run.failed', + 'processor.log', }: return visible_result = copy.deepcopy( @@ -363,11 +398,17 @@ class AgentService: ) return {'uuid': pipeline_uuid, 'kind': AGENT_KIND_PIPELINE} - if kind != AGENT_KIND_AGENT: + if kind not in {AGENT_KIND_AGENT, AGENT_KIND_EVENT_PROCESSOR}: raise ValueError(f'Unsupported agent kind: {kind}') - config = agent_data['config'] if 'config' in agent_data else await self._get_default_agent_config(context) - config, runner_id, _ = RunnerConfigResolver.resolve_agent_runner_config(config) + if kind == AGENT_KIND_EVENT_PROCESSOR: + config, runner_id, patterns = await self._prepare_event_processor(context, agent_data) + else: + config = agent_data['config'] if 'config' in agent_data else await self._get_default_agent_config(context) + config, runner_id, _ = RunnerConfigResolver.resolve_agent_runner_config(config) + if (runner_id or '').startswith('event_processor:'): + raise ValueError('EventProcessor components require an event processor instance') + patterns = agent_data.get('supported_event_patterns', AGENT_DEFAULT_EVENT_PATTERNS) new_uuid = str(uuid.uuid4()) values = { 'workspace_uuid': workspace_uuid, @@ -375,17 +416,13 @@ class AgentService: 'name': agent_data.get('name') or 'New Agent', 'description': agent_data.get('description') or '', 'emoji': agent_data.get('emoji') or '🤖', - 'kind': AGENT_KIND_AGENT, + 'kind': kind, 'component_ref': runner_id, 'config': config, - 'supported_event_patterns': ( - agent_data['supported_event_patterns'] - if 'supported_event_patterns' in agent_data - else AGENT_DEFAULT_EVENT_PATTERNS - ), + 'supported_event_patterns': patterns, } await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_agent.Agent).values(**values)) - return {'uuid': new_uuid, 'kind': AGENT_KIND_AGENT} + return {'uuid': new_uuid, 'kind': kind} async def update_agent(self, context: TenantContext, agent_uuid: str, agent_data: dict) -> None: existing_agent = await self._get_agent_row(context, agent_uuid) @@ -401,12 +438,19 @@ class AgentService: for field in ('name', 'description', 'emoji', 'config', 'supported_event_patterns') if field in agent_data } + if existing_agent.kind == AGENT_KIND_EVENT_PROCESSOR and any( + field in agent_data for field in ('config', 'component_ref', 'parameters', 'supported_event_patterns') + ): + config, runner_id, patterns = await self._prepare_event_processor(context, agent_data, existing_agent) + update_data.update(config=config, component_ref=runner_id, supported_event_patterns=patterns) if 'config' in update_data: config, runner_id, _ = RunnerConfigResolver.resolve_agent_runner_config(update_data['config']) update_data['config'] = config else: _, runner_id, _ = RunnerConfigResolver.resolve_agent_runner_config(existing_agent.config) update_data['component_ref'] = runner_id + if existing_agent.kind == AGENT_KIND_AGENT and (runner_id or '').startswith('event_processor:'): + raise ValueError('EventProcessor components require an event processor instance') result = await self.ap.persistence_mgr.execute_async( scope_statement( sqlalchemy.update(persistence_agent.Agent) @@ -438,6 +482,78 @@ class AgentService: raise ValueError(f'Agent {agent_uuid} not found') await self.ap.pipeline_service.delete_pipeline(context, agent_uuid) + async def _prepare_event_processor(self, context, data, existing=None): + """Resolve an installed component and keep its capability declaration authoritative.""" + config = copy.deepcopy(data.get('config', existing.config if existing is not None else {})) + if not isinstance(config, dict): + raise ValueError('Processor configuration must be an object') + component_ref = data.get('component_ref') or (existing.component_ref if existing is not None else None) + if not isinstance(component_ref, str) or not component_ref.startswith('event_processor:'): + raise ValueError('Select an installed EventProcessor component') + try: + descriptor = await self.ap.agent_runner_registry.get(context, component_ref) + except Exception as exc: + from ....agent.runner.errors import RunnerNotFoundError + + if isinstance(exc, RunnerNotFoundError): + raise ValueError('EventProcessor component is unavailable') from exc + raise + if descriptor.component_kind != 'EventProcessor' or not descriptor.supported_event_patterns: + raise ValueError('The component does not declare supported EBA events') + config['runner'] = {'id': component_ref} + parameters = data.get('parameters') + if parameters is None: + runner_config = config.get('runner_config', {}) + if not isinstance(runner_config, dict): + raise ValueError('Runner configuration must be an object') + parameters = runner_config.get(component_ref) + if parameters is None: + parameters = self.ap.pipeline_service._get_default_values_from_schema(descriptor.config_schema) + if not isinstance(parameters, dict): + raise ValueError('Processor parameters must be an object') + for field in descriptor.config_schema: + if field.get('required') and parameters.get(field['name']) in (None, ''): + raise ValueError(f'Required processor parameter: {field["name"]}') + config['runner_config'] = {component_ref: parameters} + return config, component_ref, descriptor.supported_event_patterns + + async def get_processor_runs(self, context, processor_id, *, before_id=None): + """Read only this Workspace's explicitly created processor instance.""" + from ....agent.runner.run_ledger_store import RunLedgerStore + + processor = await self.get_agent(context, processor_id) + if processor is None or processor.get('kind') != AGENT_KIND_EVENT_PROCESSOR: + raise ValueError('Event processor not found') + store = RunLedgerStore(self.ap.persistence_mgr.get_db_engine()) + items, cursor, has_more, total = await store.list_runs( + workspace_id=require_workspace_uuid(context), + binding_id=f'event_processor:{processor_id}', + before_id=before_id, + ) + return {'items': items, 'next_cursor': cursor, 'has_more': has_more, 'total': total} + + async def get_processor_run_events(self, context, processor_id, run_id, *, after_sequence=None): + """Authorize the parent run before exposing any trace events.""" + from ....agent.runner.run_ledger_store import RunLedgerStore + + processor = await self.get_agent(context, processor_id) + if processor is None or processor.get('kind') != AGENT_KIND_EVENT_PROCESSOR: + raise ValueError('Event processor not found') + store = RunLedgerStore(self.ap.persistence_mgr.get_db_engine()) + run = await store.get_run(run_id) + if ( + run is None + or run.get('workspace_id') != require_workspace_uuid(context) + or run.get('binding_id') != f'event_processor:{processor_id}' + ): + raise ValueError('Processor run not found') + items, next_cursor, _, has_more = await store.page_run_events( + run_id=run_id, + after_sequence=after_sequence, + limit=100, + ) + return {'run': run, 'items': items, 'next_cursor': next_cursor, 'has_more': has_more} + async def _get_agent_rows(self, context: TenantContext) -> list[persistence_agent.Agent]: result = await self.ap.persistence_mgr.execute_async( scope_statement( @@ -490,7 +606,7 @@ class AgentService: include_config: bool = False, ) -> dict[str, typing.Any]: item = self.ap.persistence_mgr.serialize_model(persistence_agent.Agent, agent) - item['kind'] = AGENT_KIND_AGENT + item['kind'] = item.get('kind') or AGENT_KIND_AGENT supported_event_patterns = item.get('supported_event_patterns') item['capability'] = { 'supported_event_patterns': ( diff --git a/src/langbot/pkg/api/http/service/bot.py b/src/langbot/pkg/api/http/service/bot.py index 768aa648f..82199904d 100644 --- a/src/langbot/pkg/api/http/service/bot.py +++ b/src/langbot/pkg/api/http/service/bot.py @@ -149,7 +149,7 @@ class BotService: return target_kind if target_type == 'discard': return 'discard' - if target_type in {'agent', 'pipeline'}: + if target_type in {'agent', 'pipeline', 'event_processor'}: return str(target_type) return None @@ -416,9 +416,9 @@ class BotService: diagnostic_steps=diagnostic_steps, ) - if target_type == 'agent': + if target_type in {'agent', 'event_processor'}: agent = await self._get_agent_entity(tenant_context, target_uuid) - if agent is None or getattr(agent, 'kind', 'agent') != 'agent': + if agent is None or getattr(agent, 'kind', 'agent') != target_type: return self._diagnostic_result( matched=False, binding=selected_binding, @@ -514,7 +514,7 @@ class BotService: ) if result.first() is None: raise ValueError('Pipeline not found') - elif target_type == 'agent': + elif target_type in {'agent', 'event_processor'}: result = await self.ap.persistence_mgr.execute_async( scope_statement( sqlalchemy.select(persistence_agent.Agent).where(persistence_agent.Agent.uuid == target_uuid), @@ -523,8 +523,8 @@ class BotService: ) ) agent = result.first() - if agent is None: - raise ValueError('Agent not found') + if agent is None or agent.kind != target_type: + raise ValueError('Processor not found') if not self._agent_supports_event_pattern(agent.supported_event_patterns, event_pattern): raise ValueError('Agent does not support this event pattern') elif target_type == 'discard': diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py index b118078c8..819e9ac06 100644 --- a/src/langbot/pkg/api/mcp/server.py +++ b/src/langbot/pkg/api/mcp/server.py @@ -175,44 +175,72 @@ class LangBotMCPServer: return _dump({'ok': True}) # ----- Processors ---------------------------------------------- # - @mcp.tool(description='List product-level processors, including Agents and Pipelines.') + @mcp.tool(description='List product-level processors, including Agents, Pipelines and Event processors.') async def list_processors() -> str: context = _authorized(Permission.RESOURCE_VIEW) return _dump(await ap.agent_service.get_agents(context)) - @mcp.tool(description='Get an Agent or Pipeline processor by UUID.') + @mcp.tool(description='Get an Agent, Pipeline or Event processor by UUID.') async def get_processor(processor_uuid: str) -> str: context = _authorized(Permission.RESOURCE_VIEW) return _dump(await ap.agent_service.get_agent(context, processor_uuid)) @mcp.tool( description=( - 'Create an Agent or Pipeline processor. Set `processor_data.kind` to ' - '`agent` or `pipeline`. Returns the new UUID and kind.' + 'Create an Agent, Pipeline or Event processor. Set `processor_data.kind` to ' + '`agent`, `pipeline` or `event_processor`. Event processors require an installed component_ref ' + 'from get_processor_metadata; optional parameters configure the instance. Returns UUID and kind.' ) ) async def create_processor(processor_data: dict) -> str: context = _authorized(Permission.RESOURCE_MANAGE) return _dump(await ap.agent_service.create_agent(context, processor_data)) - @mcp.tool(description='Update an Agent or Pipeline processor by UUID.') + @mcp.tool(description='Update an Agent, Pipeline or Event processor by UUID.') async def update_processor(processor_uuid: str, processor_data: dict) -> str: context = _authorized(Permission.RESOURCE_MANAGE) await ap.agent_service.update_agent(context, processor_uuid, processor_data) return _dump({'ok': True}) - @mcp.tool(description='Delete an Agent or Pipeline processor by UUID.') + @mcp.tool(description='Delete an Agent, Pipeline or Event processor by UUID.') async def delete_processor(processor_uuid: str) -> str: context = _authorized(Permission.RESOURCE_MANAGE) await ap.agent_service.delete_agent(context, processor_uuid) return _dump({'ok': True}) + @mcp.tool(description='Get processor kinds and installed EventProcessor components with configuration schemas.') + async def get_processor_metadata() -> str: + context = _authorized(Permission.RESOURCE_VIEW) + return _dump(await ap.agent_service.get_agent_metadata(context)) + + @mcp.tool(description='List one Event processor instance run history; use before_id to page older runs.') + async def list_processor_runs(processor_uuid: str, before_id: int | None = None) -> str: + context = _authorized(Permission.RESOURCE_VIEW) + return _dump(await ap.agent_service.get_processor_runs(context, processor_uuid, before_id=before_id)) + + @mcp.tool(description='Read logs and action results for an Event processor run; page using after_sequence.') + async def get_processor_run_events( + processor_uuid: str, + run_id: str, + after_sequence: int | None = None, + ) -> str: + context = _authorized(Permission.RESOURCE_VIEW) + return _dump( + await ap.agent_service.get_processor_run_events( + context, + processor_uuid, + run_id, + after_sequence=after_sequence, + ) + ) + # ----- Models -------------------------------------------------- # @mcp.tool( description=( - 'Run a synthetic event against an Agent processor without platform delivery. ' + 'Run a synthetic event against an Agent or Event processor without platform delivery. ' 'Returns final text and execution_events containing reported messages/thinking and tool calls. ' 'Platform tools use mock adapters; other tools execute normally. ' + 'For Event processors, data contains the complete typed EBA event fields. ' 'Requires runtime.operate; payload accepts event_type, text, data, conversation_id, actor, subject and ' 'mock (errors/results keyed by platform tool name; unsupported_apis lists unavailable platform APIs).' ) diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/event_converter.py b/src/langbot/pkg/platform/adapters/aiocqhttp/event_converter.py index 507be3d4a..752dd04dc 100644 --- a/src/langbot/pkg/platform/adapters/aiocqhttp/event_converter.py +++ b/src/langbot/pkg/platform/adapters/aiocqhttp/event_converter.py @@ -62,12 +62,25 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter): chat_id = getattr(event, 'group_id', '') group = AiocqhttpEventConverter.group_from_event(event) + sender = AiocqhttpEventConverter.user_from_sender(event) + sender_data = getattr(event, 'sender', {}) or {} + role = sender_data.get('role', 'member') + membership = None + if group is not None: + membership = platform_entities.UserGroupMember( + user=sender, + group_id=group.id, + role=role if role in {'owner', 'admin', 'member'} else 'member', + display_name=sender_data.get('card') or sender.nickname, + title=sender_data.get('title'), + ) return platform_events.MessageReceivedEvent( type='message.received', adapter_name='aiocqhttp', message_id=getattr(event, 'message_id', ''), message_chain=message_chain, - sender=AiocqhttpEventConverter.user_from_sender(event), + sender=sender, + sender_member=membership, chat_type=chat_type, chat_id=chat_id, group=group, diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index 3cdbb1ef9..4fe623d6f 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -630,6 +630,7 @@ class RuntimeBot: name=event.group.name, ) return platform_events.MessageReceivedEvent( + legacy_event=event, message_id=self._extract_message_id(event.message_chain), message_chain=event.message_chain, sender=platform_entities.User( @@ -646,6 +647,7 @@ class RuntimeBot: ) return platform_events.MessageReceivedEvent( + legacy_event=event, message_id=self._extract_message_id(event.message_chain), message_chain=event.message_chain, sender=platform_entities.User( @@ -805,7 +807,11 @@ class RuntimeBot: return None return AgentBinding( - binding_id=f'bot:{bot_uuid}:{event_binding.get("id") or uuid.uuid4()}', + binding_id=( + f'event_processor:{agent["uuid"]}' + if agent.get('kind') == 'event_processor' + else f'bot:{bot_uuid}:{event_binding.get("id") or uuid.uuid4()}' + ), scope=BindingScope(scope_type='bot', scope_id=bot_uuid), event_types=[event_type], runner_id=runner_id, @@ -820,10 +826,10 @@ class RuntimeBot: delivery_policy=DeliveryPolicy( enable_streaming=False, enable_reply=True, - enable_interactions=True, + enable_interactions=agent.get('kind') != 'event_processor', ), agent_id=agent.get('uuid'), - processor_type='agent', + processor_type=agent.get('kind', 'agent'), processor_id=agent.get('uuid'), ) @@ -898,14 +904,8 @@ class RuntimeBot: await self._handle_interaction_submission(event, adapter) return - plugin_event = self._eba_event_to_plugin_event(event) - - if plugin_event is not None: - try: - await self.ap.plugin_connector.emit_event(plugin_event) - except Exception: - await self.logger.error(f'Failed to dispatch platform event to plugins: {traceback.format_exc()}') - + # Legacy listeners run inside Pipeline stages. EBA handlers require an + # explicitly created and routed EventProcessor instance. await self._dispatch_eba_event_to_processor(event, adapter) async def _dispatch_eba_event_to_processor( @@ -983,7 +983,7 @@ class RuntimeBot: target_uuid=event_binding.get('target_uuid'), text=f'EBA event {event_type} delivered to Pipeline {event_binding.get("target_uuid") or ""}'.strip(), ) - if target_type != 'agent': + if target_type not in {'agent', 'event_processor'}: return await self._record_event_route_trace( event_type=event_type, status='failed', @@ -998,7 +998,7 @@ class RuntimeBot: target_uuid = event_binding.get('target_uuid') agent = await self.ap.agent_service.get_agent(self.execution_context, target_uuid) - if not agent or agent.get('kind') != 'agent': + if not agent or agent.get('kind') != target_type: return await self._record_event_route_trace( event_type=event_type, status='failed', @@ -1050,6 +1050,8 @@ class RuntimeBot: ) envelope = self._eba_event_to_agent_envelope(event, adapter) + if target_type == 'event_processor': + envelope.data = event.model_dump(mode='json', exclude={'source_platform_object', 'legacy_event'}) outputs: list[provider_message.Message | provider_message.MessageChunk] = [] try: async for output in self.ap.agent_run_orchestrator.run( diff --git a/tests/e2e/test_agent_runner_plugin_runtime.py b/tests/e2e/test_agent_runner_plugin_runtime.py index 29c61f4d9..d2b4f3184 100644 --- a/tests/e2e/test_agent_runner_plugin_runtime.py +++ b/tests/e2e/test_agent_runner_plugin_runtime.py @@ -74,6 +74,9 @@ def _write_qa_agent_runner_plugin(plugin_root: Path) -> None: AgentRunner: fromDirs: - path: components/agent_runner/ + EventProcessor: + fromDirs: + - path: components/event_processor/ pages: [] execution: python: @@ -160,6 +163,45 @@ def _write_qa_agent_runner_plugin(plugin_root: Path) -> None: encoding='utf-8', ) + processor_dir = plugin_root / 'components' / 'event_processor' + processor_dir.mkdir(parents=True) + (processor_dir / 'default.yaml').write_text( + textwrap.dedent(""" + apiVersion: langbot/v1 + kind: EventProcessor + metadata: + name: default + label: {en_US: Welcome processor, zh_Hans: Welcome processor} + spec: + events: [group.member_joined] + config: + - name: greeting + type: string + required: true + label: {en_US: Greeting, zh_Hans: Greeting} + default: Hello + capabilities: {tool_calling: true} + permissions: + tools: [detail, call] + execution: + python: {path: default.py, attr: WelcomeProcessor} + """) + ) + (processor_dir / 'default.py').write_text( + textwrap.dedent(""" + from langbot_plugin.api.definition.components.event_processor import EventProcessor, EventProcessorContext + from langbot_plugin.api.entities.builtin.platform.events import MemberJoinedEvent + + class WelcomeProcessor(EventProcessor): + async def initialize(self): + @self.handler(MemberJoinedEvent) + async def handle(ctx: EventProcessorContext): + await ctx.log('Handling ' + str(ctx.event.member.id)) + result = await ctx.reply(ctx.config['greeting'] + ', ' + (ctx.event.member.nickname or str(ctx.event.member.id))) + await ctx.log('Reply simulated: ' + str(result.get('mock'))) + """) + ) + def _free_port() -> int: """Reserve a currently-free localhost TCP port for this E2E process.""" @@ -384,6 +426,7 @@ def test_plugin_runtime_discovers_agent_runner( f'Runtime stderr (tail):\n{runtime_stderr[-20_000:]}' ) + def test_host_orchestrator_runs_agent_runner_and_records_ledger( agent_runner_client, agent_runner_langbot_process, @@ -455,8 +498,7 @@ def test_host_orchestrator_runs_agent_runner_and_records_ledger( event_types = { row[0] for row in conn.execute( - 'SELECT type FROM agent_run_event WHERE run_id = ' - '(SELECT run_id FROM agent_run WHERE event_id = ?)', + 'SELECT type FROM agent_run_event WHERE run_id = (SELECT run_id FROM agent_run WHERE event_id = ?)', (result['event_id'],), ).fetchall() } @@ -469,3 +511,63 @@ def test_host_orchestrator_runs_agent_runner_and_records_ledger( assert '"count": 1' in state_row[0] finally: conn.close() + + +def test_event_processor_real_runtime_logs_actions_and_instance_isolation( + agent_runner_client, + agent_runner_e2e_tmpdir, +): + client = agent_runner_client + token = _init_and_auth(client) + _ensure_qa_plugin(client, token, agent_runner_e2e_tmpdir / 'agent-runner-qa.zip') + headers = {'Authorization': f'Bearer {token}'} + metadata_response = client.get('/api/v1/agents/_/metadata', headers=headers).json() + assert metadata_response['code'] == 0, metadata_response + metadata = metadata_response['data'] + ref = 'event_processor:e2e/agent-runner-qa/default' + assert any(item['id'] == ref for item in metadata['event_processors']), metadata + assert ref not in _wait_for_qa_runner(client, token) + created = [] + for name in ('First', 'Second'): + response = client.post( + '/api/v1/agents', + headers=headers, + json={ + 'kind': 'event_processor', + 'name': name, + 'component_ref': ref, + 'parameters': {'greeting': name}, + }, + ).json() + assert response['code'] == 0, response + created.append(response['data']['uuid']) + for processor_id in created: + page = client.get(f'/api/v1/agents/{processor_id}/runs', headers=headers).json() + assert page['data']['items'] == [], page + result = client.post( + f'/api/v1/agents/{created[0]}/debug', + headers=headers, + json={ + 'event_type': 'group.member_joined', + 'data': {'member': {'id': 'member-1', 'nickname': 'Tester'}, 'group': {'id': 'group-1'}}, + }, + ).json() + assert result['code'] == 0, result + logs = [event['data']['text'] for event in result['data']['execution_events'] if event['type'] == 'processor.log'] + assert logs == ['Handling member-1', 'Reply simulated: True'], result + actions = [event for event in result['data']['execution_events'] if event['type'] == 'tool.call.completed'] + assert len(actions) == 1, result + assert actions[0]['data']['tool_name'] == 'event_reply' + assert actions[0]['data']['result']['mock'] is True + assert result['data']['final_text'] == '', result + page = client.get(f'/api/v1/agents/{created[0]}/runs', headers=headers).json()['data'] + assert len(page['items']) == 1, page + run = page['items'][0] + assert run['status'] == 'completed', run + assert run['metadata']['input_event']['member']['id'] == 'member-1' + trace = client.get(f'/api/v1/agents/{created[0]}/runs/{run["run_id"]}/events', headers=headers).json() + assert trace['code'] == 0, trace + assert any(item['type'] == 'processor.log' for item in trace['data']['items']), trace + foreign = client.get(f'/api/v1/agents/{created[1]}/runs/{run["run_id"]}/events', headers=headers) + assert foreign.status_code == 400 + assert client.get(f'/api/v1/agents/{created[1]}/runs', headers=headers).json()['data']['items'] == [] diff --git a/tests/unit_tests/agent/test_registry.py b/tests/unit_tests/agent/test_registry.py index 9e1de5276..b0c3af3d7 100644 --- a/tests/unit_tests/agent/test_registry.py +++ b/tests/unit_tests/agent/test_registry.py @@ -335,3 +335,31 @@ class TestDescriptorValidation: assert descriptor.supports_streaming() is True assert descriptor.supports_tool_calling() is False assert descriptor.supports_knowledge_retrieval() is False + + +@pytest.mark.asyncio +async def test_registry_separates_processor_kinds_with_same_plugin_component_name(): + ap = FakeApplication() + entries = [] + for kind, prefix in [('AgentRunner', 'plugin'), ('EventProcessor', 'event_processor')]: + entries.append( + { + 'plugin_author': 'test', + 'plugin_name': 'both', + 'runner_name': 'default', + 'manifest': { + 'id': f'{prefix}:test/both/default', + 'name': 'default', + 'component_kind': kind, + 'label': {'en_US': kind}, + 'supported_event_patterns': ['group.member_joined'], + }, + } + ) + ap.plugin_connector.list_agent_runners = AsyncMock(return_value=entries) + registry = AgentRunnerRegistry(ap) + agents = await registry.list_runners(TEST_CONTEXT) + processors = await registry.list_runners(TEST_CONTEXT, component_kind='EventProcessor') + assert [item.id for item in agents] == ['plugin:test/both/default'] + assert [item.id for item in processors] == ['event_processor:test/both/default'] + assert (await registry.get(TEST_CONTEXT, processors[0].id)).component_kind == 'EventProcessor' diff --git a/tests/unit_tests/agent/test_run_ledger_store.py b/tests/unit_tests/agent/test_run_ledger_store.py index c3667fcde..0a747f7e0 100644 --- a/tests/unit_tests/agent/test_run_ledger_store.py +++ b/tests/unit_tests/agent/test_run_ledger_store.py @@ -428,3 +428,30 @@ async def test_runner_stats_reports_zero_success_rate_for_failed_only_runner(sto assert stats[0]['runner_id'] == 'runner-a' assert stats[0]['failed_runs'] == 1 assert stats[0]['success_rate'] == 0.0 + + +@pytest.mark.asyncio +async def test_processor_instance_history_filters_count_and_pages(store): + for index, (workspace, binding) in enumerate( + [ + ('one', 'processor-a'), + ('one', 'processor-b'), + ('two', 'processor-a'), + ('one', 'processor-a'), + ] + ): + await store.create_run( + run_id=f'run-{index}', + event_id=f'event-{index}', + binding_id=binding, + runner_id='same-component', + workspace_id=workspace, + ) + first, cursor, more, total = await store.list_runs(workspace_id='one', binding_id='processor-a', limit=1) + assert (total, more) == (2, True) + assert first[0]['run_id'] == 'run-3' + second, _, more, total = await store.list_runs( + workspace_id='one', binding_id='processor-a', limit=1, before_id=cursor + ) + assert (total, more) == (2, False) + assert second[0]['run_id'] == 'run-0' diff --git a/tests/unit_tests/api/service/test_agent_service.py b/tests/unit_tests/api/service/test_agent_service.py index 8554b034a..e234ab414 100644 --- a/tests/unit_tests/api/service/test_agent_service.py +++ b/tests/unit_tests/api/service/test_agent_service.py @@ -145,6 +145,7 @@ class TestAgentServiceMetadata: 'supported_event_patterns': PIPELINE_EVENT_PATTERNS, 'message_only': True, }, + {'name': 'event_processor', 'supported_event_patterns': ['*'], 'message_only': False}, ] @@ -767,3 +768,77 @@ class TestAgentServiceCreateUpdateDelete: WORKSPACE_UUID, 'pipeline-1', ) + + +async def test_event_processor_creation_uses_installed_component_scope(): + app = _make_app() + ref = 'event_processor:test/welcome/default' + descriptor = SimpleNamespace( + component_kind='EventProcessor', + supported_event_patterns=['group.member_joined'], + config_schema=[{'name': 'greeting', 'required': True}], + ) + app.agent_runner_registry = SimpleNamespace(get=AsyncMock(return_value=descriptor)) + service = AgentService(app) + result = await service.create_agent( + WORKSPACE_UUID, + { + 'kind': 'event_processor', + 'name': 'Welcome', + 'component_ref': ref, + 'parameters': {'greeting': 'Hi'}, + 'supported_event_patterns': ['*'], + }, + ) + values = _compiled_params(app.persistence_mgr.execute_async.call_args.args[0]) + assert result['kind'] == 'event_processor' + assert values['component_ref'] == ref + assert values['supported_event_patterns'] == ['group.member_joined'] + assert values['config']['runner_config'][ref] == {'greeting': 'Hi'} + + +async def test_event_processor_rejects_invalid_component_and_missing_parameters(): + app = _make_app() + service = AgentService(app) + with pytest.raises(ValueError, match='Select an installed'): + await service.create_agent(WORKSPACE_UUID, {'kind': 'event_processor', 'component_ref': 'plugin:a/b/c'}) + app.agent_runner_registry = SimpleNamespace( + get=AsyncMock( + return_value=SimpleNamespace( + component_kind='EventProcessor', + supported_event_patterns=['*'], + config_schema=[{'name': 'greeting', 'required': True}], + ) + ) + ) + with pytest.raises(ValueError, match='Required processor parameter'): + await service.create_agent( + WORKSPACE_UUID, {'kind': 'event_processor', 'component_ref': 'event_processor:a/b/c'} + ) + + +async def test_unavailable_event_processor_can_still_be_renamed(): + app = _make_app() + row = _agent_row(config={'runner': {'id': 'event_processor:a/b/c'}, 'runner_config': {'event_processor:a/b/c': {}}}) + row.kind = 'event_processor' + row.component_ref = 'event_processor:a/b/c' + service = AgentService(app) + service._get_agent_row = AsyncMock(return_value=row) + await service.update_agent(WORKSPACE_UUID, row.uuid, {'name': 'Renamed'}) + assert _compiled_update_values(app.persistence_mgr.execute_async.call_args.args[0])['name'] == 'Renamed' + + +@pytest.mark.parametrize( + 'workspace,binding', [('other-workspace', 'event_processor:one'), (WORKSPACE_UUID, 'event_processor:two')] +) +async def test_processor_trace_rejects_other_workspace_or_instance(monkeypatch, workspace, binding): + service = AgentService(_make_app()) + service.get_agent = AsyncMock(return_value={'kind': 'event_processor'}) + service.ap.persistence_mgr.get_db_engine = Mock() + store = SimpleNamespace( + get_run=AsyncMock(return_value={'workspace_id': workspace, 'binding_id': binding}), page_run_events=AsyncMock() + ) + monkeypatch.setattr('langbot.pkg.agent.runner.run_ledger_store.RunLedgerStore', Mock(return_value=store)) + with pytest.raises(ValueError, match='Processor run not found'): + await service.get_processor_run_events(WORKSPACE_UUID, 'one', 'run') + store.page_run_events.assert_not_called() diff --git a/tests/unit_tests/platform/test_routing_rules.py b/tests/unit_tests/platform/test_routing_rules.py index 0378a10be..65a2c6682 100644 --- a/tests/unit_tests/platform/test_routing_rules.py +++ b/tests/unit_tests/platform/test_routing_rules.py @@ -692,3 +692,73 @@ def test_websocket_task_override_does_not_mutate_bot_default(): assert pipeline_uuid == 'connection-pipeline' assert routed is False assert bot.bot_entity.use_pipeline_uuid == 'default-uuid' + + +@pytest.mark.asyncio +async def test_installed_event_processor_never_receives_unbound_events(): + from langbot_plugin.api.entities.builtin.platform.events import MemberJoinedEvent + + bot = TestEventRouteTrace._make_bot([]) + bot.ap = SimpleNamespace(plugin_connector=SimpleNamespace(emit_event=AsyncMock())) + bot._record_adapter_event = AsyncMock() + await bot._handle_platform_event(MemberJoinedEvent(), Mock()) + bot.ap.plugin_connector.emit_event.assert_not_called() + assert bot.logger.info.await_args.kwargs['metadata']['status'] == 'not_matched' + + +@pytest.mark.asyncio +async def test_bound_event_processor_receives_one_complete_typed_event(): + from langbot_plugin.api.entities.builtin.platform.events import MemberJoinedEvent + + bot = TestEventRouteTrace._make_bot( + [ + { + 'id': 'binding', + 'enabled': True, + 'event_pattern': 'group.member_joined', + 'target_type': 'event_processor', + 'target_uuid': 'processor-1', + 'priority': 0, + 'order': 0, + } + ] + ) + calls = [] + + async def run(envelope, binding, adapter_context=None): + calls.append((envelope, binding)) + if False: + yield None + + ref = 'event_processor:test/welcome/default' + bot.ap = SimpleNamespace( + workspace_service=active_workspace_service(), + agent_service=SimpleNamespace( + get_agent=AsyncMock( + return_value={ + 'uuid': 'processor-1', + 'kind': 'event_processor', + 'supported_event_patterns': ['group.member_joined'], + 'config': {'runner': {'id': ref}, 'runner_config': {ref: {'greeting': 'Hi'}}}, + } + ) + ), + agent_run_orchestrator=SimpleNamespace(run=run), + plugin_connector=SimpleNamespace(emit_event=AsyncMock()), + ) + bot._record_adapter_event = AsyncMock() + await bot._handle_platform_event( + MemberJoinedEvent( + member={'id': 'member-1'}, group={'id': 'group-1'}, source_platform_object={'private': 'opaque'} + ), + Mock(), + ) + assert len(calls) == 1 + envelope, binding = calls[0] + assert binding.binding_id == 'event_processor:processor-1' + assert binding.processor_type == 'event_processor' + assert binding.runner_config == {'greeting': 'Hi'} + assert envelope.data['member']['id'] == 'member-1' + assert envelope.data['type'] == 'group.member_joined' + assert 'source_platform_object' not in envelope.data + bot.ap.plugin_connector.emit_event.assert_not_called() diff --git a/uv.lock b/uv.lock index 607df7225..c93fba292 100644 --- a/uv.lock +++ b/uv.lock @@ -2119,7 +2119,7 @@ requires-dist = [ { name = "ebooklib", specifier = ">=0.18" }, { name = "gewechat-client", specifier = ">=0.1.5" }, { name = "html2text", specifier = ">=2024.2.26" }, - { name = "langbot-plugin", specifier = "==0.5.3" }, + { name = "langbot-plugin", git = "https://github.com/langbot-app/langbot-plugin-sdk?rev=23011398160cedcd4ef090054692bc2a4b34ee9f" }, { name = "langchain", specifier = ">=1.3.9" }, { name = "langchain-core", specifier = ">=1.3.3" }, { name = "langchain-text-splitters", specifier = ">=1.1.2" }, @@ -2185,8 +2185,8 @@ dev = [ [[package]] name = "langbot-plugin" -version = "0.5.3" -source = { registry = "https://pypi.org/simple" } +version = "0.5.5" +source = { git = "https://github.com/langbot-app/langbot-plugin-sdk?rev=23011398160cedcd4ef090054692bc2a4b34ee9f#23011398160cedcd4ef090054692bc2a4b34ee9f" } dependencies = [ { name = "aiofiles" }, { name = "aiohttp" }, @@ -2206,10 +2206,6 @@ dependencies = [ { name = "watchdog" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/55/1d/a54daa3bc699f5186b9970946c2ecf0e9cf219f77738934e04aaf5a0c20a/langbot_plugin-0.5.3.tar.gz", hash = "sha256:2324b1f7e1f55e3692e75c8b1e427ea497474b0150ec6ca83b49b5d77ec224c6", size = 472149 } -wheels = [ - { url = "https://files.pythonhosted.org/packages/0c/5f/ae6ed59773cc9d941fbb28b4ac07ce2a29e689d693e29ad71c40c5b39aa5/langbot_plugin-0.5.3-py3-none-any.whl", hash = "sha256:75dea1b6fb79ec6087ec3284f6698fb701d5feebf5f0318a7ae51fbd17a2f41f", size = 304559 }, -] [[package]] name = "langchain" diff --git a/web/src/app/home/agents/AgentDetailContent.tsx b/web/src/app/home/agents/AgentDetailContent.tsx index 6a3c50951..b27508687 100644 --- a/web/src/app/home/agents/AgentDetailContent.tsx +++ b/web/src/app/home/agents/AgentDetailContent.tsx @@ -23,6 +23,7 @@ import { DialogTitle, } from '@/components/ui/dialog'; import PipelineDetailContent from '@/app/home/pipelines/PipelineDetailContent'; +import EventProcessorDetailContent from './EventProcessorDetailContent'; import AgentCreateContent from './components/AgentCreateContent'; import AgentDebugPanel from './components/AgentDebugPanel'; import AgentFormComponent, { @@ -167,87 +168,104 @@ export default function AgentDetailContent({ id }: { id: string }) { return ( <> - - - {t('agents.noEventsConfiguredBadge')} - - ) : undefined - } - titleAction={ - canManage ? ( - setBasicInfoOpen(true)} /> - ) : undefined - } - status={runnerStatus} - saveLabel={t('common.save')} - saveFormId="agent-form" - canSave={canManage} - isDirty={formDirty} - isSaving={formSaving} - headerActions={ - canManage ? ( - - ) : undefined - } - configTitle={t('pipelines.configuration')} - configContent={ -
- { - if (updatedAgent) { - setAgent((current) => - current ? { ...current, ...updatedAgent } : current, - ); + {agent.kind === 'event_processor' ? ( + setDeleteConfirmOpen(true)} + onEdit={() => setBasicInfoOpen(true)} + onSaved={() => { + void httpClient + .getAgent(id) + .then((response) => setAgent(response.agent)); + void refreshPipelines(); + }} + /> + ) : ( + + + {t('agents.noEventsConfiguredBadge')} + + ) : undefined + } + titleAction={ + canManage ? ( + setBasicInfoOpen(true)} /> + ) : undefined + } + status={runnerStatus} + saveLabel={t('common.save')} + saveFormId="agent-form" + canSave={canManage} + isDirty={formDirty} + isSaving={formSaving} + headerActions={ + canManage ? ( + + ) : undefined + } + configTitle={t('pipelines.configuration')} + configContent={ +
+ { + if (updatedAgent) { + setAgent((current) => + current ? { ...current, ...updatedAgent } : current, + ); + } + refreshPipelines(); + }} + onDirtyChange={setFormDirty} + onSavingChange={setFormSaving} + onRunnerStatusChange={setRunnerStatus} + onSupportedEventPatternsChange={setSupportedEventPatterns} + onPlatformToolsChange={setPlatformTools} + /> +
+ } + debugTitle={canOperate ? t('agents.debugTab') : undefined} + debugDescription={t('agents.debugPlatformNotice')} + debugContent={ + canOperate ? ( + agentFormRef.current?.save() ?? false} + onOpenRunnerConfig={() => + agentFormRef.current?.openSection('runner_config') } - refreshPipelines(); - }} - onDirtyChange={setFormDirty} - onSavingChange={setFormSaving} - onRunnerStatusChange={setRunnerStatus} - onSupportedEventPatternsChange={setSupportedEventPatterns} - onPlatformToolsChange={setPlatformTools} - /> -
- } - debugTitle={canOperate ? t('agents.debugTab') : undefined} - debugDescription={t('agents.debugPlatformNotice')} - debugContent={ - canOperate ? ( - agentFormRef.current?.save() ?? false} - onOpenRunnerConfig={() => - agentFormRef.current?.openSection('runner_config') - } - supportedEventPatterns={supportedEventPatterns} - availableEventTypes={availableEventTypes} - /> - ) : undefined - } - unsavedLabel={t('pipelines.unsavedChanges')} - /> + supportedEventPatterns={supportedEventPatterns} + availableEventTypes={availableEventTypes} + /> + ) : undefined + } + unsavedLabel={t('pipelines.unsavedChanges')} + /> + )} void; + onEdit: () => void; + onSaved: () => void; +}) { + const { t } = useTranslation(); + const [platformTools, setPlatformTools] = useState([]); + const toolLabels = Object.fromEntries( + platformTools.map((tool) => [tool.name, extractI18nObject(tool.label)]), + ); + const [components, setComponents] = useState([]); + const [componentRef, setComponentRef] = useState(agent.component_ref ?? ''); + const initialParameters = + ( + (agent.config?.runner_config ?? {}) as Record< + string, + Record + > + )[agent.component_ref ?? ''] ?? {}; + const [parameters, setParameters] = useState(initialParameters); + const [runs, setRuns] = useState([]); + const [cursor, setCursor] = useState(null); + const [selected, setSelected] = useState(null); + const [events, setEvents] = useState([]); + const [eventCursor, setEventCursor] = useState(null); + const [loading, setLoading] = useState(true); + const [saving, setSaving] = useState(false); + const [pagingRuns, setPagingRuns] = useState(false); + const [pagingEvents, setPagingEvents] = useState(false); + const [configOpen, setConfigOpen] = useState(false); + const [failed, setFailed] = useState(false); + const validate = useRef<(() => Promise) | null>(null); + const requestVersion = useRef(0); + const available = components.some((item) => item.id === agent.component_ref); + + const load = useCallback(async () => { + setFailed(false); + try { + const [metadata, page] = await Promise.all([ + httpClient.getAgentMetadata(), + httpClient.getProcessorRuns(id), + ]); + setComponents(metadata.event_processors ?? []); + setPlatformTools(metadata.platform_tools ?? []); + setRuns(page.items); + setCursor(page.has_more ? page.next_cursor : null); + } catch { + setFailed(true); + } finally { + setLoading(false); + } + }, [id]); + useEffect(() => { + void load(); + }, [load]); + useEffect( + () => () => { + requestVersion.current += 1; + }, + [id], + ); + + async function openRun(run: ProcessorRun) { + const version = ++requestVersion.current; + setSelected(run); + setEvents([]); + setEventCursor(null); + try { + const page = await httpClient.getProcessorRunEvents(id, run.run_id); + if (version !== requestVersion.current) return; + setSelected(page.run); + setEvents(page.items); + setEventCursor(page.has_more ? page.next_cursor : null); + } catch { + if (version === requestVersion.current) + toast.error(t('agents.eventProcessor.loadError')); + } + } + + async function loadMoreEvents() { + if (!selected || eventCursor === null || pagingEvents) return; + setPagingEvents(true); + const runId = selected.run_id; + const version = requestVersion.current; + try { + const page = await httpClient.getProcessorRunEvents( + id, + runId, + eventCursor, + ); + if (version !== requestVersion.current) return; + setEvents((current) => [...current, ...page.items]); + setEventCursor(page.has_more ? page.next_cursor : null); + } catch { + toast.error(t('agents.eventProcessor.loadError')); + } finally { + setPagingEvents(false); + } + } + + async function loadMoreRuns() { + if (cursor === null || pagingRuns) return; + setPagingRuns(true); + try { + const page = await httpClient.getProcessorRuns(id, cursor); + setRuns((current) => [ + ...new Map( + [...current, ...page.items].map((run) => [run.run_id, run]), + ).values(), + ]); + setCursor(page.has_more ? page.next_cursor : null); + } catch { + toast.error(t('agents.eventProcessor.loadError')); + } finally { + setPagingRuns(false); + } + } + + useEffect(() => { + let cancelled = false; + let busy = false; + const timer = window.setInterval(async () => { + if (busy || document.hidden) return; + busy = true; + try { + const page = await httpClient.getProcessorRuns(id); + if (cancelled) return; + setRuns((current) => + [ + ...new Map( + [...current, ...page.items].map((run) => [run.run_id, run]), + ).values(), + ].sort((a, b) => b.created_at - a.created_at), + ); + if ( + selected && + !['completed', 'failed', 'cancelled'].includes(selected.status) && + eventCursor === null + ) { + const trace = await httpClient.getProcessorRunEvents( + id, + selected.run_id, + events.at(-1)?.sequence, + ); + if (cancelled) return; + setSelected(trace.run); + setEvents((current) => [ + ...new Map( + [...current, ...trace.items].map((event) => [ + event.sequence, + event, + ]), + ).values(), + ]); + setEventCursor(trace.has_more ? trace.next_cursor : null); + } + } catch { + /* Keep existing records visible across transient refresh failures. */ + } finally { + busy = false; + } + }, 3000); + return () => { + cancelled = true; + window.clearInterval(timer); + }; + }, [id, selected, eventCursor, events]); + + async function save() { + if (!componentRef || !((await validate.current?.()) ?? true)) return; + setSaving(true); + try { + await httpClient.updateAgent(id, { + component_ref: componentRef, + config: { + ...agent.config, + runner: { id: componentRef }, + runner_config: { [componentRef]: parameters }, + }, + }); + toast.success(t('agents.saveSuccess')); + onSaved(); + setConfigOpen(false); + await load(); + } catch { + toast.error(t('agents.saveError')); + } finally { + setSaving(false); + } + } + + function payload(value: unknown) { + return ( + 
+        {JSON.stringify(value, null, 2)}
+
+ ); + } + + return ( +
+
+ +

{agent.name}

+ {canManage && ( + + )} + {t('agents.eventProcessor.type')} + {!loading && !failed && !available && ( + + {t('agents.eventProcessor.unavailable')} + + )} +
+ + {canManage && ( + <> + + + + )} +
+
+

+ {agent.component_ref} +

+ {configOpen && ( +
+ { + setComponentRef(value); + setParameters({}); + validate.current = null; + }} + onParametersChange={setParameters} + onValidate={(fn) => { + validate.current = fn; + }} + /> + +
+ )} + {failed && ( +

+ {t('agents.eventProcessor.loadError')} +

+ )} +
+
+

+ {t('agents.eventProcessor.runs')} +

+ {loading ? ( +

{t('common.loading')}

+ ) : runs.length === 0 && !failed ? ( +
+

{t('agents.eventProcessor.noRuns')}

+ + {t('agents.eventProcessor.bindBot')} + +
+ ) : ( + runs.map((run) => ( + + )) + )} + {cursor !== null && ( + + )} +
+
+

+ {t('agents.eventProcessor.trace')} +

+ {!selected ? ( +

+ {t('agents.eventProcessor.selectRun')} +

+ ) : ( +
+
+ + {t('agents.eventProcessor.input')} + + {payload(selected.metadata.input_event)} +
+ {selected.metadata.delivery != null && ( +
+ + {t('agents.eventProcessor.destination')} + + {payload(selected.metadata.delivery)} +
+ )} + {events.map((event) => + event.type === 'processor.log' ? ( +
+ + {String(event.data.level)} + + + {String(event.data.text)} + +
+ ) : ( +
+ + {t( + `agents.eventProcessor.trace_${event.type.replaceAll('.', '_')}`, + { defaultValue: event.type }, + )} + {typeof event.data.tool_name === 'string' && ( + + {toolLabels[event.data.tool_name] || + event.data.tool_name} + + )} + + {payload(event.data)} +
+ ), + )} + {selected.status === 'failed' && selected.status_reason && ( +

+ {selected.status_reason} +

+ )} + {eventCursor !== null && ( + + )} +
+ )} +
+
+
+ ); +} diff --git a/web/src/app/home/agents/components/AgentCreateContent.tsx b/web/src/app/home/agents/components/AgentCreateContent.tsx index 65d1b37b6..c212e34f4 100644 --- a/web/src/app/home/agents/components/AgentCreateContent.tsx +++ b/web/src/app/home/agents/components/AgentCreateContent.tsx @@ -1,12 +1,13 @@ -import { useState } from 'react'; +import { useEffect, useRef, useState } from 'react'; import { useForm } from 'react-hook-form'; import { zodResolver } from '@hookform/resolvers/zod'; import { z } from 'zod'; import { useTranslation } from 'react-i18next'; import { toast } from 'sonner'; -import { Bot, Workflow } from 'lucide-react'; +import { Bot, Workflow, FileCode2 } from 'lucide-react'; import { httpClient } from '@/app/infra/http/HttpClient'; -import { AgentKind } from '@/app/infra/entities/api'; +import { AgentKind, EventProcessorDescriptor } from '@/app/infra/entities/api'; +import EventProcessorSettings from './EventProcessorSettings'; import { Button } from '@/components/ui/button'; import { ToggleGroup, ToggleGroupItem } from '@/components/ui/toggle-group'; import { @@ -35,6 +36,23 @@ export default function AgentCreateContent({ }) { const { t } = useTranslation(); const [kind, setKind] = useState('agent'); + const [components, setComponents] = useState([]); + const [componentRef, setComponentRef] = useState(''); + const [parameters, setParameters] = useState>({}); + const validateParameters = useRef<(() => Promise) | null>(null); + useEffect(() => { + if (kind !== 'event_processor') return; + let cancelled = false; + httpClient + .getAgentMetadata() + .then((metadata) => { + if (!cancelled) setComponents(metadata.event_processors ?? []); + }) + .catch(() => toast.error(t('agents.eventProcessor.loadError'))); + return () => { + cancelled = true; + }; + }, [kind, t]); const formSchema = z.object({ name: z.string().min(1, { message: t('agents.nameRequired') }), description: z.string().optional(), @@ -51,8 +69,14 @@ export default function AgentCreateContent({ }); function handleKindChange(nextKind: AgentKind) { - const previousDefaultEmoji = kind === 'pipeline' ? '⚙️' : '🤖'; - const nextDefaultEmoji = nextKind === 'pipeline' ? '⚙️' : '🤖'; + const previousDefaultEmoji = + kind === 'pipeline' ? '⚙️' : kind === 'event_processor' ? '⚡' : '🤖'; + const nextDefaultEmoji = + nextKind === 'pipeline' + ? '⚙️' + : nextKind === 'event_processor' + ? '⚡' + : '🤖'; setKind(nextKind); const currentEmoji = form.getValues('emoji'); if (!currentEmoji || currentEmoji === previousDefaultEmoji) { @@ -60,10 +84,24 @@ export default function AgentCreateContent({ } } - function handleSubmit(values: FormValues) { + async function handleSubmit(values: FormValues) { + if ( + kind === 'event_processor' && + (!componentRef || !((await validateParameters.current?.()) ?? true)) + ) + return; httpClient .createAgent({ kind, + ...(kind === 'event_processor' + ? { + component_ref: componentRef, + config: { + runner: { id: componentRef }, + runner_config: { [componentRef]: parameters }, + }, + } + : {}), name: values.name, description: values.description ?? '', emoji: values.emoji || (kind === 'pipeline' ? '⚙️' : '🤖'), @@ -90,13 +128,26 @@ export default function AgentCreateContent({ title: t('agents.pipelineType'), description: t('agents.pipelineTypeDescription'), }, + { + kind: 'event_processor' as const, + icon: FileCode2, + title: t('agents.eventProcessor.type'), + description: t('agents.eventProcessor.description'), + }, ]; return (

{t('agents.create')}

-
@@ -158,6 +209,22 @@ export default function AgentCreateContent({ + {kind === 'event_processor' && ( + { + setComponentRef(value); + setParameters({}); + validateParameters.current = null; + }} + onParametersChange={setParameters} + onValidate={(validate) => { + validateParameters.current = validate; + }} + /> + )} {t('agents.basicInfo')} diff --git a/web/src/app/home/agents/components/EventProcessorSettings.tsx b/web/src/app/home/agents/components/EventProcessorSettings.tsx new file mode 100644 index 000000000..962c13d36 --- /dev/null +++ b/web/src/app/home/agents/components/EventProcessorSettings.tsx @@ -0,0 +1,91 @@ +import { useTranslation } from 'react-i18next'; +import { Link } from 'react-router-dom'; +import type { EventProcessorDescriptor } from '@/app/infra/entities/api'; +import { extractI18nObject } from '@/i18n/I18nProvider'; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from '@/components/ui/select'; +import DynamicFormComponent from '@/app/home/components/dynamic-form/DynamicFormComponent'; + +export default function EventProcessorSettings({ + components, + value, + parameters, + onChange, + onParametersChange, + onValidate, +}: { + components: EventProcessorDescriptor[]; + value: string; + parameters: Record; + onChange: (value: string) => void; + onParametersChange: (value: Record) => void; + onValidate?: (validate: () => Promise) => void; +}) { + const { t } = useTranslation(); + const selected = components.find((item) => item.id === value); + return ( +
+
+ + + {components.length === 0 && ( +

+ {t('agents.eventProcessor.noComponents')}{' '} + + {t('agents.eventProcessor.installPlugin')} + +

+ )} +
+ {selected && ( +

+ {selected.supported_event_patterns.join(' · ')} +

+ )} + {selected && selected.config_schema.length > 0 && ( + + onParametersChange(values as Record) + } + onValidate={onValidate} + /> + )} +
+ ); +} diff --git a/web/src/app/home/agents/components/ProcessorTypeDiagram.tsx b/web/src/app/home/agents/components/ProcessorTypeDiagram.tsx index 53380b0b2..88ff62000 100644 --- a/web/src/app/home/agents/components/ProcessorTypeDiagram.tsx +++ b/web/src/app/home/agents/components/ProcessorTypeDiagram.tsx @@ -374,5 +374,29 @@ function PipelineDiagram() { } export default function ProcessorTypeDiagram({ kind }: { kind: AgentKind }) { + const { t } = useTranslation(); + if (kind === 'event_processor') + return ( +
+

+ {t('agents.eventProcessor.type')} +

+

+ {t('agents.eventProcessor.description')} +

+ {['input', 'component', 'trace'].map((step, index) => ( +
+ {index + 1} + {t(`agents.eventProcessor.${step}`)} +
+ ))} +

+ {t('agents.eventProcessor.activation')} +

+
+ ); return kind === 'agent' ? : ; } diff --git a/web/src/app/home/bots/components/bot-form/BotForm.tsx b/web/src/app/home/bots/components/bot-form/BotForm.tsx index 7ebb143e1..baf8e2c76 100644 --- a/web/src/app/home/bots/components/bot-form/BotForm.tsx +++ b/web/src/app/home/bots/components/bot-form/BotForm.tsx @@ -75,7 +75,12 @@ const getFormSchema = (t: (key: string) => string) => z.object({ id: z.string().optional(), event_pattern: z.string(), - target_type: z.enum(['agent', 'pipeline', 'discard']), + target_type: z.enum([ + 'agent', + 'pipeline', + 'event_processor', + 'discard', + ]), target_uuid: z.string(), filters: z.array(z.record(z.string(), z.any())).optional(), priority: z.number(), 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 2b01df6c0..d4c6c2d96 100644 --- a/web/src/app/home/bots/components/bot-form/EventBindingsEditor.tsx +++ b/web/src/app/home/bots/components/bot-form/EventBindingsEditor.tsx @@ -35,6 +35,7 @@ import { UserMinus, UserPlus, Workflow, + FileCode2, XCircle, } from 'lucide-react'; import { Button } from '@/components/ui/button'; @@ -518,6 +519,11 @@ function TargetCombobox({ const pipelines = pipelineAllowed ? agentOptions.filter((a) => a.kind === 'pipeline') : []; + const eventProcessors = agentOptions.filter( + (item) => + item.kind === 'event_processor' && + agentSupportsEventPattern(item, binding.event_pattern), + ); function currentLabel() { if (targetType === 'discard') @@ -531,7 +537,9 @@ function TargetCombobox({ if (agent) return ( - {agent.kind === 'pipeline' ? ( + {agent.kind === 'event_processor' ? ( + + ) : agent.kind === 'pipeline' ? ( ) : ( @@ -585,6 +593,26 @@ function TargetCombobox({ ))} )} + {eventProcessors.length > 0 && ( + + {eventProcessors.map((item) => ( + + select(encodeTarget('event_processor', item.uuid || '')) + } + > + + {targetLabel(item)} + {current === + encodeTarget('event_processor', item.uuid || '') && ( + + )} + + ))} + + )} {pipelines.length > 0 && ( {pipelines.map((a) => ( diff --git a/web/src/app/home/components/home-sidebar/HomeSidebar.tsx b/web/src/app/home/components/home-sidebar/HomeSidebar.tsx index 0fddd1b2a..28d7de295 100644 --- a/web/src/app/home/components/home-sidebar/HomeSidebar.tsx +++ b/web/src/app/home/components/home-sidebar/HomeSidebar.tsx @@ -730,13 +730,16 @@ function NavItems({ const showAgentGroupHeaders = isAgents && !inPopover && sidebarData.agentsGroupByKind; - const agentGroupOrder: Array<'agent' | 'pipeline'> = [ - 'agent', - 'pipeline', - ]; - const agentGroupLabelKey: Record<'agent' | 'pipeline', string> = { + const agentGroupOrder: Array< + 'agent' | 'pipeline' | 'event_processor' + > = ['agent', 'pipeline', 'event_processor']; + const agentGroupLabelKey: Record< + 'agent' | 'pipeline' | 'event_processor', + string + > = { agent: 'agents.kindBadgeAgent', pipeline: 'agents.kindBadgePipeline', + event_processor: 'agents.eventProcessor.type', }; const groupOrder: Array<'plugin' | 'mcp' | 'skill'> = [ @@ -889,12 +892,16 @@ function NavItems({ - {item.kind === 'pipeline' ? ( + {item.kind === 'event_processor' ? ( + ⚡ + ) : item.kind === 'pipeline' ? ( ) : ( diff --git a/web/src/app/home/components/home-sidebar/SidebarDataContext.tsx b/web/src/app/home/components/home-sidebar/SidebarDataContext.tsx index 18e915a74..871d4d7b4 100644 --- a/web/src/app/home/components/home-sidebar/SidebarDataContext.tsx +++ b/web/src/app/home/components/home-sidebar/SidebarDataContext.tsx @@ -31,7 +31,7 @@ export interface SidebarEntityItem { // Set when this item appears in the unified extensions list extensionType?: 'plugin' | 'mcp' | 'skill'; // Agent-specific: distinguishes Agent processors from Pipelines - kind?: 'agent' | 'pipeline'; + kind?: 'agent' | 'pipeline' | 'event_processor'; } // Plugin page registered by a plugin diff --git a/web/src/app/infra/entities/api/index.ts b/web/src/app/infra/entities/api/index.ts index c44a7185f..af5b67b9c 100644 --- a/web/src/app/infra/entities/api/index.ts +++ b/web/src/app/infra/entities/api/index.ts @@ -162,7 +162,43 @@ export interface ApiRespPipelines { pipelines: Pipeline[]; } -export type AgentKind = 'agent' | 'pipeline'; +export type AgentKind = 'agent' | 'pipeline' | 'event_processor'; + +export interface EventProcessorDescriptor { + id: string; + label: Record; + plugin_author: string; + plugin_name: string; + config_schema: import('../form/dynamic').IDynamicFormItemSchema[]; + supported_event_patterns: string[]; +} + +export interface ProcessorRun { + run_id: string; + status: string; + status_reason?: string; + created_at: number; + metadata: { event_type?: string; input_event?: unknown; delivery?: unknown }; +} + +export interface ProcessorRunEvent { + sequence: number; + type: string; + data: Record; +} + +export interface ProcessorRunPage { + items: ProcessorRun[]; + next_cursor: number | null; + has_more: boolean; +} + +export interface ProcessorRunEventPage { + run: ProcessorRun; + items: ProcessorRunEvent[]; + next_cursor: number | null; + has_more: boolean; +} export interface AgentCapability { supported_event_patterns: string[]; @@ -192,6 +228,7 @@ export interface ApiRespAgent { } export interface GetAgentMetadataResponseData { + event_processors?: EventProcessorDescriptor[]; runner_config?: PipelineConfigTab; platform_tools: AgentPlatformTool[]; host_tools?: PluginTool[] | null; @@ -274,7 +311,7 @@ export interface Bot { export interface EventBinding { id?: string; event_pattern: string; - target_type: 'agent' | 'pipeline' | 'discard'; + target_type: AgentKind | 'discard'; target_uuid: string; filters?: Array>; priority: number; diff --git a/web/src/app/infra/http/BackendClient.ts b/web/src/app/infra/http/BackendClient.ts index c5fe9b7b7..8267c4cb5 100644 --- a/web/src/app/infra/http/BackendClient.ts +++ b/web/src/app/infra/http/BackendClient.ts @@ -268,6 +268,25 @@ export class BackendClient extends BaseHttpClient { return this.get('/api/v1/agents/_/metadata'); } + public getProcessorRuns( + uuid: string, + beforeId?: number, + ): Promise { + return this.get( + `/api/v1/agents/${encodeURIComponent(uuid)}/runs${beforeId === undefined ? '' : `?before_id=${beforeId}`}`, + ); + } + + public getProcessorRunEvents( + uuid: string, + runId: string, + afterSequence?: number, + ): Promise { + return this.get( + `/api/v1/agents/${encodeURIComponent(uuid)}/runs/${encodeURIComponent(runId)}/events${afterSequence === undefined ? '' : `?after_sequence=${afterSequence}`}`, + ); + } + public createAgent(agent: Agent): Promise<{ uuid: string; kind: string }> { return this.post('/api/v1/agents', agent); } diff --git a/web/src/i18n/locales/en-US.ts b/web/src/i18n/locales/en-US.ts index 4fe7cdf79..3fddf6f35 100644 --- a/web/src/i18n/locales/en-US.ts +++ b/web/src/i18n/locales/en-US.ts @@ -395,20 +395,19 @@ const enUS = { commonScenarios: 'Common scenarios', dragEventRoute: 'Drag route {{index}}', behaviorReplyMessages: 'Reply to messages', - behaviorReplyMessagesDescription: - 'Send incoming messages to an Agent or Pipeline.', + behaviorReplyMessagesDescription: 'Send incoming messages to a processor.', behaviorWelcomeMembers: 'Welcome new members', behaviorWelcomeMembersDescription: - 'Run an Agent when someone joins a group.', + 'Run a processor when someone joins a group.', behaviorHandleDepartures: 'Handle member departures', behaviorHandleDeparturesDescription: - 'Run an Agent when someone leaves or is removed.', + 'Run a processor when someone leaves or is removed.', behaviorReviewFriendRequests: 'Review friend requests', behaviorReviewFriendRequestsDescription: - 'Let an Agent decide how to handle a new request.', + 'Send new friend requests to a processor.', behaviorHandleModeration: 'Handle moderation events', behaviorHandleModerationDescription: - 'Run an Agent when a group member is restricted.', + 'Run a processor when a group member is restricted.', behaviorCustom: 'Configure another event', behaviorCustomDescription: 'Add a route and choose from every event supported by this adapter.', @@ -704,6 +703,36 @@ const enUS = { }, }, agents: { + eventProcessor: { + type: 'Event processor', + description: 'Process platform events using plugin code.', + component: 'Plugin component', + selectComponent: 'Select an event processor', + unavailable: 'Component unavailable', + noComponents: 'No event processor components installed.', + installPlugin: 'Install a plugin', + loadError: 'Unable to load processor details.', + refresh: 'Refresh', + runs: 'Runs', + noRuns: 'No runs yet. Bind a Bot event to start.', + bindBot: 'Bind Bot events', + trace: 'Logs and message flow', + selectRun: 'Select a run to view details.', + input: 'Incoming event', + destination: 'Delivery destination', + loadMore: 'Load more', + activation: 'Install a plugin, create an instance, then bind Bot events.', + status_pending: 'Pending', + status_running: 'Running', + status_completed: 'Completed', + status_failed: 'Failed', + status_cancelled: 'Cancelled', + status_queued: 'Queued', + trace_run_completed: 'Run completed', + trace_run_failed: 'Run failed', + trace_tool_call_started: 'Action started', + trace_tool_call_completed: 'Action result', + }, debugData: { title: 'Event data', form: 'Common fields', @@ -735,7 +764,7 @@ const enUS = { description: 'Create reusable processors and use them in bot event routing', create: 'Create Processor', editAgent: 'Edit Agent', - selectFromSidebar: 'Select an Agent or Pipeline from the sidebar', + selectFromSidebar: 'Select a processor from the sidebar', agentType: 'Agent', agentTypeDescription: 'Use a runner to handle messages, group members, friends, feedback, and other platform events. Best for scenarios that need autonomous decisions, tool use, or non-message events.', diff --git a/web/src/i18n/locales/es-ES.ts b/web/src/i18n/locales/es-ES.ts index c2f4a6643..6e27b9474 100644 --- a/web/src/i18n/locales/es-ES.ts +++ b/web/src/i18n/locales/es-ES.ts @@ -506,6 +506,37 @@ const esES = { }, }, agents: { + eventProcessor: { + type: 'Procesador de eventos', + description: 'Procesa eventos con código del plugin.', + component: 'Componente del plugin', + selectComponent: 'Seleccionar un procesador', + unavailable: 'Componente no disponible', + noComponents: 'No hay componentes de eventos instalados.', + installPlugin: 'Instalar un plugin', + loadError: 'No se pudieron cargar los detalles.', + refresh: 'Actualizar', + runs: 'Ejecuciones', + noRuns: 'Sin ejecuciones. Vincula eventos de un Bot para empezar.', + bindBot: 'Vincular eventos del Bot', + trace: 'Registros y flujo de mensajes', + selectRun: 'Selecciona una ejecución para ver los detalles.', + input: 'Evento recibido', + destination: 'Destino de entrega', + loadMore: 'Cargar más', + activation: + 'Instala un plugin, crea una instancia y vincula eventos del Bot.', + status_pending: 'Pendiente', + status_running: 'En ejecución', + status_completed: 'Completado', + status_failed: 'Error', + status_cancelled: 'Cancelado', + status_queued: 'En cola', + trace_run_completed: 'Ejecución completada', + trace_run_failed: 'Ejecución fallida', + trace_tool_call_started: 'Acción iniciada', + trace_tool_call_completed: 'Resultado de la acción', + }, debugData: { title: 'Datos del evento', form: 'Campos comunes', @@ -538,7 +569,7 @@ const esES = { 'Crea procesadores reutilizables y úsalos en el enrutamiento de eventos del bot', create: 'Crear procesador', editAgent: 'Editar Agent', - selectFromSidebar: 'Selecciona un Agent o Pipeline desde la barra lateral', + selectFromSidebar: 'Selecciona un procesador en la barra lateral', agentType: 'Agent', agentTypeDescription: 'Usa un runner para procesar mensajes, miembros de grupo, amigos, retroalimentación y otros eventos de plataforma.', diff --git a/web/src/i18n/locales/ja-JP.ts b/web/src/i18n/locales/ja-JP.ts index 958d879fe..d4d3470cb 100644 --- a/web/src/i18n/locales/ja-JP.ts +++ b/web/src/i18n/locales/ja-JP.ts @@ -402,19 +402,19 @@ const jaJP = { dragEventRoute: 'ルート {{index}} をドラッグ', behaviorReplyMessages: '受信メッセージに返信', behaviorReplyMessagesDescription: - '受信メッセージを Agent または Pipeline で処理します。', + '受信メッセージをプロセッサーに渡します。', behaviorWelcomeMembers: '新しいメンバーを歓迎', behaviorWelcomeMembersDescription: - 'メンバーがグループに参加したときに Agent を実行します。', + 'グループへの参加時にプロセッサーを実行します。', behaviorHandleDepartures: 'メンバーの退出を処理', behaviorHandleDeparturesDescription: - 'メンバーが退出または削除されたときに Agent を実行します。', + 'グループからの退出時にプロセッサーを実行します。', behaviorReviewFriendRequests: '友だち申請を確認', behaviorReviewFriendRequestsDescription: - '新しい申請の処理方法を Agent に判断させます。', + '新しい友達リクエストをプロセッサーに渡します。', behaviorHandleModeration: 'モデレーションイベントを処理', behaviorHandleModerationDescription: - 'グループメンバーが制限されたときに Agent を実行します。', + 'グループメンバーの制限時にプロセッサーを実行します。', behaviorCustom: '別のイベントを設定', behaviorCustomDescription: 'ルートを追加し、このアダプターが対応する全イベントから選択します。', @@ -716,6 +716,37 @@ const jaJP = { }, }, agents: { + eventProcessor: { + type: 'イベントプロセッサー', + description: 'プラグインのコードでイベントを処理します。', + component: 'プラグインコンポーネント', + selectComponent: 'イベントプロセッサーを選択', + unavailable: 'コンポーネントを利用できません', + noComponents: 'イベントプロセッサーがインストールされていません。', + installPlugin: 'プラグインをインストール', + loadError: '詳細を読み込めません。', + refresh: '更新', + runs: '実行履歴', + noRuns: '実行履歴はありません。Bot イベントを紐付けて開始します。', + bindBot: 'Bot イベントを紐付ける', + trace: 'ログとメッセージの流れ', + selectRun: '実行履歴を選択して詳細を表示します。', + input: '受信イベント', + destination: '送信先', + loadMore: 'さらに読み込む', + activation: + 'プラグインをインストールし、インスタンスを作成して Bot イベントを紐付けます。', + status_pending: '待機中', + status_running: '実行中', + status_completed: '完了', + status_failed: '失敗', + status_cancelled: 'キャンセル済み', + status_queued: 'キュー待ち', + trace_run_completed: '実行完了', + trace_run_failed: '実行失敗', + trace_tool_call_started: 'アクション開始', + trace_tool_call_completed: 'アクション結果', + }, debugData: { title: 'イベントデータ', form: '基本項目', @@ -772,8 +803,7 @@ const jaJP = { '再利用可能なプロセッサーを作成し、ボットのイベントルーティングで使用します', create: 'プロセッサーを作成', editAgent: 'Agent を編集', - selectFromSidebar: - 'サイドバーから Agent または Pipeline を選択してください', + selectFromSidebar: 'サイドバーからプロセッサーを選択', agentType: 'Agent', agentTypeDescription: 'Runner を使ってメッセージ、グループメンバー、友だち、フィードバックなどのプラットフォームイベントを処理します。自律的な判断、ツール利用、メッセージ以外のイベント対応が必要な場合に適しています。', diff --git a/web/src/i18n/locales/ru-RU.ts b/web/src/i18n/locales/ru-RU.ts index f9880e911..b9b41431c 100644 --- a/web/src/i18n/locales/ru-RU.ts +++ b/web/src/i18n/locales/ru-RU.ts @@ -503,6 +503,37 @@ const ruRU = { }, }, agents: { + eventProcessor: { + type: 'Обработчик событий', + description: 'Обрабатывает события кодом плагина.', + component: 'Компонент плагина', + selectComponent: 'Выберите обработчик', + unavailable: 'Компонент недоступен', + noComponents: 'Компоненты обработки событий не установлены.', + installPlugin: 'Установить плагин', + loadError: 'Не удалось загрузить данные.', + refresh: 'Обновить', + runs: 'Запуски', + noRuns: 'Запусков пока нет. Привяжите события бота.', + bindBot: 'Привязать события бота', + trace: 'Журнал и поток сообщений', + selectRun: 'Выберите запуск для просмотра.', + input: 'Входящее событие', + destination: 'Получатель', + loadMore: 'Загрузить ещё', + activation: + 'Установите плагин, создайте экземпляр и привяжите события бота.', + status_pending: 'Ожидание', + status_running: 'Выполняется', + status_completed: 'Завершено', + status_failed: 'Ошибка', + status_cancelled: 'Отменено', + status_queued: 'В очереди', + trace_run_completed: 'Выполнение завершено', + trace_run_failed: 'Ошибка выполнения', + trace_tool_call_started: 'Действие начато', + trace_tool_call_completed: 'Результат действия', + }, debugData: { title: 'Данные события', form: 'Основные поля', @@ -535,7 +566,7 @@ const ruRU = { 'Создавайте переиспользуемые обработчики и используйте их в маршрутизации событий бота', create: 'Создать обработчик', editAgent: 'Редактировать Agent', - selectFromSidebar: 'Выберите Agent или Pipeline на боковой панели', + selectFromSidebar: 'Выберите обработчик на боковой панели', agentType: 'Agent', agentTypeDescription: 'Используйте runner для обработки сообщений, участников групп, друзей, обратной связи и других событий платформы.', diff --git a/web/src/i18n/locales/th-TH.ts b/web/src/i18n/locales/th-TH.ts index da70384df..f492bc9f4 100644 --- a/web/src/i18n/locales/th-TH.ts +++ b/web/src/i18n/locales/th-TH.ts @@ -490,6 +490,36 @@ const thTH = { }, }, agents: { + eventProcessor: { + type: 'ตัวประมวลผลเหตุการณ์', + description: 'ประมวลผลเหตุการณ์ด้วยโค้ดปลั๊กอิน', + component: 'ส่วนประกอบปลั๊กอิน', + selectComponent: 'เลือกตัวประมวลผลเหตุการณ์', + unavailable: 'ส่วนประกอบไม่พร้อมใช้งาน', + noComponents: 'ยังไม่ได้ติดตั้งส่วนประกอบประมวลผลเหตุการณ์', + installPlugin: 'ติดตั้งปลั๊กอิน', + loadError: 'ไม่สามารถโหลดรายละเอียดได้', + refresh: 'รีเฟรช', + runs: 'ประวัติการทำงาน', + noRuns: 'ยังไม่มีการทำงาน เชื่อมโยงเหตุการณ์บอทเพื่อเริ่มต้น', + bindBot: 'เชื่อมโยงเหตุการณ์บอท', + trace: 'บันทึกและเส้นทางข้อความ', + selectRun: 'เลือกการทำงานเพื่อดูรายละเอียด', + input: 'เหตุการณ์ขาเข้า', + destination: 'ปลายทางการส่ง', + loadMore: 'โหลดเพิ่มเติม', + activation: 'ติดตั้งปลั๊กอิน สร้างอินสแตนซ์ แล้วเชื่อมโยงเหตุการณ์บอท', + status_pending: 'รอดำเนินการ', + status_running: 'กำลังทำงาน', + status_completed: 'เสร็จสิ้น', + status_failed: 'ล้มเหลว', + status_cancelled: 'ยกเลิกแล้ว', + status_queued: 'อยู่ในคิว', + trace_run_completed: 'ทำงานเสร็จสิ้น', + trace_run_failed: 'การทำงานล้มเหลว', + trace_tool_call_started: 'เริ่มดำเนินการ', + trace_tool_call_completed: 'ผลการดำเนินการ', + }, debugData: { title: 'ข้อมูลเหตุการณ์', form: 'ฟิลด์ทั่วไป', @@ -521,7 +551,7 @@ const thTH = { description: 'สร้างตัวประมวลผลที่ใช้ซ้ำได้และใช้ในเส้นทางเหตุการณ์ของบอท', create: 'สร้างตัวประมวลผล', editAgent: 'แก้ไข Agent', - selectFromSidebar: 'เลือก Agent หรือ Pipeline จากแถบด้านข้าง', + selectFromSidebar: 'เลือกตัวประมวลผลจากแถบด้านข้าง', agentType: 'Agent', agentTypeDescription: 'ใช้ runner เพื่อประมวลผลข้อความ สมาชิกกลุ่ม เพื่อน ฟีดแบ็ก และเหตุการณ์แพลตฟอร์มอื่นๆ', diff --git a/web/src/i18n/locales/vi-VN.ts b/web/src/i18n/locales/vi-VN.ts index 1316cdb75..066f00cd2 100644 --- a/web/src/i18n/locales/vi-VN.ts +++ b/web/src/i18n/locales/vi-VN.ts @@ -499,6 +499,36 @@ const viVN = { }, }, agents: { + eventProcessor: { + type: 'Bộ xử lý sự kiện', + description: 'Xử lý sự kiện bằng mã plugin.', + component: 'Thành phần plugin', + selectComponent: 'Chọn bộ xử lý sự kiện', + unavailable: 'Thành phần không khả dụng', + noComponents: 'Chưa cài thành phần xử lý sự kiện.', + installPlugin: 'Cài plugin', + loadError: 'Không thể tải chi tiết.', + refresh: 'Làm mới', + runs: 'Lịch sử chạy', + noRuns: 'Chưa có lần chạy nào. Liên kết sự kiện Bot để bắt đầu.', + bindBot: 'Liên kết sự kiện Bot', + trace: 'Nhật ký và luồng tin nhắn', + selectRun: 'Chọn một lần chạy để xem chi tiết.', + input: 'Sự kiện đầu vào', + destination: 'Đích gửi', + loadMore: 'Tải thêm', + activation: 'Cài plugin, tạo phiên bản rồi liên kết sự kiện Bot.', + status_pending: 'Đang chờ', + status_running: 'Đang chạy', + status_completed: 'Hoàn tất', + status_failed: 'Thất bại', + status_cancelled: 'Đã hủy', + status_queued: 'Trong hàng đợi', + trace_run_completed: 'Chạy hoàn tất', + trace_run_failed: 'Chạy thất bại', + trace_tool_call_started: 'Bắt đầu hành động', + trace_tool_call_completed: 'Kết quả hành động', + }, debugData: { title: 'Dữ liệu sự kiện', form: 'Trường thường dùng', @@ -531,7 +561,7 @@ const viVN = { 'Tạo bộ xử lý có thể tái sử dụng và dùng chúng trong định tuyến sự kiện của bot', create: 'Tạo bộ xử lý', editAgent: 'Chỉnh sửa Agent', - selectFromSidebar: 'Chọn một Agent hoặc Pipeline từ thanh bên', + selectFromSidebar: 'Chọn bộ xử lý từ thanh bên', agentType: 'Agent', agentTypeDescription: 'Dùng runner để xử lý tin nhắn, thành viên nhóm, bạn bè, phản hồi và các sự kiện nền tảng khác.', diff --git a/web/src/i18n/locales/zh-Hans.ts b/web/src/i18n/locales/zh-Hans.ts index 8040bf922..974d6c14d 100644 --- a/web/src/i18n/locales/zh-Hans.ts +++ b/web/src/i18n/locales/zh-Hans.ts @@ -378,17 +378,15 @@ const zhHans = { commonScenarios: '常用场景', dragEventRoute: '拖动第 {{index}} 条路由', behaviorReplyMessages: '回复收到的消息', - behaviorReplyMessagesDescription: - '把收到的消息交给 Agent 或 Pipeline 处理。', + behaviorReplyMessagesDescription: '把收到的消息交给处理器。', behaviorWelcomeMembers: '欢迎新成员', - behaviorWelcomeMembersDescription: '有人加入群组时运行 Agent。', + behaviorWelcomeMembersDescription: '有人加入群组时运行处理器。', behaviorHandleDepartures: '处理成员离群', - behaviorHandleDeparturesDescription: '有人离开或被移出群组时运行 Agent。', + behaviorHandleDeparturesDescription: '有人离开或被移出群组时运行处理器。', behaviorReviewFriendRequests: '审核好友请求', - behaviorReviewFriendRequestsDescription: - '让 Agent 决定如何处理新的好友请求。', + behaviorReviewFriendRequestsDescription: '把新的好友请求交给处理器。', behaviorHandleModeration: '处理群管理事件', - behaviorHandleModerationDescription: '群成员受到限制时运行 Agent。', + behaviorHandleModerationDescription: '群成员受到限制时运行处理器。', behaviorCustom: '配置其他事件', behaviorCustomDescription: '添加路由，并从此适配器支持的全部事件中选择。', eventPattern: '事件', @@ -670,6 +668,36 @@ const zhHans = { }, }, agents: { + eventProcessor: { + type: '事件处理器', + description: '通过插件代码处理平台事件。', + component: '插件组件', + selectComponent: '选择事件处理器组件', + unavailable: '组件不可用', + noComponents: '尚未安装事件处理器组件。', + installPlugin: '安装插件', + loadError: '无法加载处理器详情。', + refresh: '刷新', + runs: '运行记录', + noRuns: '暂无运行记录，绑定机器人事件后开始处理。', + bindBot: '绑定机器人事件', + trace: '日志与消息流向', + selectRun: '选择一条运行记录查看详情。', + input: '传入事件', + destination: '投递目标', + loadMore: '加载更多', + activation: '安装插件，创建实例，再绑定机器人事件。', + status_pending: '待执行', + status_running: '运行中', + status_completed: '已完成', + status_failed: '失败', + status_cancelled: '已取消', + status_queued: '排队中', + trace_run_completed: '运行完成', + trace_run_failed: '运行失败', + trace_tool_call_started: '开始执行动作', + trace_tool_call_completed: '动作结果', + }, debugData: { title: '事件数据', form: '常用字段', @@ -701,7 +729,7 @@ const zhHans = { description: '创建可复用的处理器，并在机器人事件路由中使用', create: '创建处理器', editAgent: '编辑 Agent', - selectFromSidebar: '从侧边栏选择一个 Agent 或 Pipeline', + selectFromSidebar: '从侧边栏选择一个处理器', agentType: 'Agent', agentTypeDescription: '通过运行器处理消息、群成员、好友、反馈等平台事件。适合需要自主判断、调用工具或响应非消息事件的场景。', diff --git a/web/src/i18n/locales/zh-Hant.ts b/web/src/i18n/locales/zh-Hant.ts index 8c0e6a76a..3e72e1aa1 100644 --- a/web/src/i18n/locales/zh-Hant.ts +++ b/web/src/i18n/locales/zh-Hant.ts @@ -474,6 +474,36 @@ const zhHant = { }, }, agents: { + eventProcessor: { + type: '事件處理器', + description: '透過外掛程式碼處理平台事件。', + component: '外掛元件', + selectComponent: '選擇事件處理器元件', + unavailable: '元件無法使用', + noComponents: '尚未安裝事件處理器元件。', + installPlugin: '安裝外掛', + loadError: '無法載入處理器詳情。', + refresh: '重新整理', + runs: '執行記錄', + noRuns: '尚無執行記錄，綁定機器人事件後開始處理。', + bindBot: '綁定機器人事件', + trace: '日誌與訊息流向', + selectRun: '選擇一筆執行記錄查看詳情。', + input: '傳入事件', + destination: '傳送目標', + loadMore: '載入更多', + activation: '安裝外掛、建立實例，再綁定機器人事件。', + status_pending: '待執行', + status_running: '執行中', + status_completed: '已完成', + status_failed: '失敗', + status_cancelled: '已取消', + status_queued: '排隊中', + trace_run_completed: '執行完成', + trace_run_failed: '執行失敗', + trace_tool_call_started: '開始執行動作', + trace_tool_call_completed: '動作結果', + }, debugData: { title: '事件資料', form: '常用欄位', @@ -505,7 +535,7 @@ const zhHant = { description: '建立可重用的處理器，並在機器人事件路由中使用', create: '建立處理器', editAgent: '編輯 Agent', - selectFromSidebar: '從側邊欄選擇一個 Agent 或 Pipeline', + selectFromSidebar: '從側邊欄選擇一個處理器', agentType: 'Agent', agentTypeDescription: '透過執行器處理訊息、群成員、好友、回饋等平台事件。', pipelineType: '流程線', diff --git a/web/tests/e2e/crud-smoke.spec.ts b/web/tests/e2e/crud-smoke.spec.ts index 057e4c1f5..26e04093d 100644 --- a/web/tests/e2e/crud-smoke.spec.ts +++ b/web/tests/e2e/crud-smoke.spec.ts @@ -252,7 +252,7 @@ test.describe('frontend CRUD smoke flows', () => { await expect(page).toHaveURL(/\/home\/agents$/); await expect( - page.getByText('Select an Agent or Pipeline from the sidebar'), + page.getByText('Select a processor from the sidebar'), ).toBeVisible(); }); @@ -1430,7 +1430,7 @@ test.describe('empty states', () => { await page.goto('/home/agents'); await expect( - page.getByText('Select an Agent or Pipeline from the sidebar'), + page.getByText('Select a processor from the sidebar'), ).toBeVisible(); }); diff --git a/web/tests/e2e/event-processor.spec.ts b/web/tests/e2e/event-processor.spec.ts new file mode 100644 index 000000000..e3a446df0 --- /dev/null +++ b/web/tests/e2e/event-processor.spec.ts @@ -0,0 +1,115 @@ +import { expect, test } from '@playwright/test'; +import { installLangBotApiMocks } from './fixtures/langbot-api'; + +test('event processor shows isolated logs, paginates and scrolls expanded payloads', async ({ + page, +}) => { + await installLangBotApiMocks(page, { authenticated: true }); + const ref = 'event_processor:qa/welcome/default'; + const processor = { + uuid: 'processor-qa', + kind: 'event_processor', + name: 'Welcome processor', + component_ref: ref, + supported_event_patterns: ['group.member_joined'], + config: { runner: { id: ref }, runner_config: { [ref]: {} } }, + }; + const run = { + run_id: 'run-one', + status: 'completed', + status_reason: 'stop', + created_at: 1788000000, + metadata: { + event_type: 'group.member_joined', + input_event: { member: { id: 'one' } }, + delivery: { + reply_target: { target_type: 'group', target_id: 'test-group' }, + }, + }, + }; + const cursors: string[] = []; + await page.route('**/api/v1/agents**', async (route) => { + const url = new URL(route.request().url()); + let data: unknown; + if (url.pathname.endsWith('/_/metadata')) { + data = { + kinds: [], + event_processors: [ + { + id: ref, + label: { en_US: 'Welcome' }, + supported_event_patterns: ['group.member_joined'], + config_schema: [], + plugin_author: 'qa', + plugin_name: 'welcome', + }, + ], + }; + } else if (url.pathname.endsWith('/runs/run-one/events')) { + cursors.push(url.searchParams.get('after_sequence') ?? ''); + data = { + run, + items: url.searchParams.has('after_sequence') + ? [ + { + sequence: 101, + type: 'processor.log', + data: { level: 'info', text: 'Final log after pagination' }, + }, + ] + : [ + { + sequence: 1, + type: 'processor.log', + data: { level: 'info', text: 'Member received' }, + }, + { + sequence: 100, + type: 'tool.call.completed', + data: { + result: Array.from( + { length: 100 }, + (_, i) => `Payload line ${i}`, + ), + }, + }, + ], + has_more: !url.searchParams.has('after_sequence'), + next_cursor: url.searchParams.has('after_sequence') ? null : 100, + }; + } else if (url.pathname.endsWith('/runs')) { + data = { items: [run], has_more: false, next_cursor: null, total: 1 }; + } else if (url.pathname.endsWith('/processor-qa')) { + data = { agent: processor }; + } else { + data = { agents: [processor] }; + } + await route.fulfill({ json: { code: 0, data } }); + }); + await page.goto('/home/agents?id=processor-qa'); + await expect( + page.getByRole('heading', { name: 'Welcome processor' }), + ).toBeVisible(); + await expect( + page.getByRole('heading', { name: 'Logs and message flow' }), + ).toBeVisible(); + await expect(page.getByText('stop', { exact: true })).toHaveCount(0); + await page + .getByRole('button') + .filter({ hasText: 'group.member_joined' }) + .click(); + await expect( + page.getByText('Member received', { exact: true }), + ).toBeVisible(); + await expect( + page.getByText('Payload line 99', { exact: false }), + ).toBeHidden(); + await page.locator('summary').filter({ hasText: 'Action result' }).click(); + await page.getByRole('button', { name: 'Load more', exact: true }).click(); + await page.getByText('Final log after pagination').scrollIntoViewIfNeeded(); + await expect(page.getByText('Final log after pagination')).toBeInViewport(); + expect(cursors).toEqual(['', '100']); + await expect( + page.getByRole('button', { name: 'Load more', exact: true }), + ).toHaveCount(0); +});