Merge origin/master into feat/totp-two-factor-auth

Resolve the revision-graph conflict introduced by the release line.

Conflicts
- tests/integration/persistence/test_rag_document_identity.py: master had
  independently introduced the same dynamic-head helper (`current_head`) plus
  a `DOCUMENT_IDENTITY_REVISION` constant, and it explicitly upgrades to that
  revision. Keep master's semantics (the explicit revision matters because
  upgrading to the head now also traverses the TOTP branch) while keeping the
  module-level alembic imports on this branch.

New migration
- 0030_merge_totp_into_release joins the TOTP branch's own merge revision
  (0026_merge_totp_and_rag_identity) with the release head
  (0029_merge_rag_identity). Both reached 0025_rag_document_identity without
  including each other, which left Alembic with two heads and made
  `upgrade head` fail with "Multiple head revisions are present".

Verification
- Single Alembic head confirmed (0030_merge_totp_into_release).
- tests/integration/persistence/test_rag_document_identity.py: 34 passed.
- Full fast integration suite: 356 passed, 84 skipped.

src/langbot/pkg/api/http/controller/group.py

@@ -27,6 +27,7 @@ from ..authz import (
 )
 from ..context import PrincipalContext, PrincipalType, RequestContext, WorkspaceContext
 from ....cloud.support_admin import SupportAdminSessionError
+from ... import management_diagnostics as diagnostics
 
 if typing.TYPE_CHECKING:
     from ....core.app import Application
@@ -224,6 +225,7 @@ class RouterGroup(abc.ABC):
 
                 try:
                     if request_context is not None:
+                        diagnostics.workspace(request_context)
                         with bounded_executor.blocking_work_scope(request_context.workspace_uuid):
                             persistence_mgr = getattr(
                                 self.ap,
@@ -277,7 +279,14 @@ class RouterGroup(abc.ABC):
                     )
                     return self.internal_error_response(request_id)
 
-            new_f = handler_error
+            # Observe outside authentication, using the registered Core handler
+            # identity rather than the URL (which can contain user identifiers).
+            new_f = diagnostics.observe(
+                diagnostics.operation_id('http', f, rule=rule, methods=options.get('methods')),
+                source='http',
+                ap=self.ap,
+                http=True,
+            )(handler_error)
             # Quart/Flask requires a unique endpoint name even when the same URL
             # intentionally has separate handlers for different HTTP methods.
             # Include the method set so CRUD routes can declare distinct
@@ -564,6 +573,7 @@ class RouterGroup(abc.ABC):
     def fail(self, code: int | str, msg: str) -> quart.Response:
         """Return an error response"""
 
+        diagnostics.outcome('failed')
         return quart.jsonify(
             {
                 'code': code,

src/langbot/pkg/api/http/controller/groups/agent_debug_stream.py

@@ -0,0 +1,72 @@
+"""Bounded, cancellable NDJSON transport for Agent debug execution."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import json
+
+import quart
+from .... import management_diagnostics as diagnostics
+
+from .....agent.runner.errors import (
+    RunnerError,
+    RunnerExecutionError,
+    RunnerNotAuthorizedError,
+    RunnerNotFoundError,
+    RunnerProtocolError,
+)
+
+
+def debug_stream_response(service, context, agent_uuid: str, payload: dict) -> quart.Response:
+    async def stream():
+        queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=32)
+
+        async def on_result(result: dict) -> None:
+            await queue.put({'kind': 'result', 'data': result})
+
+        async def execute() -> None:
+            try:
+                result = await service.debug_agent(context, agent_uuid, payload, on_result=on_result)
+                await queue.put({'kind': 'completed', 'data': result})
+            except Exception as exc:
+                # The stream still uses HTTP 200 when execution returns an
+                # error frame. Mark the inherited request, not its contents.
+                diagnostics.outcome('failed')
+                if isinstance(exc, RunnerExecutionError):
+                    code, message = exc.error_code or 'runner_execution_failed', exc.message
+                elif isinstance(exc, RunnerNotFoundError):
+                    code, message = 'runner_not_found', 'The configured Agent runner is unavailable'
+                elif isinstance(exc, RunnerNotAuthorizedError):
+                    code, message = 'runner_not_authorized', 'The configured Agent runner is not authorized'
+                elif isinstance(exc, RunnerProtocolError):
+                    code, message = 'runner_protocol_error', 'The Agent runner returned an invalid response'
+                elif isinstance(exc, ValueError):
+                    code, message = 'invalid_request', str(exc)
+                elif isinstance(exc, RunnerError):
+                    code, message = 'runner_error', 'The Agent runner could not complete this test'
+                else:
+                    code, message = 'runner_error', 'The Agent debug execution failed'
+                await queue.put({'kind': 'error', 'code': code, 'msg': message})
+
+        task = asyncio.create_task(execute())
+        try:
+            while True:
+                try:
+                    frame = await asyncio.wait_for(queue.get(), timeout=15)
+                except TimeoutError:
+                    yield '\n'
+                    continue
+                yield json.dumps(frame, ensure_ascii=False) + '\n'
+                if frame['kind'] in {'completed', 'error'}:
+                    break
+        finally:
+            task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await task
+
+    response = quart.Response(stream(), content_type='application/x-ndjson; charset=utf-8')
+    response.timeout = None
+    response.headers['Cache-Control'] = 'no-store'
+    response.headers['X-Accel-Buffering'] = 'no'
+    return response

src/langbot/pkg/api/http/controller/groups/agents.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+import quart
+
+from .....agent.runner.errors import (
+    RunnerError,
+    RunnerExecutionError,
+    RunnerNotAuthorizedError,
+    RunnerNotFoundError,
+    RunnerProtocolError,
+)
+from ...authz import Permission, require_permission
+from ...context import RequestContext
+from .. import group
+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(
+            '/<agent_uuid>/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(
+            '/<agent_uuid>/runs/<run_id>/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(
+            '/<agent_uuid>/debug/stream',
+            methods=['POST'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RUNTIME_OPERATE,
+        )
+        async def stream_debug(agent_uuid: str, request_context: RequestContext):
+            payload = await quart.request.get_json()
+            if not isinstance(payload, dict):
+                return self.http_status(400, -1, 'Debug payload must be an object')
+            return debug_stream_response(self.ap.agent_service, request_context, agent_uuid, payload)
+
+        @self.route(
+            '',
+            methods=['GET', 'POST'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
+        async def _(request_context: RequestContext) -> str:
+            if quart.request.method == 'GET':
+                sort_by = quart.request.args.get('sort_by', 'updated_at')
+                sort_order = quart.request.args.get('sort_order', 'DESC')
+                return self.success(
+                    data={
+                        'agents': await self.ap.agent_service.get_agents(
+                            request_context,
+                            sort_by,
+                            sort_order,
+                        )
+                    }
+                )
+
+            json_data = await quart.request.json
+            require_permission(request_context, Permission.RESOURCE_MANAGE)
+            try:
+                created = await self.ap.agent_service.create_agent(request_context, json_data)
+            except ValueError as exc:
+                return self.http_status(400, -1, str(exc))
+            return self.success(data=created)
+
+        @self.route(
+            '/_/metadata',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
+        async def _(request_context: RequestContext) -> str:
+            return self.success(data=await self.ap.agent_service.get_agent_metadata(request_context))
+
+        @self.route(
+            '/<agent_uuid>/debug',
+            methods=['POST'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RUNTIME_OPERATE,
+        )
+        async def _(agent_uuid: str, request_context: RequestContext) -> str:
+            json_data = await quart.request.json
+            try:
+                result = await self.ap.agent_service.debug_agent(
+                    request_context,
+                    agent_uuid,
+                    json_data or {},
+                )
+            except ValueError as exc:
+                return self.http_status(400, -1, str(exc))
+            except RunnerExecutionError as exc:
+                return self.http_status(
+                    422,
+                    exc.error_code or 'runner_execution_failed',
+                    exc.message,
+                )
+            except RunnerNotFoundError:
+                return self.http_status(
+                    409,
+                    'runner_not_found',
+                    'The configured Agent runner is unavailable',
+                )
+            except RunnerNotAuthorizedError:
+                return self.http_status(
+                    403,
+                    'runner_not_authorized',
+                    'The configured Agent runner is not authorized',
+                )
+            except RunnerProtocolError:
+                return self.http_status(
+                    502,
+                    'runner_protocol_error',
+                    'The Agent runner returned an invalid response',
+                )
+            except RunnerError:
+                return self.http_status(
+                    502,
+                    'runner_error',
+                    'The Agent runner could not complete this test',
+                )
+            return self.success(data=result)
+
+        @self.route(
+            '/<agent_uuid>',
+            methods=['GET', 'PUT', 'DELETE'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
+        async def _(agent_uuid: str, request_context: RequestContext) -> str:
+            if quart.request.method == 'GET':
+                agent = await self.ap.agent_service.get_agent(request_context, agent_uuid)
+                if agent is None:
+                    return self.http_status(404, -1, 'agent not found')
+                return self.success(data={'agent': agent})
+
+            if quart.request.method == 'PUT':
+                require_permission(request_context, Permission.RESOURCE_MANAGE)
+                json_data = await quart.request.json
+                try:
+                    await self.ap.agent_service.update_agent(request_context, agent_uuid, json_data)
+                except ValueError as exc:
+                    return self.http_status(400, -1, str(exc))
+                return self.success()
+
+            require_permission(request_context, Permission.RESOURCE_MANAGE)
+            await self.ap.agent_service.delete_agent(request_context, agent_uuid)
+            return self.success()

src/langbot/pkg/api/http/controller/groups/pipelines/embed.py

@@ -3,7 +3,7 @@
 All user-facing URLs are keyed by **bot_uuid** (not pipeline_uuid) so that
 internal pipeline identifiers are never exposed to end-users.  Each handler
 resolves the bot_uuid to the owning ``web_page_bot`` RuntimeBot and extracts
-the bound pipeline_uuid for internal routing.
+the message event route's Pipeline target for internal routing.
 """
 
 import asyncio
@@ -13,6 +13,7 @@ import logging
 import uuid
 import hmac
 import hashlib
+import os
 import time
 import re
 import httpx
@@ -20,6 +21,7 @@ import httpx
 import quart
 
 from ... import group
+from ..... import management_diagnostics as diagnostics
 from ......utils import httpclient, paths
 from ......platform.sources.websocket_manager import WebSocketScope, is_valid_session_id, ws_connection_manager
 from .websocket_chat import create_scoped_duplex_tasks, wait_for_duplex_tasks
@@ -29,6 +31,7 @@ _AUTH_TIMEOUT_SECONDS = 10.0
 
 # Cache the widget template content
 _widget_template_cache: str | None = None
+_widget_template_cache_mtime_ns: int | None = None
 _logo_bytes_cache: bytes | None = None
 
 
@@ -37,12 +40,14 @@ def _is_valid_uuid(s: str) -> bool:
 
 
 def _get_widget_template() -> str:
-    """Load and cache the widget JS template."""
-    global _widget_template_cache
-    if _widget_template_cache is None:
-        template_path = paths.get_resource_path('templates/embed/widget.js')
+    """Load the widget template and refresh the cache when the file changes."""
+    global _widget_template_cache, _widget_template_cache_mtime_ns
+    template_path = paths.get_resource_path('templates/embed/widget.js')
+    template_mtime_ns = os.stat(template_path).st_mtime_ns
+    if _widget_template_cache is None or _widget_template_cache_mtime_ns != template_mtime_ns:
         with open(template_path, 'r', encoding='utf-8') as f:
             _widget_template_cache = f.read()
+        _widget_template_cache_mtime_ns = template_mtime_ns
     return _widget_template_cache
 
 
@@ -64,16 +69,12 @@ class EmbedRouterGroup(group.RouterGroup):
         """Resolve *bot_uuid* to ``(runtime_bot, pipeline_uuid)``.
 
         Returns ``(None, None)`` when the bot does not exist, is not a
-        ``web_page_bot``, is disabled, or has no pipeline bound.
+        ``web_page_bot``, is disabled, or has no Pipeline target for messages.
         """
         bot = await self.ap.platform_mgr.resolve_public_bot(bot_uuid)
-        if (
-            bot is not None
-            and bot.bot_entity.adapter == 'web_page_bot'
-            and bot.bot_entity.enable
-            and bot.bot_entity.use_pipeline_uuid
-        ):
-            return bot, bot.bot_entity.use_pipeline_uuid
+        pipeline_uuid = bot.get_pipeline_target_for_event_type('message.received') if bot is not None else None
+        if bot is not None and bot.bot_entity.adapter == 'web_page_bot' and bot.bot_entity.enable and pipeline_uuid:
+            return bot, pipeline_uuid
         return None, None
 
     @staticmethod
@@ -198,7 +199,9 @@ class EmbedRouterGroup(group.RouterGroup):
 
             base_url = quart.request.host_url.rstrip('/')
             webhook_prefix = self.ap.instance_config.data.get('api', {}).get('webhook_prefix', '')
-            if webhook_prefix:
+            # The wizard previews the currently connected backend, which can
+            # differ from the public address used by external website embeds.
+            if webhook_prefix and quart.request.args.get('preview') != 'wizard':
                 base_url = webhook_prefix.rstrip('/')
 
             if not re.match(r'^https?://[a-zA-Z0-9._:/-]+$', base_url):
@@ -327,20 +330,24 @@ class EmbedRouterGroup(group.RouterGroup):
         # -- Embed WebSocket endpoint ----------------------------------------
 
         @self.quart_app.websocket(self.path + '/<bot_uuid>/ws/connect')
+        @diagnostics.observe('websocket.embed.session', source='websocket', ap=self.ap)
         async def embed_websocket_connect(bot_uuid: str):
             """WebSocket connection for embed widget, keyed by bot_uuid."""
             await quart.websocket.accept()
             if not _is_valid_uuid(bot_uuid):
                 await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Invalid bot_uuid format'}))
+                diagnostics.outcome('rejected')
                 return
 
             runtime_bot, pipeline_uuid = await self._resolve_bot(bot_uuid)
             if runtime_bot is None:
                 await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Bot not found or not available'}))
+                diagnostics.outcome('rejected')
                 return
 
             session_type = quart.websocket.args.get('session_type', 'person')
             if session_type not in ['person', 'group']:
+                diagnostics.outcome('rejected')
                 await quart.websocket.send(
                     json.dumps({'type': 'error', 'message': 'session_type must be person or group'})
                 )
@@ -349,6 +356,7 @@ class EmbedRouterGroup(group.RouterGroup):
             session_id = quart.websocket.args.get('session_id', '')
             if not is_valid_session_id(session_id):
                 await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Valid session_id is required'}))
+                diagnostics.outcome('rejected')
                 return
 
             try:
@@ -356,13 +364,16 @@ class EmbedRouterGroup(group.RouterGroup):
                 await self._assert_execution_active(runtime_bot)
             except Exception:
                 await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Unauthorized'}))
+                diagnostics.outcome('rejected')
                 return
 
+            diagnostics.workspace(runtime_bot.execution_context)
             try:
                 proxy_bot = await self.ap.platform_mgr.get_websocket_proxy_bot(runtime_bot.execution_context)
                 websocket_adapter = proxy_bot.adapter
                 if not websocket_adapter:
                     await quart.websocket.send(json.dumps({'type': 'error', 'message': 'WebSocket adapter not found'}))
+                    diagnostics.outcome('rejected')
                     return
 
                 connection = await ws_connection_manager.add_connection(
@@ -420,11 +431,13 @@ class EmbedRouterGroup(group.RouterGroup):
                 try:
                     await wait_for_duplex_tasks(receive_task, send_task)
                 except Exception as e:
+                    diagnostics.outcome('failed')
                     logger.error(f'Embed WebSocket task error: {e}')
                 finally:
                     await ws_connection_manager.remove_connection(connection.connection_id)
 
             except Exception as e:
+                diagnostics.outcome('failed')
                 logger.error(f'Embed WebSocket connection error: {e}', exc_info=True)
                 try:
                     await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Internal server error'}))
@@ -439,28 +452,35 @@ class EmbedRouterGroup(group.RouterGroup):
                 message = await quart.websocket.receive()
                 await ws_connection_manager.update_activity(connection.connection_id)
 
-                try:
-                    data = await asyncio.to_thread(json.loads, message)
-                    message_type = data.get('type', 'message')
+                with diagnostics.scope(self.ap, 'websocket.embed.message', source='websocket'):
+                    diagnostics.workspace(getattr(owner_bot, 'execution_context', None))
+                    try:
+                        data = await asyncio.to_thread(json.loads, message)
+                        message_type = data.get('type', 'message')
 
-                    if message_type == 'ping':
-                        await connection.send_queue.put(
-                            {'type': 'pong', 'timestamp': datetime.datetime.now().isoformat()}
-                        )
-                    elif message_type == 'message':
-                        try:
-                            current_bot = await self._resolve_connected_bot(owner_bot, pipeline_uuid)
-                        except Exception:
-                            await connection.send_queue.put({'type': 'error', 'message': 'Bot is unavailable'})
+                        if message_type == 'ping':
+                            await connection.send_queue.put(
+                                {'type': 'pong', 'timestamp': datetime.datetime.now().isoformat()}
+                            )
+                        elif message_type == 'message':
+                            try:
+                                current_bot = await self._resolve_connected_bot(owner_bot, pipeline_uuid)
+                            except Exception:
+                                diagnostics.outcome('rejected')
+                                await connection.send_queue.put({'type': 'error', 'message': 'Bot is unavailable'})
+                                break
+                            await websocket_adapter.handle_websocket_message(connection, data, owner_bot=current_bot)
+                        elif message_type == 'disconnect':
                             break
-                        await websocket_adapter.handle_websocket_message(connection, data, owner_bot=current_bot)
-                    elif message_type == 'disconnect':
-                        break
+                        else:
+                            diagnostics.outcome('skipped')
 
-                except json.JSONDecodeError:
-                    await connection.send_queue.put({'type': 'error', 'message': 'Invalid JSON format'})
+                    except json.JSONDecodeError:
+                        diagnostics.outcome('rejected')
+                        await connection.send_queue.put({'type': 'error', 'message': 'Invalid JSON format'})
 
         except Exception as e:
+            diagnostics.outcome('failed')
             logger.error(f'Embed receive error: {e}', exc_info=True)
         finally:
             connection.is_active = False
@@ -476,11 +496,13 @@ class EmbedRouterGroup(group.RouterGroup):
                     message = await asyncio.wait_for(connection.send_queue.get(), timeout=1.0)
                     if message is None:
                         break
-                    encoded = await asyncio.to_thread(json.dumps, message)
-                    await quart.websocket.send(encoded)
+                    with diagnostics.scope(self.ap, 'websocket.embed.send', source='websocket'):
+                        encoded = await asyncio.to_thread(json.dumps, message)
+                        await quart.websocket.send(encoded)
                 except asyncio.TimeoutError:
                     continue
         except Exception as e:
+            diagnostics.outcome('failed')
             logger.error(f'Embed send error: {e}', exc_info=True)
         finally:
             connection.is_active = False

src/langbot/pkg/api/http/controller/groups/pipelines/migration.py

@@ -0,0 +1,33 @@
+"""User-confirmed migration only; deliberately not API-key/MCP accessible."""
+
+import quart
+
+from ... import group
+from ....authz import Permission
+from ....context import RequestContext
+from ....service.pipeline_migration import PipelineMigrationService, MigrationError
+
+
+@group.group_class('pipeline_migration', '/api/v1/pipelines/_/migration')
+class PipelineMigrationRouterGroup(group.RouterGroup):
+    async def initialize(self):
+        self.service = PipelineMigrationService(self.ap)
+
+        @self.route(
+            '/preview', methods=['GET'], auth_type=group.AuthType.USER_TOKEN, permission=Permission.RESOURCE_VIEW
+        )
+        async def preview(request_context: RequestContext):
+            try:
+                return self.success(data=await self.service.preview(request_context))
+            except MigrationError as exc:
+                return self.http_status(exc.status_code, -1, exc.code)
+
+        @self.route(
+            '/execute', methods=['POST'], auth_type=group.AuthType.USER_TOKEN, permission=Permission.RESOURCE_MANAGE
+        )
+        async def execute(request_context: RequestContext):
+            try:
+                body = await quart.request.get_json(silent=True)
+                return self.success(data=await self.service.execute(request_context, body))
+            except MigrationError as exc:
+                return self.http_status(exc.status_code, -1, exc.code)

src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py

@@ -6,6 +6,10 @@ from ....authz import Permission, has_permission
 from ....context import RequestContext
 from ....service.secrets import redact_secrets
 from ... import group
+from ......pipeline.extension_preferences import (
+    normalize_extension_preferences,
+    validate_extension_preferences,
+)
 
 
 @group.group_class('pipelines', '/api/v1/pipelines')
@@ -41,11 +45,14 @@ class PipelinesRouterGroup(group.RouterGroup):
         async def _(request_context: RequestContext) -> str:
             pipeline_data = await quart.request.json
             create_as_default = pipeline_data.get('is_default') is True
-            pipeline_uuid = await self.ap.pipeline_service.create_pipeline(
-                request_context,
-                pipeline_data,
-                default=create_as_default,
-            )
+            try:
+                pipeline_uuid = await self.ap.pipeline_service.create_pipeline(
+                    request_context,
+                    pipeline_data,
+                    default=create_as_default,
+                )
+            except ValueError as exc:
+                return self.http_status(400, -1, str(exc))
             return self.success(data={'uuid': pipeline_uuid})
 
         @self.route(
@@ -122,8 +129,12 @@ class PipelinesRouterGroup(group.RouterGroup):
                 await self.ap.plugin_connector.require_workspace_context(request_context)
             plugins = await self.ap.plugin_connector.list_plugins(component_kinds=pipeline_component_kinds)
             mcp_servers = await self.ap.mcp_service.get_mcp_servers(request_context, contain_runtime_info=True)
-            available_skills = await self.ap.skill_service.list_skills(request_context)
-            extensions_prefs = pipeline.get('extensions_preferences', {})
+            try:
+                available_skills = await self.ap.skill_service.list_skills(request_context)
+            except Exception as exc:
+                self.ap.logger.warning('Unable to list skills for pipeline extensions: %s', exc)
+                available_skills = []
+            extensions_prefs = normalize_extension_preferences(pipeline.get('extensions_preferences'))
             return self.success(
                 data={
                     'enable_all_plugins': extensions_prefs.get('enable_all_plugins', True),
@@ -148,16 +159,42 @@ class PipelinesRouterGroup(group.RouterGroup):
         )
         async def _(pipeline_uuid: str, request_context: RequestContext) -> str:
             json_data = await quart.request.json
-            await self.ap.pipeline_service.update_pipeline_extensions(
-                request_context,
-                pipeline_uuid,
-                json_data.get('bound_plugins', []),
-                json_data.get('bound_mcp_servers', []),
-                json_data.get('enable_all_plugins', True),
-                json_data.get('enable_all_mcp_servers', True),
-                bound_skills=json_data.get('bound_skills', []),
-                enable_all_skills=json_data.get('enable_all_skills', True),
-                bound_mcp_resources=json_data.get('bound_mcp_resources'),
-                mcp_resource_agent_read_enabled=json_data.get('mcp_resource_agent_read_enabled'),
-            )
+            try:
+                validate_extension_preferences(
+                    {
+                        canonical: json_data[field]
+                        for field, canonical in {
+                            'bound_plugins': 'plugins',
+                            'bound_mcp_servers': 'mcp_servers',
+                            'bound_skills': 'skills',
+                            'bound_mcp_resources': 'mcp_resources',
+                            'enable_all_plugins': 'enable_all_plugins',
+                            'enable_all_mcp_servers': 'enable_all_mcp_servers',
+                            'enable_all_skills': 'enable_all_skills',
+                            'mcp_resource_agent_read_enabled': 'mcp_resource_agent_read_enabled',
+                        }.items()
+                        if field in json_data
+                    },
+                    context='Pipeline extension',
+                    field_aliases={
+                        'plugins': 'bound_plugins',
+                        'mcp_servers': 'bound_mcp_servers',
+                        'skills': 'bound_skills',
+                        'mcp_resources': 'bound_mcp_resources',
+                    },
+                )
+                await self.ap.pipeline_service.update_pipeline_extensions(
+                    request_context,
+                    pipeline_uuid,
+                    json_data.get('bound_plugins', []),
+                    json_data.get('bound_mcp_servers', []),
+                    json_data.get('enable_all_plugins', True),
+                    json_data.get('enable_all_mcp_servers', True),
+                    bound_skills=json_data.get('bound_skills', []),
+                    enable_all_skills=json_data.get('enable_all_skills', True),
+                    bound_mcp_resources=json_data.get('bound_mcp_resources'),
+                    mcp_resource_agent_read_enabled=json_data.get('mcp_resource_agent_read_enabled'),
+                )
+            except ValueError as exc:
+                return self.http_status(400, -1, str(exc))
             return self.success()

src/langbot/pkg/api/http/controller/groups/pipelines/websocket_chat.py

@@ -14,6 +14,7 @@ import quart
 from ....authz import Permission, permissions_for_role, require_permission
 from ....context import PrincipalContext, PrincipalType, RequestContext, WorkspaceContext
 from ... import group
+from ..... import management_diagnostics as diagnostics
 from ......core.task_boundary import run_in_workspace_uow
 from ......platform.sources.websocket_manager import WebSocketScope, ws_connection_manager
 from ......utils import bounded_executor
@@ -210,6 +211,7 @@ class WebSocketChatRouterGroup(group.RouterGroup):
 
     async def initialize(self) -> None:
         @self.quart_app.websocket(self.path + '/connect')
+        @diagnostics.observe('websocket.dashboard.session', source='websocket', ap=self.ap)
         async def websocket_connect(pipeline_uuid: str):
             """Open one authenticated dashboard debug connection."""
 
@@ -217,11 +219,14 @@ class WebSocketChatRouterGroup(group.RouterGroup):
             try:
                 request_context, token = await self._authenticate_websocket()
             except Exception:
+                diagnostics.outcome('rejected')
                 await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Unauthorized'}))
                 return
 
+            diagnostics.workspace(request_context)
             session_type = quart.websocket.args.get('session_type', 'person')
             if session_type not in ['person', 'group']:
+                diagnostics.outcome('rejected')
                 await quart.websocket.send(
                     json.dumps({'type': 'error', 'message': 'session_type must be person or group'})
                 )
@@ -230,6 +235,7 @@ class WebSocketChatRouterGroup(group.RouterGroup):
             try:
                 websocket_adapter = await self._get_scoped_adapter(request_context, pipeline_uuid)
                 if websocket_adapter is None:
+                    diagnostics.outcome('rejected')
                     await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Pipeline not found'}))
                     return
 
@@ -288,11 +294,13 @@ class WebSocketChatRouterGroup(group.RouterGroup):
                 try:
                     await wait_for_duplex_tasks(receive_task, send_task)
                 except Exception as exc:
+                    diagnostics.outcome('failed')
                     logger.error(f'WebSocket task execution error: {exc}')
                 finally:
                     await ws_connection_manager.remove_connection(connection.connection_id)
 
             except Exception:
+                diagnostics.outcome('failed')
                 logger.error('Dashboard WebSocket connection error', exc_info=True)
                 try:
                     await quart.websocket.send(json.dumps({'type': 'error', 'message': 'Internal server error'}))
@@ -410,28 +418,34 @@ class WebSocketChatRouterGroup(group.RouterGroup):
                 message = await quart.websocket.receive()
                 await ws_connection_manager.update_activity(connection.connection_id)
 
-                try:
-                    data = await asyncio.to_thread(json.loads, message)
-                    message_type = data.get('type', 'message')
-                    if message_type == 'ping':
-                        await connection.send_queue.put(
-                            {'type': 'pong', 'timestamp': datetime.datetime.now().isoformat()}
-                        )
-                    elif message_type == 'message':
-                        try:
-                            request_context = await self._revalidate_websocket_authorization(request_context, token)
-                        except Exception:
-                            await connection.send_queue.put({'type': 'error', 'message': 'Unauthorized'})
+                with diagnostics.scope(self.ap, 'websocket.dashboard.message', source='websocket'):
+                    diagnostics.workspace(request_context)
+                    try:
+                        data = await asyncio.to_thread(json.loads, message)
+                        message_type = data.get('type', 'message')
+                        if message_type == 'ping':
+                            await connection.send_queue.put(
+                                {'type': 'pong', 'timestamp': datetime.datetime.now().isoformat()}
+                            )
+                        elif message_type == 'message':
+                            try:
+                                request_context = await self._revalidate_websocket_authorization(request_context, token)
+                            except Exception:
+                                diagnostics.outcome('rejected')
+                                await connection.send_queue.put({'type': 'error', 'message': 'Unauthorized'})
+                                break
+                            await websocket_adapter.handle_websocket_message(connection, data)
+                        elif message_type == 'disconnect':
                             break
-                        await websocket_adapter.handle_websocket_message(connection, data)
-                    elif message_type == 'disconnect':
-                        break
-                    else:
-                        logger.warning(f'Unknown WebSocket message type: {message_type}')
-                except json.JSONDecodeError:
-                    await connection.send_queue.put({'type': 'error', 'message': 'Invalid JSON format'})
+                        else:
+                            diagnostics.outcome('skipped')
+                            logger.warning(f'Unknown WebSocket message type: {message_type}')
+                    except json.JSONDecodeError:
+                        diagnostics.outcome('rejected')
+                        await connection.send_queue.put({'type': 'error', 'message': 'Invalid JSON format'})
 
         except Exception:
+            diagnostics.outcome('failed')
             logger.error('Dashboard WebSocket receive error', exc_info=True)
         finally:
             connection.is_active = False
@@ -447,11 +461,13 @@ class WebSocketChatRouterGroup(group.RouterGroup):
                     message = await asyncio.wait_for(connection.send_queue.get(), timeout=1.0)
                     if message is None:
                         break
-                    encoded = await asyncio.to_thread(json.dumps, message)
-                    await quart.websocket.send(encoded)
+                    with diagnostics.scope(self.ap, 'websocket.dashboard.send', source='websocket'):
+                        encoded = await asyncio.to_thread(json.dumps, message)
+                        await quart.websocket.send(encoded)
                 except asyncio.TimeoutError:
                     continue
         except Exception:
+            diagnostics.outcome('failed')
             logger.error('Dashboard WebSocket send error', exc_info=True)
         finally:
             connection.is_active = False

src/langbot/pkg/api/http/controller/groups/platform/bots.py

@@ -3,11 +3,20 @@ from sqlalchemy.exc import IntegrityError
 
 from ....authz import Permission, has_permission
 from ....context import RequestContext
+from ....service.bot_errors import BotApplyError, bot_error_message
 from ... import group
 
 
 @group.group_class('bots', '/api/v1/platform/bots')
 class BotsRouterGroup(group.RouterGroup):
+    def _apply_error_response(self, exc: BotApplyError):
+        request_id = self.request_id()
+        logger = getattr(self.ap, 'logger', self.quart_app.logger)
+        logger.error(f'Bot configuration apply failed request_id={request_id} bot_uuid={exc.bot_uuid}', exc_info=True)
+        return quart.jsonify(
+            code='bot_apply_failed', msg=str(exc), data={'uuid': exc.bot_uuid}, request_id=request_id
+        ), 400
+
     async def initialize(self) -> None:
         @self.route(
             '',
@@ -34,7 +43,14 @@ class BotsRouterGroup(group.RouterGroup):
         )
         async def _(request_context: RequestContext) -> str:
             json_data = await quart.request.json
-            bot_uuid = await self.ap.bot_service.create_bot(request_context, json_data)
+            if not isinstance(json_data, dict):
+                return self.http_status(400, 'invalid_bot_config', 'Bot configuration must be an object')
+            try:
+                bot_uuid = await self.ap.bot_service.create_bot(request_context, json_data)
+            except BotApplyError as exc:
+                return self._apply_error_response(exc)
+            except ValueError as exc:
+                return self.http_status(400, 'invalid_bot_config', bot_error_message(exc, json_data))
             return self.success(data={'uuid': bot_uuid})
 
         @self.route(
@@ -63,7 +79,14 @@ class BotsRouterGroup(group.RouterGroup):
         async def _(bot_uuid: str, request_context: RequestContext) -> str:
             if quart.request.method == 'PUT':
                 json_data = await quart.request.json
-                await self.ap.bot_service.update_bot(request_context, bot_uuid, json_data)
+                if not isinstance(json_data, dict):
+                    return self.http_status(400, 'invalid_bot_config', 'Bot configuration must be an object')
+                try:
+                    await self.ap.bot_service.update_bot(request_context, bot_uuid, json_data)
+                except BotApplyError as exc:
+                    return self._apply_error_response(exc)
+                except ValueError as exc:
+                    return self.http_status(400, 'invalid_bot_config', bot_error_message(exc, json_data))
             else:
                 await self.ap.bot_service.delete_bot(request_context, bot_uuid)
             return self.success()
@@ -83,6 +106,45 @@ class BotsRouterGroup(group.RouterGroup):
             )
             return self.success(data={'logs': logs, 'total_count': total_count})
 
+        @self.route(
+            '/<bot_uuid>/event-routes/status',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
+        async def _(bot_uuid: str, request_context: RequestContext) -> str:
+            return self.success(data=await self.ap.bot_service.list_event_route_statuses(request_context, bot_uuid))
+
+        async def _dry_run_event_route(bot_uuid: str, request_context: RequestContext) -> str:
+            json_data = await quart.request.json
+            if not isinstance(json_data, dict):
+                return self.http_status(400, -1, 'invalid request body')
+
+            result = await self.ap.bot_service.dry_run_event_route(
+                request_context,
+                bot_uuid=bot_uuid,
+                event_type=json_data.get('event_type'),
+                event_data=json_data.get('event_data', json_data.get('payload')),
+                context=json_data.get('context'),
+                event_bindings=json_data.get('event_bindings'),
+            )
+            return self.success(data=result)
+
+        self.route(
+            '/<bot_uuid>/event-routes/dry-run',
+            methods=['POST'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )(_dry_run_event_route)
+        # Backward-compatible alias for early local clients/tests created before
+        # the product route naming was settled.
+        self.route(
+            '/<bot_uuid>/event_route/dry_run',
+            methods=['POST'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )(_dry_run_event_route)
+
         @self.route(
             '/<bot_uuid>/send_message',
             methods=['POST'],

src/langbot/pkg/api/http/controller/groups/plugins.py

@@ -426,6 +426,15 @@ class PluginsRouterGroup(group.RouterGroup):
         async def _(author: str, plugin_name: str, request_context: RequestContext) -> str:
             execution_context = await self.ap.plugin_connector.require_workspace_context(request_context)
             ctx = taskmgr.TaskContext.new()
+            ctx.metadata.update(
+                {
+                    'plugin_name': f'{author}/{plugin_name}',
+                    'install_source': 'marketplace',
+                    'operation': 'upgrade',
+                    'progress_percent': 3,
+                }
+            )
+            ctx.set_current_action('checking for latest version')
             wrapper = self.ap.task_mgr.create_user_task(
                 self._run_fenced_plugin_operation(
                     execution_context,

src/langbot/pkg/api/http/controller/groups/provider/models.py

@@ -1,11 +1,15 @@
 import quart
 
+from langbot.pkg.provider.modelmgr import errors as provider_errors
 from ....authz import Permission, has_permission
 from ....context import RequestContext
 from ... import group
 from .query import resolve_include_secret
 
 
+MODEL_TEST_ERROR_CODE = 'model_test_failed'
+
+
 @group.group_class('models/llm', '/api/v1/provider/models/llm')
 class LLMModelsRouterGroup(group.RouterGroup):
     async def initialize(self) -> None:
@@ -108,7 +112,14 @@ class LLMModelsRouterGroup(group.RouterGroup):
             permission=Permission.PROVIDER_SECRET_MANAGE,
         )
         async def _(model_uuid: str, request_context: RequestContext) -> str:
-            await self.ap.llm_model_service.test_llm_model(request_context, model_uuid, await quart.request.json)
+            try:
+                await self.ap.llm_model_service.test_llm_model(
+                    request_context,
+                    model_uuid,
+                    await quart.request.json,
+                )
+            except provider_errors.RequesterError as exc:
+                return self.http_status(400, MODEL_TEST_ERROR_CODE, str(exc))
             return self.success()
 
 
@@ -214,9 +225,14 @@ class EmbeddingModelsRouterGroup(group.RouterGroup):
             permission=Permission.PROVIDER_SECRET_MANAGE,
         )
         async def _(model_uuid: str, request_context: RequestContext) -> str:
-            await self.ap.embedding_models_service.test_embedding_model(
-                request_context, model_uuid, await quart.request.json
-            )
+            try:
+                await self.ap.embedding_models_service.test_embedding_model(
+                    request_context,
+                    model_uuid,
+                    await quart.request.json,
+                )
+            except provider_errors.RequesterError as exc:
+                return self.http_status(400, MODEL_TEST_ERROR_CODE, str(exc))
             return self.success()
 
 
@@ -322,5 +338,12 @@ class RerankModelsRouterGroup(group.RouterGroup):
             permission=Permission.PROVIDER_SECRET_MANAGE,
         )
         async def _(model_uuid: str, request_context: RequestContext) -> str:
-            await self.ap.rerank_models_service.test_rerank_model(request_context, model_uuid, await quart.request.json)
+            try:
+                await self.ap.rerank_models_service.test_rerank_model(
+                    request_context,
+                    model_uuid,
+                    await quart.request.json,
+                )
+            except provider_errors.RequesterError as exc:
+                return self.http_status(400, MODEL_TEST_ERROR_CODE, str(exc))
             return self.success()

src/langbot/pkg/api/http/controller/groups/resources/tools.py

@@ -5,10 +5,40 @@ import quart
 from ....authz import Permission
 from ....context import RequestContext
 from ... import group
+from ......pipeline.extension_preferences import normalize_extension_preferences
 
 
 @group.group_class('tools', '/api/v1/tools')
 class ToolsRouterGroup(group.RouterGroup):
+    async def _get_scoped_tool_catalog(
+        self,
+        request_context: RequestContext,
+    ) -> list[dict] | None:
+        pipeline_uuid = quart.request.args.get('pipeline_uuid') or quart.request.args.get('pipeline_id')
+        bound_plugins: list[str] | None = None
+        bound_mcp_servers: list[str] | None = None
+
+        if pipeline_uuid:
+            pipeline = await self.ap.pipeline_service.get_pipeline(
+                request_context,
+                pipeline_uuid,
+            )
+            if pipeline is None:
+                return None
+
+            extensions_prefs = normalize_extension_preferences(pipeline.get('extensions_preferences'))
+            if not extensions_prefs['enable_all_plugins']:
+                bound_plugins = [f'{plugin["author"]}/{plugin["name"]}' for plugin in extensions_prefs['plugins']]
+            if not extensions_prefs['enable_all_mcp_servers']:
+                bound_mcp_servers = extensions_prefs['mcp_servers']
+
+        return await self.ap.tool_mgr.get_resolved_tool_catalog(
+            request_context,
+            bound_plugins,
+            bound_mcp_servers,
+            include_skill_authoring=True,
+        )
+
     async def initialize(self) -> None:
         @self.route(
             '',
@@ -18,57 +48,35 @@ class ToolsRouterGroup(group.RouterGroup):
         )
         async def _(request_context: RequestContext) -> str:
             """获取所有可用工具列表"""
-            pipeline_uuid = quart.request.args.get('pipeline_uuid') or quart.request.args.get('pipeline_id')
-            bound_plugins: list[str] | None = None
-            bound_mcp_servers: list[str] | None = None
-
-            if pipeline_uuid:
-                pipeline = await self.ap.pipeline_service.get_pipeline(request_context, pipeline_uuid)
-                if pipeline is None:
-                    return self.http_status(404, -1, 'pipeline not found')
-
-                extensions_prefs = pipeline.get('extensions_preferences', {}) or {}
-                if not extensions_prefs.get('enable_all_plugins', True):
-                    bound_plugins = [
-                        f'{plugin.get("author", "")}/{plugin.get("name", "")}'
-                        for plugin in extensions_prefs.get('plugins', [])
-                        if isinstance(plugin, dict) and plugin.get('name')
-                    ]
-                if not extensions_prefs.get('enable_all_mcp_servers', True):
-                    bound_mcp_servers = [
-                        server for server in (extensions_prefs.get('mcp_servers', []) or []) if isinstance(server, str)
-                    ]
-
-            return self.success(
-                data={
-                    'tools': await self.ap.tool_mgr.get_tool_catalog(
-                        request_context,
-                        bound_plugins,
-                        bound_mcp_servers,
-                        include_skill_authoring=True,
-                    )
-                }
-            )
+            catalog = await self._get_scoped_tool_catalog(request_context)
+            if catalog is None:
+                return self.http_status(404, -1, 'pipeline not found')
+            return self.success(data={'tools': catalog})
 
         @self.route(
-            '/<tool_name>',
+            '/<path:tool_name>',
             methods=['GET'],
             auth_type=group.AuthType.USER_TOKEN,
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(tool_name: str, request_context: RequestContext) -> str:
             """获取特定工具详情"""
-            tools = await self.ap.tool_mgr.get_all_tools(request_context, include_skill_authoring=True)
+            catalog = await self._get_scoped_tool_catalog(request_context)
+            if catalog is None:
+                return self.http_status(404, -1, 'pipeline not found')
 
-            for tool in tools:
-                if tool.name == tool_name:
+            for tool in catalog:
+                if tool.get('name') == tool_name:
                     return self.success(
                         data={
                             'tool': {
-                                'name': tool.name,
-                                'description': tool.description,
-                                'human_desc': tool.human_desc,
-                                'parameters': tool.parameters,
+                                'name': tool['name'],
+                                'description': tool.get('description') or '',
+                                'human_desc': tool.get('human_desc') or '',
+                                'parameters': tool.get('parameters') or {},
+                                'source': tool.get('source'),
+                                'source_name': tool.get('source_name'),
+                                'source_id': tool.get('source_id'),
                             }
                         }
                     )

src/langbot/pkg/api/http/controller/groups/system.py

@@ -273,8 +273,8 @@ class SystemRouterGroup(group.RouterGroup):
             """Save wizard progress to metadata table.
 
             Accepts JSON body with wizard state fields:
-            { "step": int, "selected_adapter": str|null, "created_bot_uuid": str|null,
-              "bot_saved": bool, "selected_runner": str|null }
+            { "step": int, "selected_scenario": str|null, "selected_adapter": str|null,
+              "created_bot_uuid": str|null, "bot_saved": bool, "selected_runner": str|null }
             """
             data = await quart.request.get_json(silent=True) or {}
             progress_json = json.dumps(data, ensure_ascii=False)
@@ -322,6 +322,21 @@ class SystemRouterGroup(group.RouterGroup):
                 return self.http_status(503, -1, str(exc))
             return self.success(data=model)
 
+        @self.route(
+            '/model-availability',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN,
+            permission=Permission.RESOURCE_VIEW,
+        )
+        async def _(request_context: RequestContext) -> str:
+            """Expose Space's latest persisted model probes to the WebUI."""
+            try:
+                models = await self.ap.space_service.get_model_selection()
+            except Exception as exc:
+                self.ap.logger.warning(f'Failed to load LangBot Models availability: {exc}')
+                return self.http_status(503, -1, 'Model availability is unavailable')
+            return self.success(data={'models': [model.model_dump(mode='json') for model in models]})
+
         @self.route(
             '/tasks',
             methods=['GET'],