feat(agent-runner): add plugin runner host integration

src/langbot/pkg/pipeline/controller.py

@@ -21,11 +21,45 @@ class Controller:
         self.ap = ap
         self.semaphore = asyncio.Semaphore(self.ap.instance_config.data['concurrency']['pipeline'])
 
+    async def _try_claim_steering_before_session_slot(
+        self,
+        query: pipeline_query.Query,
+    ) -> bool:
+        """Claim steering while the normal per-session slot is still busy.
+
+        Follow-up input must be claimed before it waits behind the session
+        semaphore; otherwise the active run can finish before the query reaches
+        ChatMessageHandler.try_claim_steering_from_query.
+        """
+        try:
+            pipeline_uuid = query.pipeline_uuid
+            if not pipeline_uuid:
+                return False
+
+            pipeline = await self.ap.pipeline_mgr.get_pipeline_by_uuid(pipeline_uuid)
+            if not pipeline:
+                return False
+
+            session = await self.ap.sess_mgr.get_session(query)
+            query.session = session
+            query.pipeline_config = pipeline.pipeline_entity.config
+            query.variables['_pipeline_bound_plugins'] = pipeline.bound_plugins
+            query.variables['_pipeline_bound_mcp_servers'] = pipeline.bound_mcp_servers
+
+            return await self.ap.agent_run_orchestrator.try_claim_steering_from_query(query)
+        except Exception as exc:
+            self.ap.logger.warning(
+                f'Failed to claim query {query.query_id} as steering input: {exc}',
+                exc_info=True,
+            )
+            return False
+
     async def consumer(self):
         """事件处理循环"""
         try:
             while True:
                 selected_query: pipeline_query.Query = None
+                claimed_steering_query: pipeline_query.Query = None
 
                 # 取请求
                 async with self.ap.query_pool:
@@ -36,6 +70,13 @@ class Controller:
                         # Debug logging removed from tight loop to prevent excessive log generation
                         # that can cause memory overflow in high-traffic scenarios
 
+                        if session._semaphore.locked():
+                            if await self._try_claim_steering_before_session_slot(query):
+                                claimed_steering_query = query
+                                self.ap.logger.debug(f'Claimed query {query.query_id} as steering before session slot')
+                                break
+                            continue
+
                         if not session._semaphore.locked():
                             selected_query = query
                             await session._semaphore.acquire()
@@ -44,7 +85,12 @@ class Controller:
 
                             break
 
-                    if selected_query:  # 找到了
+                    if claimed_steering_query:
+                        queries.remove(claimed_steering_query)
+                        self.ap.query_pool.cached_queries.pop(claimed_steering_query.query_id, None)
+                        self.ap.query_pool.condition.notify_all()
+                        continue
+                    elif selected_query:  # 找到了
                         queries.remove(selected_query)
                     else:  # 没找到 说明：没有请求 或者 所有query对应的session都已达到并发上限
                         await self.ap.query_pool.condition.wait()

src/langbot/pkg/pipeline/msgtrun/msgtrun.py

@@ -1,35 +0,0 @@
-from __future__ import annotations
-
-from .. import stage, entities
-from . import truncator
-from ...utils import importutil
-import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
-from . import truncators
-
-importutil.import_modules_in_pkg(truncators)
-
-
-@stage.stage_class('ConversationMessageTruncator')
-class ConversationMessageTruncator(stage.PipelineStage):
-    """Conversation message truncator
-
-    Used to truncate the conversation message chain to adapt to the LLM message length limit.
-    """
-
-    trun: truncator.Truncator
-
-    async def initialize(self, pipeline_config: dict):
-        use_method = 'round'
-
-        for trun in truncator.preregistered_truncators:
-            if trun.name == use_method:
-                self.trun = trun(self.ap)
-                break
-        else:
-            raise ValueError(f'Unknown truncator: {use_method}')
-
-    async def process(self, query: pipeline_query.Query, stage_inst_name: str) -> entities.StageProcessResult:
-        """处理"""
-        query = await self.trun.truncate(query)
-
-        return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)

src/langbot/pkg/pipeline/msgtrun/truncator.py

@@ -1,56 +0,0 @@
-from __future__ import annotations
-
-import typing
-import abc
-
-from ...core import app
-import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
-
-preregistered_truncators: list[typing.Type[Truncator]] = []
-
-
-def truncator_class(
-    name: str,
-) -> typing.Callable[[typing.Type[Truncator]], typing.Type[Truncator]]:
-    """截断器类装饰器
-
-    Args:
-        name (str): 截断器名称
-
-    Returns:
-        typing.Callable[[typing.Type[Truncator]], typing.Type[Truncator]]: 装饰器
-    """
-
-    def decorator(cls: typing.Type[Truncator]) -> typing.Type[Truncator]:
-        assert issubclass(cls, Truncator)
-
-        cls.name = name
-
-        preregistered_truncators.append(cls)
-
-        return cls
-
-    return decorator
-
-
-class Truncator(abc.ABC):
-    """消息截断器基类"""
-
-    name: str
-
-    ap: app.Application
-
-    def __init__(self, ap: app.Application):
-        self.ap = ap
-
-    async def initialize(self):
-        pass
-
-    @abc.abstractmethod
-    async def truncate(self, query: pipeline_query.Query) -> pipeline_query.Query:
-        """截断
-
-        一般只需要操作query.messages，也可以扩展操作query.prompt, query.user_message。
-        请勿操作其他字段。
-        """
-        pass

src/langbot/pkg/pipeline/msgtrun/truncators/round.py

@@ -1,30 +0,0 @@
-from __future__ import annotations
-
-from .. import truncator
-import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
-
-
-@truncator.truncator_class('round')
-class RoundTruncator(truncator.Truncator):
-    """Truncate the conversation message chain to adapt to the LLM message length limit."""
-
-    async def truncate(self, query: pipeline_query.Query) -> pipeline_query.Query:
-        """截断"""
-        max_round = query.pipeline_config['ai']['local-agent']['max-round']
-
-        temp_messages = []
-
-        current_round = 0
-
-        # Traverse from back to front
-        for msg in query.messages[::-1]:
-            if current_round < max_round:
-                temp_messages.append(msg)
-                if msg.role == 'user':
-                    current_round += 1
-            else:
-                break
-
-        query.messages = temp_messages[::-1]
-
-        return query

src/langbot/pkg/pipeline/pipelinemgr.py

@@ -28,7 +28,6 @@ from . import (
     wrapper,
     preproc,
     ratelimit,
-    msgtrun,
 )
 
 importutil.import_modules_in_pkgs(
@@ -42,7 +41,6 @@ importutil.import_modules_in_pkgs(
         wrapper,
         preproc,
         ratelimit,
-        msgtrun,
     ]
 )
 
@@ -289,8 +287,10 @@ class RuntimePipeline:
 
         # Get runner name from pipeline config
         runner_name = None
-        if query.pipeline_config and 'ai' in query.pipeline_config and 'runner' in query.pipeline_config['ai']:
-            runner_name = query.pipeline_config['ai']['runner'].get('runner')
+        if query.pipeline_config:
+            from ..agent.runner.config_migration import ConfigMigration
+
+            runner_name = ConfigMigration.resolve_runner_id(query.pipeline_config)
 
         # Record query start and store message_id
         message_id = ''
@@ -449,6 +449,9 @@ class PipelineManager:
         # initialize stage containers according to pipeline_entity.stages
         stage_containers: list[StageInstContainer] = []
         for stage_name in pipeline_entity.stages:
+            if stage_name not in self.stage_dict:
+                self.ap.logger.warning(f'Pipeline stage {stage_name} is not registered; skipping')
+                continue
             stage_containers.append(StageInstContainer(inst_name=stage_name, inst=self.stage_dict[stage_name](self.ap)))
 
         for stage_container in stage_containers:

src/langbot/pkg/pipeline/preproc/preproc.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import datetime
+import typing
 
 from .. import stage, entities
 from langbot_plugin.api.entities.builtin.provider import message as provider_message
@@ -9,6 +10,15 @@ import langbot_plugin.api.entities.builtin.platform.message as platform_message
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
 import langbot_plugin.api.entities.builtin.platform.events as platform_events
 
+from ...agent.runner.descriptor import AgentRunnerDescriptor
+from ...agent.runner.config_migration import ConfigMigration
+from ...agent.runner import config_schema
+
+
+DEFAULT_PROMPT_CONFIG = [
+    {'role': 'system', 'content': 'You are a helpful assistant.'},
+]
+
 
 @stage.stage_class('PreProcessor')
 class PreProcessor(stage.PipelineStage):
@@ -25,20 +35,131 @@ class PreProcessor(stage.PipelineStage):
         - use_funcs
     """
 
-    @staticmethod
-    def _filter_selected_tools(
-        tools: list,
-        local_agent_config: dict,
-    ) -> list:
-        if local_agent_config.get('enable-all-tools', True) is not False:
-            return tools
+    async def _get_runner_descriptor(
+        self,
+        runner_id: str | None,
+        bound_plugins: list[str] | None,
+    ) -> AgentRunnerDescriptor | None:
+        if not runner_id:
+            return None
 
-        selected_tools = local_agent_config.get('tools', [])
-        if not isinstance(selected_tools, list):
-            return []
+        registry = getattr(self.ap, 'agent_runner_registry', None)
+        if registry is None:
+            return None
 
-        selected_tool_names = {tool for tool in selected_tools if isinstance(tool, str)}
-        return [tool for tool in tools if tool.name in selected_tool_names]
+        try:
+            return await registry.get(runner_id, bound_plugins)
+        except Exception as e:
+            self.ap.logger.debug(f'Unable to load AgentRunner descriptor for {runner_id}: {e}')
+            return None
+
+    async def _resolve_llm_model(
+        self,
+        primary_uuid: str,
+    ) -> typing.Any | None:
+        if primary_uuid in config_schema.NONE_SENTINELS:
+            return None
+        try:
+            return await self.ap.model_mgr.get_model_by_uuid(primary_uuid)
+        except ValueError:
+            self.ap.logger.warning(f'LLM model {primary_uuid} not found or not configured')
+            return None
+
+    async def _resolve_fallback_models(self, fallback_uuids: list[str]) -> list[str]:
+        valid_fallbacks = []
+        for fallback_uuid in fallback_uuids:
+            if fallback_uuid in config_schema.NONE_SENTINELS:
+                continue
+            try:
+                await self.ap.model_mgr.get_model_by_uuid(fallback_uuid)
+                valid_fallbacks.append(fallback_uuid)
+            except ValueError:
+                self.ap.logger.warning(f'Fallback model {fallback_uuid} not found, skipping')
+        return valid_fallbacks
+
+    def _runner_accepts_multimodal_input(self, descriptor: AgentRunnerDescriptor | None) -> bool:
+        if descriptor is None:
+            return True
+        return descriptor.capabilities.multimodal_input
+
+    def _model_supports_vision(self, llm_model: typing.Any | None) -> bool:
+        if not llm_model:
+            return False
+        abilities = getattr(getattr(llm_model, 'model_entity', None), 'abilities', [])
+        return 'vision' in (abilities or [])
+
+    def _should_keep_image_inputs(
+        self,
+        descriptor: AgentRunnerDescriptor | None,
+        uses_host_models: bool,
+        llm_model: typing.Any | None,
+    ) -> bool:
+        if not self._runner_accepts_multimodal_input(descriptor):
+            return False
+        if uses_host_models:
+            return self._model_supports_vision(llm_model)
+        return True
+
+    def _strip_images_from_history(self, query: pipeline_query.Query) -> None:
+        for msg in query.messages:
+            if isinstance(msg.content, list):
+                msg.content = [elem for elem in msg.content if elem.type != 'image_url']
+
+    def _has_declared_db_engine(self) -> bool:
+        persistence_mgr = getattr(self.ap, 'persistence_mgr', None)
+        if persistence_mgr is None:
+            return False
+        if 'get_db_engine' in getattr(persistence_mgr, '__dict__', {}):
+            return True
+        return hasattr(type(persistence_mgr), 'get_db_engine')
+
+    async def _load_agent_runner_history_messages(
+        self,
+        runner_id: str | None,
+        conversation_uuid: str | None,
+        bot_id: str | None = None,
+        workspace_id: str | None = None,
+        thread_id: str | None = None,
+    ) -> list[provider_message.Message] | None:
+        if not runner_id or not conversation_uuid or not self._has_declared_db_engine():
+            return None
+
+        try:
+            from ...agent.runner.transcript_store import TranscriptStore
+
+            store = TranscriptStore(self.ap.persistence_mgr.get_db_engine())
+            messages = await store.get_legacy_provider_messages(
+                str(conversation_uuid),
+                bot_id=bot_id,
+                workspace_id=workspace_id,
+                thread_id=thread_id,
+                strict_thread=True,
+            )
+        except Exception as e:
+            self.ap.logger.warning(
+                f'Unable to load Transcript history view for conversation {conversation_uuid}: {e}'
+            )
+            return None
+
+        return messages or None
+
+    async def _resolve_history_messages(
+        self,
+        runner_id: str | None,
+        conversation: typing.Any,
+        bot_id: str | None = None,
+        workspace_id: str | None = None,
+    ) -> list[provider_message.Message]:
+        transcript_messages = await self._load_agent_runner_history_messages(
+            runner_id,
+            getattr(conversation, 'uuid', None),
+            bot_id=bot_id,
+            workspace_id=workspace_id,
+            thread_id=getattr(conversation, 'thread_id', None),
+        )
+        if transcript_messages is not None:
+            return transcript_messages
+        return conversation.messages.copy()
 
     async def process(
         self,
@@ -46,50 +167,39 @@ class PreProcessor(stage.PipelineStage):
         stage_inst_name: str,
     ) -> entities.StageProcessResult:
         """Process"""
-        selected_runner = query.pipeline_config['ai']['runner']['runner']
-        local_agent_config = query.pipeline_config.get('ai', {}).get('local-agent', {})
-        include_skill_authoring = (
-            selected_runner == 'local-agent' and getattr(self.ap, 'skill_service', None) is not None
-        )
+        # Resolve runner ID from the current ai.runner.id shape.
+        runner_id = ConfigMigration.resolve_runner_id(query.pipeline_config)
+
+        # Get runner config from ai.runner_config[runner_id].
+        runner_config = ConfigMigration.resolve_runner_config(query.pipeline_config, runner_id) if runner_id else {}
+        query.variables = query.variables or {}
+        bound_plugins = query.variables.get('_pipeline_bound_plugins', None)
+        bound_mcp_servers = query.variables.get('_pipeline_bound_mcp_servers', None)
+        include_mcp_resource_tools = query.variables.get('_pipeline_mcp_resource_agent_read_enabled', True)
+        descriptor = await self._get_runner_descriptor(runner_id, bound_plugins)
 
         session = await self.ap.sess_mgr.get_session(query)
 
-        # When not local-agent, llm_model is None
+        uses_host_models = config_schema.uses_host_models(descriptor)
+        uses_host_tools = config_schema.uses_host_tools(descriptor)
+        include_skill_authoring = (
+            config_schema.supports_skill_authoring(descriptor)
+            and getattr(self.ap, 'skill_service', None) is not None
+        )
         llm_model = None
-        if selected_runner == 'local-agent':
-            # Read model config — new format is { primary: str, fallbacks: [str] },
-            # but handle legacy plain string for backward compatibility
-            model_config = local_agent_config.get('model', {})
-            if isinstance(model_config, str):
-                # Legacy format: plain UUID string
-                primary_uuid = model_config
-                fallback_uuids = []
-            else:
-                primary_uuid = model_config.get('primary', '')
-                fallback_uuids = model_config.get('fallbacks', [])
+        if uses_host_models:
+            primary_uuid, fallback_uuids = config_schema.extract_model_selection(descriptor, runner_config)
+            llm_model = await self._resolve_llm_model(primary_uuid)
+            valid_fallbacks = await self._resolve_fallback_models(fallback_uuids)
+            if valid_fallbacks:
+                query.variables['_fallback_model_uuids'] = valid_fallbacks
 
-            if primary_uuid:
-                try:
-                    llm_model = await self.ap.model_mgr.get_model_by_uuid(primary_uuid)
-                except ValueError:
-                    self.ap.logger.warning(f'LLM model {primary_uuid} not found or not configured')
-
-            # Resolve fallback model UUIDs
-            if fallback_uuids:
-                valid_fallbacks = []
-                for fb_uuid in fallback_uuids:
-                    try:
-                        await self.ap.model_mgr.get_model_by_uuid(fb_uuid)
-                        valid_fallbacks.append(fb_uuid)
-                    except ValueError:
-                        self.ap.logger.warning(f'Fallback model {fb_uuid} not found, skipping')
-                if valid_fallbacks:
-                    query.variables['_fallback_model_uuids'] = valid_fallbacks
+        prompt_config = config_schema.extract_prompt_config(descriptor, runner_config, DEFAULT_PROMPT_CONFIG)
 
         conversation = await self.ap.sess_mgr.get_conversation(
             query,
             session,
-            query.pipeline_config['ai']['local-agent']['prompt'],
+            prompt_config,
             query.pipeline_uuid,
             query.bot_uuid,
         )
@@ -98,7 +208,7 @@ class PreProcessor(stage.PipelineStage):
         # been idle for longer than the configured conversation expire time.
         # The idle window is measured from the last preprocess/update time, not
         # from the conversation creation time.
-        conversation_expire_time = query.pipeline_config.get('ai', {}).get('runner', {}).get('expire-time', None)
+        conversation_expire_time = ConfigMigration.get_expire_time(query.pipeline_config)
         now = datetime.datetime.now()
         if conversation_expire_time is not None and conversation_expire_time > 0:
             last_update_time = getattr(conversation, 'update_time', None) or getattr(conversation, 'create_time', None)
@@ -115,28 +225,27 @@ class PreProcessor(stage.PipelineStage):
         # time instead of the first message/creation time.
         conversation.update_time = now
 
-        # 设置query
+        # Attach resolved session state to the query.
         query.session = session
         query.prompt = conversation.prompt.copy()
-        query.messages = conversation.messages.copy()
+        query.messages = await self._resolve_history_messages(
+            runner_id,
+            conversation,
+            bot_id=query.bot_uuid,
+        )
 
-        if selected_runner == 'local-agent':
+        if uses_host_models:
             query.use_funcs = []
             if llm_model:
                 query.use_llm_model_uuid = llm_model.model_entity.uuid
 
-                if 'func_call' in (llm_model.model_entity.abilities or []):
-                    # Get bound plugins and MCP servers for filtering tools
-                    bound_plugins = query.variables.get('_pipeline_bound_plugins', None)
-                    bound_mcp_servers = query.variables.get('_pipeline_bound_mcp_servers', None)
-                    include_mcp_resource_tools = query.variables.get('_pipeline_mcp_resource_agent_read_enabled', True)
-                    all_tools = await self.ap.tool_mgr.get_all_tools(
+                if uses_host_tools and 'func_call' in (llm_model.model_entity.abilities or []):
+                    query.use_funcs = await self.ap.tool_mgr.get_all_tools(
                         bound_plugins,
                         bound_mcp_servers,
                         include_skill_authoring=include_skill_authoring,
                         include_mcp_resource_tools=include_mcp_resource_tools,
                     )
-                    query.use_funcs = self._filter_selected_tools(all_tools, local_agent_config)
 
                     self.ap.logger.debug(f'Bound plugins: {bound_plugins}')
                     self.ap.logger.debug(f'Bound MCP servers: {bound_mcp_servers}')
@@ -144,17 +253,24 @@ class PreProcessor(stage.PipelineStage):
 
             # If primary model doesn't support func_call but fallback models exist,
             # load tools anyway since fallback models may support them
-            if not query.use_funcs and query.variables.get('_fallback_model_uuids'):
-                bound_plugins = query.variables.get('_pipeline_bound_plugins', None)
-                bound_mcp_servers = query.variables.get('_pipeline_bound_mcp_servers', None)
-                include_mcp_resource_tools = query.variables.get('_pipeline_mcp_resource_agent_read_enabled', True)
-                all_tools = await self.ap.tool_mgr.get_all_tools(
+            if uses_host_tools and not query.use_funcs and query.variables.get('_fallback_model_uuids'):
+                query.use_funcs = await self.ap.tool_mgr.get_all_tools(
                     bound_plugins,
                     bound_mcp_servers,
                     include_skill_authoring=include_skill_authoring,
                     include_mcp_resource_tools=include_mcp_resource_tools,
                 )
-                query.use_funcs = self._filter_selected_tools(all_tools, local_agent_config)
+        elif uses_host_tools:
+            query.use_funcs = await self.ap.tool_mgr.get_all_tools(
+                bound_plugins,
+                bound_mcp_servers,
+                include_skill_authoring=include_skill_authoring,
+                include_mcp_resource_tools=include_mcp_resource_tools,
+            )
+
+            self.ap.logger.debug(f'Bound plugins: {bound_plugins}')
+            self.ap.logger.debug(f'Bound MCP servers: {bound_mcp_servers}')
+            self.ap.logger.debug(f'Use funcs: {query.use_funcs}')
 
         sender_name = ''
 
@@ -179,40 +295,28 @@ class PreProcessor(stage.PipelineStage):
         }
         query.variables.update(variables)
 
-        # Check if this model supports vision, if not, remove all images
-        # TODO this checking should be performed in runner, and in this stage, the image should be reserved
-        if selected_runner == 'local-agent' and llm_model and 'vision' not in (llm_model.model_entity.abilities or []):
-            for msg in query.messages:
-                if isinstance(msg.content, list):
-                    for me in msg.content:
-                        if me.type == 'image_url':
-                            msg.content.remove(me)
+        keep_image_inputs = self._should_keep_image_inputs(descriptor, uses_host_models, llm_model)
+        if not keep_image_inputs:
+            self._strip_images_from_history(query)
 
         content_list: list[provider_message.ContentElement] = []
 
         plain_text = ''
-        quote_msg = query.pipeline_config['trigger'].get('misc', '').get('combine-quote-message')
-        local_agent_without_vision = (
-            selected_runner == 'local-agent'
-            and llm_model
-            and not llm_model.model_entity.abilities.__contains__('vision')
-        )
+        quote_msg = query.pipeline_config['trigger'].get('misc', {}).get('combine-quote-message', False)
 
         for me in query.message_chain:
             if isinstance(me, platform_message.Plain):
                 content_list.append(provider_message.ContentElement.from_text(me.text))
                 plain_text += me.text
             elif isinstance(me, platform_message.Image):
-                if local_agent_without_vision:
-                    content_list.append(provider_message.ContentElement.from_text('[Image]'))
-                    plain_text += '[Image]'
-                elif selected_runner != 'local-agent' or (
-                    llm_model and 'vision' in (llm_model.model_entity.abilities or [])
-                ):
+                if keep_image_inputs:
                     if me.base64 is not None:
                         content_list.append(provider_message.ContentElement.from_image_base64(me.base64))
+                else:
+                    content_list.append(provider_message.ContentElement.from_text('[Image]'))
+                    plain_text += '[Image]'
             elif isinstance(me, platform_message.Voice):
-                # 转成文件链接，让下游 runner 上传到目标模型
+                # Convert voice input into file content for downstream model upload.
                 if me.base64:
                     content_list.append(provider_message.ContentElement.from_file_base64(me.base64, 'voice.silk'))
                 elif me.url:
@@ -227,14 +331,12 @@ class PreProcessor(stage.PipelineStage):
                     if isinstance(msg, platform_message.Plain):
                         content_list.append(provider_message.ContentElement.from_text(msg.text))
                     elif isinstance(msg, platform_message.Image):
-                        if local_agent_without_vision:
-                            content_list.append(provider_message.ContentElement.from_text('[Image]'))
-                            plain_text += '[Image]'
-                        elif selected_runner != 'local-agent' or (
-                            llm_model and 'vision' in (llm_model.model_entity.abilities or [])
-                        ):
+                        if keep_image_inputs:
                             if msg.base64 is not None:
                                 content_list.append(provider_message.ContentElement.from_image_base64(msg.base64))
+                        else:
+                            content_list.append(provider_message.ContentElement.from_text('[Image]'))
+                            plain_text += '[Image]'
                     elif isinstance(msg, platform_message.File):
                         if msg.base64:
                             content_list.append(provider_message.ContentElement.from_file_base64(msg.base64, msg.name))
@@ -252,16 +354,14 @@ class PreProcessor(stage.PipelineStage):
 
         query.user_message = provider_message.Message(role='user', content=content_list)
 
-        # Extract knowledge base UUIDs into query variables so plugins can modify them
-        # during PromptPreProcessing before the runner performs retrieval.
-        kb_uuids = query.pipeline_config['ai']['local-agent'].get('knowledge-bases', [])
-        if not kb_uuids:
-            old_kb_uuid = query.pipeline_config['ai']['local-agent'].get('knowledge-base', '')
-            if old_kb_uuid and old_kb_uuid != '__none__':
-                kb_uuids = [old_kb_uuid]
-        query.variables['_knowledge_base_uuids'] = list(kb_uuids)
+        # Extract configured KB UUIDs into query variables so PromptPreProcessing
+        # plugins can still adjust the authorized retrieval set before run_agent.
+        query.variables['_knowledge_base_uuids'] = config_schema.extract_knowledge_base_uuids(
+            descriptor,
+            runner_config,
+        )
 
-        # =========== 触发事件 PromptPreProcessing
+        # Emit PromptPreProcessing before the runner receives the query.
 
         event = events.PromptPreProcessing(
             session_name=f'{query.session.launcher_type.value}_{query.session.launcher_id}',
@@ -277,19 +377,7 @@ class PreProcessor(stage.PipelineStage):
         query.prompt.messages = event_ctx.event.default_prompt
         query.messages = event_ctx.event.prompt
 
-        # =========== Skill awareness for the local-agent runner ===========
-        # The actual activation goes through the ``activate`` Tool Call so the
-        # LLM doesn't see full SKILL.md instructions until it commits to a
-        # skill (Claude Code's progressive disclosure). But the LLM still has
-        # to KNOW which skills exist to make that choice, so we:
-        #   1. resolve the pipeline's bound skills and stash them in
-        #      ``query.variables['_pipeline_bound_skills']`` for downstream
-        #      visibility checks (skill loader, native exec workdir);
-        #   2. inject a short ``Available Skills`` index (name + description
-        #      only) into the system prompt. The contributor's original PR
-        #      relied on this injection; without it the LLM never discovers
-        #      the skills are there and just calls native tools instead.
-        if selected_runner == 'local-agent' and self.ap.skill_mgr:
+        if include_skill_authoring and getattr(self.ap, 'skill_mgr', None) is not None:
             pipeline_data = await self.ap.pipeline_service.get_pipeline(query.pipeline_uuid)
             extensions_prefs = (pipeline_data or {}).get('extensions_preferences', {})
             enable_all_skills = extensions_prefs.get('enable_all_skills', True)
@@ -301,43 +389,4 @@ class PreProcessor(stage.PipelineStage):
 
             query.variables['_pipeline_bound_skills'] = bound_skills
 
-            skill_addition = self.ap.skill_mgr.build_skill_aware_prompt_addition(
-                bound_skills=bound_skills,
-            )
-            if skill_addition:
-                # Append to the first system message; create one if the
-                # prompt has none. Handles both plain-string and
-                # content-element (list) message bodies.
-                if query.prompt.messages and query.prompt.messages[0].role == 'system':
-                    head = query.prompt.messages[0]
-                    if isinstance(head.content, str):
-                        head.content = head.content + skill_addition
-                    elif isinstance(head.content, list):
-                        appended = False
-                        for ce in head.content:
-                            if getattr(ce, 'type', None) == 'text':
-                                ce.text = (ce.text or '') + skill_addition
-                                appended = True
-                                break
-                        if not appended:
-                            head.content.append(provider_message.ContentElement(type='text', text=skill_addition))
-                else:
-                    query.prompt.messages.insert(
-                        0,
-                        provider_message.Message(role='system', content=skill_addition.strip()),
-                    )
-                self.ap.logger.debug(
-                    f'Skill index injected into system prompt: '
-                    f'pipeline={query.pipeline_uuid} '
-                    f'bound_skills={bound_skills or "all"} '
-                    f'loaded_skills={len(self.ap.skill_mgr.skills)}'
-                )
-            else:
-                self.ap.logger.debug(
-                    f'No skills available for prompt injection: '
-                    f'pipeline={query.pipeline_uuid} '
-                    f'loaded_skills={len(self.ap.skill_mgr.skills)} '
-                    f'bound_skills={bound_skills}'
-                )
-
         return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)

src/langbot/pkg/pipeline/process/handlers/chat.py

@@ -10,30 +10,36 @@ from datetime import datetime
 from .. import handler
 from ... import entities
 from ... import plugin_diagnostics
-from ....provider import runner as runner_module
 
 import langbot_plugin.api.entities.events as events
-from ....utils import importutil, constants, runner as runner_utils
+from ....agent.runner.config_migration import ConfigMigration
+from ....agent.runner import config_schema
+from ....utils import constants, runner as runner_utils
 from ....telemetry import features as telemetry_features
-from ....provider import runners
 import langbot_plugin.api.entities.builtin.provider.session as provider_session
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
 import langbot_plugin.api.entities.builtin.provider.message as provider_message
 
 
-importutil.import_modules_in_pkg(runners)
+DEFAULT_PROMPT_CONFIG = [
+    {'role': 'system', 'content': 'You are a helpful assistant.'},
+]
 
 
 class ChatMessageHandler(handler.MessageHandler):
+    """Chat message handler using AgentRunOrchestrator.
+
+    This handler delegates all runner execution to the agent_run_orchestrator,
+    which resolves runner ID, builds context, invokes plugin runtime,
+    and normalizes results.
+    """
+
     async def handle(
         self,
         query: pipeline_query.Query,
     ) -> typing.AsyncGenerator[entities.StageProcessResult, None]:
-        """处理"""
-        # 调API
-        #   生成器
-
-        # 触发插件事件
+        """Handle chat message by delegating to AgentRunOrchestrator."""
+        # Trigger plugin event
         event_class = (
             events.PersonNormalMessageReceived
             if query.launcher_type == provider_session.LauncherTypes.PERSON
@@ -54,7 +60,7 @@ class ChatMessageHandler(handler.MessageHandler):
         bound_plugins = query.variables.get('_pipeline_bound_plugins', None)
         event_ctx = await self.ap.plugin_connector.emit_event(event, bound_plugins)
 
-        is_create_card = False  # 判断下是否需要创建流式卡片
+        is_create_card = False  # Track if streaming card was created
 
         if event_ctx.is_prevented_default():
             if event_ctx.event.reply_message_chain is not None:
@@ -87,40 +93,51 @@ class ChatMessageHandler(handler.MessageHandler):
 
             text_length = 0
             try:
-                is_stream = await query.adapter.is_stream_output_supported()
-            except AttributeError:
-                is_stream = False
-
-            try:
-                for r in runner_module.preregistered_runners:
-                    if r.name == query.pipeline_config['ai']['runner']['runner']:
-                        runner = r(self.ap, query.pipeline_config)
-                        break
-                else:
-                    raise ValueError(f'Request Runner not found: {query.pipeline_config["ai"]["runner"]["runner"]}')
                 # Mark start time for telemetry
                 start_ts = time.time()
 
-                if is_stream:
-                    resp_message_id = uuid.uuid4()
-                    chunk_count = 0  # Track streaming chunks to reduce excessive logging
+                try_claim_steering = getattr(
+                    self.ap.agent_run_orchestrator,
+                    'try_claim_steering_from_query',
+                    None,
+                )
+                if try_claim_steering and await try_claim_steering(query):
+                    yield entities.StageProcessResult(result_type=entities.ResultType.INTERRUPT, new_query=query)
+                    return
 
-                    async for result in runner.run(query):
-                        result.resp_message_id = str(resp_message_id)
+                try:
+                    is_stream = await query.adapter.is_stream_output_supported()
+                except AttributeError:
+                    is_stream = False
+
+                # Create a single resp_message_id for the entire streaming response
+                resp_message_id = uuid.uuid4()
+                chunk_count = 0
+
+                # Use AgentRunOrchestrator to run the agent
+                # This replaces direct runner lookup and PluginAgentRunnerWrapper
+                async for result in self.ap.agent_run_orchestrator.run_from_query(query):
+                    result.resp_message_id = str(resp_message_id)
+
+                    # For streaming mode, pop previous response before adding new chunk
+                    # This allows incremental card updates
+                    if is_stream:
                         if query.resp_messages:
                             query.resp_messages.pop()
                         if query.resp_message_chain:
                             query.resp_message_chain.pop()
-                        # 此时连接外部 AI 服务正常,创建卡片
-                        if not is_create_card:  # 只有不是第一次才创建卡片
+
+                        # Create streaming card on first result (connection established)
+                        if not is_create_card:
                             await query.adapter.create_message_card(str(resp_message_id), query.message_event)
                             is_create_card = True
-                        query.resp_messages.append(result)
 
+                    query.resp_messages.append(result)
+
+                    if is_stream:
                         chunk_count += 1
-                        # Only log every 10th chunk to reduce excessive logging during streaming
-                        # This prevents memory overflow from thousands of log entries per conversation
-                        # First chunk uses INFO level to confirm connection establishment
+                        # Only log every 10th chunk to reduce excessive logging during streaming.
+                        # First chunk uses INFO level to confirm connection establishment.
                         if chunk_count == 1:
                             summary = self.format_result_log(result)
                             if summary is not None:
@@ -131,46 +148,59 @@ class ChatMessageHandler(handler.MessageHandler):
                             self.ap.logger.debug(
                                 f'Conversation({query.query_id}) Streaming chunk {chunk_count}: {self.cut_str(result.readable_str())}'
                             )
-
-                        if result.content is not None:
-                            text_length += len(result.content)
-
-                        yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
-
-                    # Log final summary after streaming completes
-                    self.ap.logger.info(
-                        f'Conversation({query.query_id}) Streaming completed: {chunk_count} chunks, {text_length} chars'
-                    )
-
-                else:
-                    async for result in runner.run(query):
-                        query.resp_messages.append(result)
-
+                    else:
                         summary = self.format_result_log(result)
                         if summary is not None:
                             self.ap.logger.info(f'Conversation({query.query_id}) Response: {summary}')
 
-                        if result.content is not None:
-                            text_length += len(result.content)
+                    if result.content is not None:
+                        text_length += len(result.content)
 
-                        yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
+                    yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
 
-                query.session.using_conversation.messages.append(query.user_message)
+                # Log final summary after streaming completes
+                if is_stream:
+                    self.ap.logger.info(
+                        f'Conversation({query.query_id}) Streaming completed: {chunk_count} chunks, {text_length} chars'
+                    )
+
+                # Keep a conversation object available for downstream legacy
+                # readers, but do not mirror AgentRunner history into
+                # conversation.messages. TranscriptStore is the canonical
+                # history source for this path.
+                await self._ensure_conversation_for_history(query)
 
-                query.session.using_conversation.messages.extend(query.resp_messages)
             except Exception as e:
+                # Import orchestrator errors for specific handling
+                from ....agent.runner.errors import (
+                    RunnerNotFoundError,
+                    RunnerNotAuthorizedError,
+                    RunnerExecutionError,
+                )
+
                 error_info = f'{traceback.format_exc()}'
                 self.ap.logger.error(f'Conversation({query.query_id}) Request Failed: {error_info}')
-                traceback.print_exc()
 
-                exception_handling = query.pipeline_config['output']['misc'].get('exception-handling', 'show-hint')
+                # Handle specific runner errors with appropriate messages
+                if isinstance(e, RunnerNotFoundError):
+                    user_notice = f'Agent runner not found: {e.runner_id}'
+                elif isinstance(e, RunnerNotAuthorizedError):
+                    user_notice = 'Agent runner not authorized for this pipeline'
+                elif isinstance(e, RunnerExecutionError):
+                    if e.retryable:
+                        user_notice = 'Agent runner temporarily unavailable. Please try again.'
+                    else:
+                        user_notice = 'Agent runner execution failed.'
+                else:
+                    # Use existing exception handling
+                    exception_handling = query.pipeline_config['output']['misc'].get('exception-handling', 'show-hint')
 
-                if exception_handling == 'show-error':
-                    user_notice = f'{e}'
-                elif exception_handling == 'show-hint':
-                    user_notice = query.pipeline_config['output']['misc'].get('failure-hint', 'Request failed.')
-                else:  # hide
-                    user_notice = None
+                    if exception_handling == 'show-error':
+                        user_notice = f'{e}'
+                    elif exception_handling == 'show-hint':
+                        user_notice = query.pipeline_config['output']['misc'].get('failure-hint', 'Request failed.')
+                    else:  # hide
+                        user_notice = None
 
                 yield entities.StageProcessResult(
                     result_type=entities.ResultType.INTERRUPT,
@@ -180,7 +210,7 @@ class ChatMessageHandler(handler.MessageHandler):
                     debug_notice=traceback.format_exc(),
                 )
             finally:
-                # Telemetry reporting: collect minimal per-query execution info and send asynchronously
+                # Telemetry reporting
                 try:
                     end_ts = time.time()
                     duration_ms = None
@@ -188,16 +218,14 @@ class ChatMessageHandler(handler.MessageHandler):
                         duration_ms = int((end_ts - start_ts) * 1000)
 
                     adapter_name = query.adapter.__class__.__name__ if hasattr(query, 'adapter') else None
-                    runner_name = (
-                        query.pipeline_config.get('ai', {}).get('runner', {}).get('runner')
-                        if query.pipeline_config
-                        else None
-                    )
 
-                    # Model name if using localagent
+                    # Use orchestrator to resolve runner ID for telemetry
+                    runner_name = self.ap.agent_run_orchestrator.resolve_runner_id_for_telemetry(query)
+
+                    # Model name if available
                     model_name = None
                     try:
-                        if runner_name == 'local-agent' and getattr(query, 'use_llm_model_uuid', None):
+                        if getattr(query, 'use_llm_model_uuid', None):
                             m = await self.ap.model_mgr.get_model_by_uuid(query.use_llm_model_uuid)
                             if m and getattr(m, 'model_entity', None):
                                 model_name = getattr(m.model_entity, 'name', None)
@@ -207,7 +235,7 @@ class ChatMessageHandler(handler.MessageHandler):
                     pipeline_plugins = query.variables.get('_pipeline_bound_plugins', None)
 
                     runner_category = runner_utils.get_runner_category_from_runner(
-                        runner_name, runner, query.pipeline_config
+                        runner_name, None, query.pipeline_config
                     )
 
                     # Feature usage collected during query processing (tool calls,
@@ -231,7 +259,6 @@ class ChatMessageHandler(handler.MessageHandler):
                         'timestamp': datetime.utcnow().isoformat(),
                     }
 
-                    # Send telemetry asynchronously and do not block pipeline via app's telemetry manager
                     await self.ap.telemetry.start_send_task(payload)
 
                     # Trigger survey events on successful non-WebSocket responses
@@ -241,5 +268,70 @@ class ChatMessageHandler(handler.MessageHandler):
                             # Counts toward the bot_response_success_100 milestone event
                             await self.ap.survey.record_bot_response_success()
                 except Exception as ex:
-                    # Ensure telemetry issues do not affect normal flow
                     self.ap.logger.warning(f'Failed to send telemetry: {ex}')
+
+    async def _ensure_conversation_for_history(
+        self,
+        query: pipeline_query.Query,
+    ) -> provider_session.Conversation:
+        session = getattr(query, 'session', None)
+        conversation = getattr(session, 'using_conversation', None)
+        if conversation is not None:
+            return conversation
+
+        if session is None or getattr(self.ap, 'sess_mgr', None) is None:
+            raise RuntimeError('Conversation is not available for history update')
+
+        prompt_config = await self._build_history_prompt_config(query)
+        conversation = await self.ap.sess_mgr.get_conversation(
+            query,
+            session,
+            prompt_config,
+            query.pipeline_uuid,
+            query.bot_uuid,
+        )
+        if conversation is None:
+            raise RuntimeError('Conversation manager did not return a conversation')
+
+        if getattr(session, 'using_conversation', None) is None:
+            session.using_conversation = conversation
+        return conversation
+
+    async def _build_history_prompt_config(
+        self,
+        query: pipeline_query.Query,
+    ) -> list[dict[str, typing.Any]]:
+        prompt_messages = getattr(getattr(query, 'prompt', None), 'messages', None)
+        if prompt_messages:
+            prompt_config = []
+            for message in prompt_messages:
+                if hasattr(message, 'model_dump'):
+                    prompt_config.append(message.model_dump(mode='python'))
+                elif isinstance(message, dict):
+                    prompt_config.append(message)
+            if prompt_config:
+                return prompt_config
+
+        runner_id = ConfigMigration.resolve_runner_id(query.pipeline_config)
+        runner_config = ConfigMigration.resolve_runner_config(query.pipeline_config, runner_id) if runner_id else {}
+        bound_plugins = query.variables.get('_pipeline_bound_plugins', None)
+        descriptor = await self._get_runner_descriptor(runner_id, bound_plugins)
+        return config_schema.extract_prompt_config(descriptor, runner_config, DEFAULT_PROMPT_CONFIG)
+
+    async def _get_runner_descriptor(
+        self,
+        runner_id: str | None,
+        bound_plugins: list[str] | None,
+    ) -> typing.Any | None:
+        if not runner_id:
+            return None
+
+        registry = getattr(self.ap, 'agent_runner_registry', None)
+        if registry is None:
+            return None
+
+        try:
+            return await registry.get(runner_id, bound_plugins)
+        except Exception as e:
+            self.ap.logger.debug(f'Unable to load AgentRunner descriptor for {runner_id}: {e}')
+            return None