feat: support dynamic agent runner defaults

src/langbot/pkg/agent/runner/context_builder.py

@@ -6,6 +6,7 @@ import time
 import typing
 
 from langbot_plugin.api.entities.builtin.pipeline import query as pipeline_query
+from langbot_plugin.api.entities.builtin.platform import message as platform_message
 
 from ...core import app
 from .descriptor import AgentRunnerDescriptor
@@ -117,9 +118,9 @@ class AgentRunContextV1(typing.TypedDict):
     run_id: str
     trigger: AgentTrigger
     conversation: ConversationContext | None
-    event: dict[str, typing.Any] | None  # Reserved for EBA
-    actor: dict[str, typing.Any] | None  # Reserved for EBA
-    subject: dict[str, typing.Any] | None  # Reserved for EBA
+    event: dict[str, typing.Any] | None
+    actor: dict[str, typing.Any] | None
+    subject: dict[str, typing.Any] | None
     messages: list[dict[str, typing.Any]]
     input: AgentInput
     params: dict[str, typing.Any]
@@ -226,7 +227,7 @@ class AgentRunContextBuilder:
             'sdk_protocol_version': descriptor.protocol_version,
             'query_id': query.query_id,
             'trace_id': run_id,  # Use run_id as trace_id for now
-            'deadline_at': None,  # TODO: set from runner config timeout
+            'deadline_at': self._build_deadline(runner_config),
             'metadata': {
                 'bot_name': query.variables.get('_monitoring_bot_name', 'Unknown'),
                 'pipeline_name': query.variables.get('_monitoring_pipeline_name', 'Unknown'),
@@ -238,9 +239,9 @@ class AgentRunContextBuilder:
             'run_id': run_id,
             'trigger': trigger,
             'conversation': conversation,
-            'event': None,  # Reserved for EBA
-            'actor': None,  # Reserved for EBA
-            'subject': None,  # Reserved for EBA
+            'event': self._build_event(query),
+            'actor': self._build_actor(query),
+            'subject': self._build_subject(query),
             'messages': messages,
             'input': input,
             'params': params,
@@ -278,9 +279,200 @@ class AgentRunContextBuilder:
             'text': text,
             'contents': contents,
             'message_chain': message_chain_dict,
-            'attachments': [],  # TODO: extract attachments from message_chain
+            'attachments': self._build_attachments(query, contents),
         }
 
+    def _build_attachments(
+        self,
+        query: pipeline_query.Query,
+        contents: list[dict[str, typing.Any]],
+    ) -> list[dict[str, typing.Any]]:
+        """Extract runner-consumable attachment data from input contents."""
+        attachments: list[dict[str, typing.Any]] = []
+
+        for elem in contents:
+            elem_type = elem.get('type')
+            if elem_type == 'image_url':
+                image_url = elem.get('image_url') or {}
+                attachments.append(
+                    {
+                        'type': 'image',
+                        'source': 'url',
+                        'url': image_url.get('url') if isinstance(image_url, dict) else str(image_url),
+                    }
+                )
+            elif elem_type == 'image_base64':
+                image_base64 = elem.get('image_base64')
+                attachments.append(
+                    {
+                        'type': 'image',
+                        'source': 'base64',
+                        'content': image_base64,
+                        'content_type': self._infer_base64_content_type(image_base64, 'image/jpeg'),
+                        'name': 'image',
+                        'has_content': bool(image_base64),
+                    }
+                )
+            elif elem_type == 'file_url':
+                attachments.append(
+                    {
+                        'type': 'file',
+                        'source': 'url',
+                        'url': elem.get('file_url'),
+                        'name': elem.get('file_name'),
+                    }
+                )
+            elif elem_type == 'file_base64':
+                file_base64 = elem.get('file_base64')
+                attachments.append(
+                    {
+                        'type': 'file',
+                        'source': 'base64',
+                        'name': elem.get('file_name'),
+                        'content': file_base64,
+                        'content_type': self._infer_base64_content_type(file_base64, 'application/octet-stream'),
+                        'has_content': bool(file_base64),
+                    }
+                )
+
+        message_chain = getattr(query, 'message_chain', None)
+        if message_chain:
+            for component in message_chain:
+                if isinstance(component, platform_message.Image):
+                    attachments.append(
+                        {
+                            'type': 'image',
+                            'source': 'message_chain',
+                            'id': component.image_id or None,
+                            'url': component.url or None,
+                            'path': str(component.path) if component.path else None,
+                            'content': component.base64 or None,
+                            'content_type': self._infer_base64_content_type(component.base64, 'image/jpeg'),
+                            'name': 'image',
+                            'has_content': bool(component.base64),
+                        }
+                    )
+                elif isinstance(component, platform_message.File):
+                    attachments.append(
+                        {
+                            'type': 'file',
+                            'source': 'message_chain',
+                            'id': component.id or None,
+                            'name': component.name or None,
+                            'size': component.size or 0,
+                            'url': component.url or None,
+                            'path': component.path or None,
+                            'content': component.base64 or None,
+                            'content_type': self._infer_base64_content_type(component.base64, 'application/octet-stream'),
+                            'has_content': bool(component.base64),
+                        }
+                    )
+                elif isinstance(component, platform_message.Voice):
+                    attachments.append(
+                        {
+                            'type': 'voice',
+                            'source': 'message_chain',
+                            'id': component.voice_id or None,
+                            'url': component.url or None,
+                            'path': component.path or None,
+                            'duration': component.length or 0,
+                            'content': component.base64 or None,
+                            'content_type': self._infer_base64_content_type(component.base64, 'audio/mpeg'),
+                            'name': 'voice',
+                            'has_content': bool(component.base64),
+                        }
+                    )
+
+        return attachments
+
+    def _infer_base64_content_type(self, value: typing.Any, default: str) -> str:
+        """Infer MIME type from a data URL base64 value."""
+        if not isinstance(value, str):
+            return default
+        if value.startswith('data:') and ';base64,' in value:
+            return value[5:value.find(';base64,')] or default
+        return default
+
+    def _build_event(self, query: pipeline_query.Query) -> dict[str, typing.Any]:
+        """Build a minimal event envelope from the platform message event."""
+        message_event = getattr(query, 'message_event', None)
+        event_data: dict[str, typing.Any] = {}
+
+        if message_event and hasattr(message_event, 'model_dump'):
+            try:
+                event_data = message_event.model_dump(mode='json')
+            except TypeError:
+                event_data = message_event.model_dump()
+            except Exception:
+                event_data = {}
+            event_data.pop('source_platform_object', None)
+
+        message_chain = getattr(query, 'message_chain', None)
+        message_id = getattr(message_chain, 'message_id', None)
+        if message_id == -1:
+            message_id = None
+
+        event_time = getattr(message_event, 'time', None) if message_event else None
+        event_timestamp = int(event_time) if isinstance(event_time, (int, float)) else None
+
+        return {
+            'event_type': getattr(message_event, 'type', None) or 'message.received',
+            'event_id': str(message_id or getattr(query, 'query_id', '')),
+            'event_timestamp': event_timestamp,
+            'event_data': event_data,
+        }
+
+    def _build_actor(self, query: pipeline_query.Query) -> dict[str, typing.Any]:
+        """Build actor context for the sender that triggered the run."""
+        message_event = getattr(query, 'message_event', None)
+        sender = getattr(message_event, 'sender', None) if message_event else None
+        actor_id = getattr(sender, 'id', None) or getattr(query, 'sender_id', None)
+        actor_name = sender.get_name() if sender and hasattr(sender, 'get_name') else None
+
+        return {
+            'actor_type': 'user',
+            'actor_id': str(actor_id) if actor_id is not None else None,
+            'actor_name': actor_name,
+        }
+
+    def _build_subject(self, query: pipeline_query.Query) -> dict[str, typing.Any]:
+        """Build subject context for the current message."""
+        message_chain = getattr(query, 'message_chain', None)
+        message_id = getattr(message_chain, 'message_id', None)
+        if message_id == -1:
+            message_id = None
+
+        launcher_type = getattr(query, 'launcher_type', None)
+        launcher_type_value = getattr(launcher_type, 'value', launcher_type)
+
+        return {
+            'subject_type': 'message',
+            'subject_id': str(message_id or getattr(query, 'query_id', '')),
+            'subject_data': {
+                'launcher_type': launcher_type_value,
+                'launcher_id': getattr(query, 'launcher_id', None),
+                'sender_id': str(getattr(query, 'sender_id', '')),
+                'bot_uuid': getattr(query, 'bot_uuid', None),
+                'pipeline_uuid': getattr(query, 'pipeline_uuid', None),
+            },
+        }
+
+    def _build_deadline(self, runner_config: dict[str, typing.Any]) -> int | None:
+        """Build deadline timestamp from runner timeout config if present."""
+        timeout = runner_config.get('timeout')
+        if timeout is None:
+            return None
+
+        try:
+            timeout_seconds = float(timeout)
+        except (TypeError, ValueError):
+            return None
+
+        if timeout_seconds <= 0:
+            return None
+
+        return int(time.time() + timeout_seconds)
+
     def _build_messages(self, query: pipeline_query.Query) -> list[dict[str, typing.Any]]:
         """Build messages list from query."""
         messages: list[dict[str, typing.Any]] = []
@@ -357,4 +549,4 @@ class AgentRunContextBuilder:
             )
         # Pydantic models and other complex types are not directly serializable
         # as params (they may have internal structure not meant for runners)
-        return False
+        return False

src/langbot/pkg/agent/runner/registry.py

@@ -1,4 +1,5 @@
 """Agent runner registry for discovering and caching runner descriptors."""
+
 from __future__ import annotations
 
 import typing
@@ -109,11 +110,14 @@ class AgentRunnerRegistry:
         if not label:
             label = {name: name}  # fallback
 
-        # SDK now provides these directly extracted from spec
-        protocol_version = runner_data.get('protocol_version', '1')
-        config_schema = runner_data.get('config', [])
-        capabilities = runner_data.get('capabilities', {})
-        permissions = runner_data.get('permissions', {})
+        spec = manifest.get('spec', {})
+
+        # SDK now provides these directly extracted from spec. Fall back to
+        # manifest.spec for older runtimes/tests that return the raw manifest.
+        protocol_version = runner_data.get('protocol_version') or spec.get('protocol_version', '1')
+        config_schema = runner_data.get('config') or spec.get('config', [])
+        capabilities = runner_data.get('capabilities') or spec.get('capabilities', {})
+        permissions = runner_data.get('permissions') or spec.get('permissions', {})
 
         # Build descriptor
         runner_id = format_runner_id(
@@ -259,19 +263,23 @@ class AgentRunnerRegistry:
 
         for descriptor in runners:
             # Add runner option
-            options.append({
-                'name': descriptor.id,
-                'label': descriptor.label,
-                'description': descriptor.description,
-            })
-
-            # Add config schema as stage if not empty
-            if descriptor.config_schema:
-                stages.append({
+            options.append(
+                {
                     'name': descriptor.id,
                     'label': descriptor.label,
                     'description': descriptor.description,
-                    'config': descriptor.config_schema,
-                })
+                }
+            )
 
-        return options, stages
+            # Add config schema as stage if not empty
+            if descriptor.config_schema:
+                stages.append(
+                    {
+                        'name': descriptor.id,
+                        'label': descriptor.label,
+                        'description': descriptor.description,
+                        'config': descriptor.config_schema,
+                    }
+                )
+
+        return options, stages

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

@@ -3,10 +3,13 @@ from __future__ import annotations
 import uuid
 import json
 import sqlalchemy
+import typing
 
 from ....core import app
 from ....entity.persistence import pipeline as persistence_pipeline
 
+DEFAULT_RUNNER_ID = 'plugin:langbot/local-agent/default'
+
 
 default_stage_order = [
     'GroupRespondRuleCheckStage',  # 群响应规则检查
@@ -30,6 +33,46 @@ class PipelineService:
     def __init__(self, ap: app.Application) -> None:
         self.ap = ap
 
+    def _get_default_values_from_schema(self, config_schema: list[dict[str, typing.Any]]) -> dict[str, typing.Any]:
+        """Build runner config defaults from a DynamicForm schema."""
+        defaults: dict[str, typing.Any] = {}
+        for item in config_schema:
+            name = item.get('name')
+            if not name:
+                continue
+            if 'default' in item:
+                defaults[name] = item['default']
+        return defaults
+
+    async def get_default_pipeline_config(self) -> dict[str, typing.Any]:
+        """Get the default pipeline config, rendering runner defaults from installed plugins."""
+        from ....utils import paths as path_utils
+
+        template_path = path_utils.get_resource_path('templates/default-pipeline-config.json')
+        with open(template_path, 'r', encoding='utf-8') as f:
+            config = json.load(f)
+
+        try:
+            runners = await self.ap.agent_runner_registry.list_runners(bound_plugins=None)
+        except Exception as e:
+            self.ap.logger.warning(f'Failed to load plugin agent runners for default pipeline config: {e}')
+            return config
+
+        if not runners:
+            return config
+
+        selected_runner = next((runner for runner in runners if runner.id == DEFAULT_RUNNER_ID), runners[0])
+        ai_config = config.setdefault('ai', {})
+        runner_config = ai_config.setdefault('runner', {})
+        runner_config['id'] = selected_runner.id
+        runner_config.setdefault('expire-time', 0)
+
+        ai_config['runner_config'] = {
+            selected_runner.id: self._get_default_values_from_schema(selected_runner.config_schema),
+        }
+
+        return config
+
     async def get_pipeline_metadata(self) -> list[dict]:
         """Get pipeline metadata with dynamically loaded plugin runners from registry"""
         import copy
@@ -50,15 +93,22 @@ class PipelineService:
                 if config_item.get('name') == 'id':
                     # Get plugin agent runners from registry
                     try:
-                        runner_options, runner_stages = await self.ap.agent_runner_registry.get_runner_metadata_for_pipeline()
+                        (
+                            runner_options,
+                            runner_stages,
+                        ) = await self.ap.agent_runner_registry.get_runner_metadata_for_pipeline()
 
                         # Replace options entirely with registry options
                         # Only installed/available runners should be shown
                         config_item['options'] = runner_options
 
-                        # Set default to first available runner if not specified
+                        # Set default to the official local-agent when installed, otherwise first available runner.
                         if runner_options and 'default' not in config_item:
-                            config_item['default'] = runner_options[0]['name']
+                            default_option = next(
+                                (option for option in runner_options if option['name'] == DEFAULT_RUNNER_ID),
+                                runner_options[0],
+                            )
+                            config_item['default'] = default_option['name']
 
                         # Add corresponding stage configuration for each runner
                         for stage_config in runner_stages:
@@ -113,8 +163,6 @@ class PipelineService:
         return self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
 
     async def create_pipeline(self, pipeline_data: dict, default: bool = False) -> str:
-        from ....utils import paths as path_utils
-
         # Check limitation
         limitation = self.ap.instance_config.data.get('system', {}).get('limitation', {})
         max_pipelines = limitation.get('max_pipelines', -1)
@@ -128,9 +176,7 @@ class PipelineService:
         pipeline_data['stages'] = default_stage_order.copy()
         pipeline_data['is_default'] = default
 
-        template_path = path_utils.get_resource_path('templates/default-pipeline-config.json')
-        with open(template_path, 'r', encoding='utf-8') as f:
-            pipeline_data['config'] = json.load(f)
+        pipeline_data['config'] = await self.get_default_pipeline_config()
 
         # Ensure extensions_preferences is set with enable_all_plugins and enable_all_mcp_servers=True by default
         if 'extensions_preferences' not in pipeline_data:

src/langbot/pkg/plugin/connector.py

@@ -187,6 +187,15 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
     async def initialize_plugins(self):
         pass
 
+    async def _refresh_agent_runner_registry(self) -> None:
+        registry = getattr(self.ap, 'agent_runner_registry', None)
+        if registry is None:
+            return
+        try:
+            await registry.refresh()
+        except Exception as e:
+            self.ap.logger.warning(f'Failed to refresh agent runner registry: {e}')
+
     async def ping_plugin_runtime(self):
         if not hasattr(self, 'handler'):
             raise PluginRuntimeNotConnectedError('Plugin runtime is not connected')
@@ -546,6 +555,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                 task_context.metadata.update(metadata)
 
         await self._wait_for_installed_plugin_ready(plugin_author, plugin_name, task_context)
+        await self._refresh_agent_runner_registry()
 
     async def upgrade_plugin(
         self,
@@ -564,6 +574,8 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                 if task_context is not None:
                     task_context.trace(trace)
 
+        await self._refresh_agent_runner_registry()
+
     async def delete_plugin(
         self,
         plugin_author: str,
@@ -588,6 +600,8 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                 task_context.trace('Cleaning up plugin configuration and storage...')
             await self.handler.cleanup_plugin_data(plugin_author, plugin_name)
 
+        await self._refresh_agent_runner_registry()
+
     async def list_plugins(self, component_kinds: list[str] | None = None) -> list[dict[str, Any]]:
         """List plugins, optionally filtered by component kinds.
 

src/langbot/pkg/plugin/handler.py

@@ -41,6 +41,63 @@ def _make_rag_error_response(error: Exception, error_type: str, **extra_context)
     return handler.ActionResponse.error(message=message)
 
 
+def _i18n_to_dict(value: Any) -> dict[str, Any]:
+    """Convert SDK i18n values to plain dictionaries."""
+    if value is None:
+        return {}
+    if isinstance(value, dict):
+        return value
+    if hasattr(value, 'to_dict'):
+        return value.to_dict()
+    if hasattr(value, 'model_dump'):
+        return value.model_dump()
+    return {'en_US': str(value)}
+
+
+def _i18n_to_text(value: Any) -> str:
+    """Return a stable human-readable text from SDK i18n values."""
+    data = _i18n_to_dict(value)
+    for key in ('en_US', 'zh_Hans', 'zh_Hant'):
+        text = data.get(key)
+        if text:
+            return str(text)
+    for text in data.values():
+        if text:
+            return str(text)
+    return ''
+
+
+def _build_tool_detail(tool: Any, requested_tool_name: str | None = None) -> dict[str, Any]:
+    """Normalize LLMTool and plugin ComponentManifest objects for tool detail APIs."""
+    if hasattr(tool, 'metadata') and hasattr(tool, 'spec'):
+        metadata = tool.metadata
+        spec = tool.spec or {}
+        description = spec.get('llm_prompt') or _i18n_to_text(getattr(metadata, 'description', None))
+        parameters = spec.get('parameters') or {}
+
+        return {
+            'name': requested_tool_name or getattr(metadata, 'name', ''),
+            'label': _i18n_to_dict(getattr(metadata, 'label', None)),
+            'description': description,
+            'human_desc': description,
+            'parameters': parameters,
+            'spec': spec,
+        }
+
+    name = getattr(tool, 'name', requested_tool_name or '')
+    description = getattr(tool, 'description', None) or getattr(tool, 'human_desc', '') or ''
+    parameters = getattr(tool, 'parameters', None) or {}
+
+    return {
+        'name': name,
+        'label': {},
+        'description': description,
+        'human_desc': getattr(tool, 'human_desc', description) or description,
+        'parameters': parameters,
+        'spec': {'parameters': parameters},
+    }
+
+
 async def _validate_run_authorization(
     run_id: str,
     resource_type: str,
@@ -462,7 +519,13 @@ class RuntimeConnectionHandler(handler.Handler):
                 return
 
             messages_obj = [provider_message.Message.model_validate(message) for message in messages]
-            funcs_obj = [resource_tool.LLMTool.model_validate(func) for func in funcs]
+
+            # The func field is excluded during model_dump() in plugin side
+            # but required by LLMTool validation on Host.
+            async def _placeholder_func(**kwargs):
+                pass
+
+            funcs_obj = [resource_tool.LLMTool.model_validate({**func, 'func': _placeholder_func}) for func in funcs]
 
             async for chunk in llm_model.provider.invoke_llm_stream(
                 query=None,
@@ -538,30 +601,26 @@ class RuntimeConnectionHandler(handler.Handler):
             """
             tool_name = data['tool_name']
             run_id = data.get('run_id')  # Optional: present for AgentRunner calls
+            caller_plugin_identity = data.get('caller_plugin_identity')  # Optional: for cross-plugin validation
 
             # Permission validation for AgentRunner calls
             if run_id:
                 session, error = await _validate_run_authorization(
-                    run_id, 'tool', tool_name, self.ap
+                    run_id, 'tool', tool_name, self.ap, caller_plugin_identity
                 )
                 if error:
                     return error
 
             try:
-                tool = self.ap.tool_mgr.get_tool_by_name(tool_name)
+                tool = await self.ap.tool_mgr.get_tool_by_name(tool_name)
                 if tool is None:
                     return handler.ActionResponse.error(
                         message=f'Tool {tool_name} not found',
                     )
 
-                # Build tool detail for LLM function calling
-                tool_detail = {
-                    'name': tool.name,
-                    'description': tool.description or '',
-                    'parameters': tool.parameters or {},
-                }
+                tool_detail = _build_tool_detail(tool, requested_tool_name=tool_name)
 
-                return handler.ActionResponse.success(data=tool_detail)
+                return handler.ActionResponse.success(data={'tool': tool_detail})
             except Exception as e:
                 traceback.print_exc()
                 return handler.ActionResponse.error(

src/langbot/pkg/provider/modelmgr/modelmgr.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import sqlalchemy
 import traceback
 
@@ -55,7 +56,21 @@ class ModelManager:
             return
 
         try:
-            await self.sync_new_models_from_space()
+            sync_timeout = float(space_config.get('models_sync_timeout', 10))
+        except (TypeError, ValueError):
+            sync_timeout = 10
+
+        try:
+            self.ap.logger.info('Syncing new models from LangBot Space...')
+            if sync_timeout > 0:
+                await asyncio.wait_for(self.sync_new_models_from_space(), timeout=sync_timeout)
+            else:
+                await self.sync_new_models_from_space()
+            self.ap.logger.info('LangBot Space model sync completed.')
+        except asyncio.TimeoutError:
+            self.ap.logger.warning(
+                f'LangBot Space model sync timed out after {sync_timeout:g}s, skipping startup sync.'
+            )
         except Exception as e:
             self.ap.logger.warning('Failed to sync new models from LangBot Space, model list may not be updated.')
             self.ap.logger.warning(f'  - Error: {e}')
@@ -73,6 +88,9 @@ class ModelManager:
         )
         for provider in providers_result.all():
             try:
+                self.ap.logger.info(
+                    f'Loading model provider {provider.uuid} ({provider.name}, requester={provider.requester})...'
+                )
                 runtime_provider = await self.load_provider(provider)
                 self.provider_dict[provider.uuid] = runtime_provider
             except provider_errors.RequesterNotFoundError as e:
@@ -127,6 +145,14 @@ class ModelManager:
             except Exception as e:
                 self.ap.logger.error(f'Failed to load model {rerank_model.uuid}: {e}\n{traceback.format_exc()}')
 
+        self.ap.logger.info(
+            'Loaded models from db: '
+            f'{len(self.provider_dict)} providers, '
+            f'{len(self.llm_models)} llm models, '
+            f'{len(self.embedding_models)} embedding models, '
+            f'{len(self.rerank_models)} rerank models.'
+        )
+
     async def sync_new_models_from_space(self):
         """Sync models from Space"""
         space_model_provider = await self.ap.persistence_mgr.execute_async(

src/langbot/templates/config.yaml

@@ -137,4 +137,6 @@ space:
     # OAuth authorization page URL (user will be redirected here)
     oauth_authorize_url: 'https://space.langbot.app/auth/authorize'
     disable_models_service: false
+    # Max seconds to wait for startup model-list sync. Set to 0 to disable the timeout.
+    models_sync_timeout: 10
     disable_telemetry: false

src/langbot/templates/default-pipeline-config.json

@@ -38,12 +38,10 @@
     },
     "ai": {
         "runner": {
-            "id": "plugin:langbot/local-agent/default",
+            "id": "",
             "expire-time": 0
         },
-        "runner_config": {
-            "plugin:langbot/local-agent/default": {}
-        }
+        "runner_config": {}
     },
     "output": {
         "long-text-processing": {
@@ -64,4 +62,4 @@
             "remove-think": false
         }
     }
-}
+}