feat(api): enable CLI diagnostics on current master

skills/skills/langbot-mcp-ops/SKILL.md

@@ -76,6 +76,11 @@ The tools wrap the LangBot service layer. Current tools (v1):
 | `list_knowledge_bases` / `get_knowledge_base` / `retrieve_knowledge_base` | RAG knowledge bases (incl. semantic search) |
 | `list_mcp_servers` | External MCP servers LangBot connects to (as a client) |
 | `list_skills` / `get_skill` | Installed skills |
+| `list_knowledge_engines` / `get_knowledge_engine_schema` / `list_knowledge_parsers` | Discover RAG configuration |
+| `get_pipeline_extensions` / `update_pipeline_extensions` | Read or completely replace extension bindings; all lists and switches required |
+| `run_pipeline` | One fresh-session turn; requires `runtime.operate`, executes configured models/tools, never auto-retry an unknown outcome |
+| `get_monitoring_records` / `get_monitoring_details` | Bounded Workspace records and existing message/session details |
+| `get_sandbox_diagnostics` | Read status (`resource.view`), sessions/errors (`audit.view`); managed sandbox admission still applies |
 
 Mutating tools (`create_*`, `update_*`) take a JSON object matching the same
 shape as the corresponding HTTP API request body. Discover resources with the

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

@@ -16,7 +16,7 @@ class BoxRouterGroup(group.RouterGroup):
         @self.route(
             '/status',
             methods=['GET'],
-            auth_type=group.AuthType.USER_TOKEN,
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(request_context: RequestContext) -> str:
@@ -42,7 +42,7 @@ class BoxRouterGroup(group.RouterGroup):
         @self.route(
             '/sessions',
             methods=['GET'],
-            auth_type=group.AuthType.USER_TOKEN,
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
             permission=Permission.AUDIT_VIEW,
         )
         async def _(request_context: RequestContext) -> str:
@@ -55,7 +55,7 @@ class BoxRouterGroup(group.RouterGroup):
         @self.route(
             '/errors',
             methods=['GET'],
-            auth_type=group.AuthType.USER_TOKEN,
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
             permission=Permission.AUDIT_VIEW,
         )
         async def _(request_context: RequestContext) -> str:

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

@@ -75,7 +75,12 @@ class MonitoringRouterGroup(group.RouterGroup):
 
             return self.success(data=stats)
 
-        @self.route('/messages', methods=['GET'], permission=Permission.RESOURCE_VIEW)
+        @self.route(
+            '/messages',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
         async def get_messages(request_context: RequestContext) -> str:
             """Get message logs"""
             # Parse query parameters
@@ -84,8 +89,9 @@ class MonitoringRouterGroup(group.RouterGroup):
             session_ids = quart.request.args.getlist('sessionId')
             start_time_str = quart.request.args.get('startTime')
             end_time_str = quart.request.args.get('endTime')
-            limit = int(quart.request.args.get('limit', 100))
-            offset = int(quart.request.args.get('offset', 0))
+            limit, offset = self.ap.monitoring_service.normalize_page_window(
+                quart.request.args.get('limit', 100), quart.request.args.get('offset', 0)
+            )
 
             # Parse datetime
             start_time = parse_iso_datetime(start_time_str)
@@ -111,7 +117,12 @@ class MonitoringRouterGroup(group.RouterGroup):
                 }
             )
 
-        @self.route('/llm-calls', methods=['GET'], permission=Permission.RESOURCE_VIEW)
+        @self.route(
+            '/llm-calls',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
         async def get_llm_calls(request_context: RequestContext) -> str:
             """Get LLM call records"""
             # Parse query parameters
@@ -119,8 +130,9 @@ class MonitoringRouterGroup(group.RouterGroup):
             pipeline_ids = quart.request.args.getlist('pipelineId')
             start_time_str = quart.request.args.get('startTime')
             end_time_str = quart.request.args.get('endTime')
-            limit = int(quart.request.args.get('limit', 100))
-            offset = int(quart.request.args.get('offset', 0))
+            limit, offset = self.ap.monitoring_service.normalize_page_window(
+                quart.request.args.get('limit', 100), quart.request.args.get('offset', 0)
+            )
 
             # Parse datetime
             start_time = parse_iso_datetime(start_time_str)
@@ -145,7 +157,12 @@ class MonitoringRouterGroup(group.RouterGroup):
                 }
             )
 
-        @self.route('/tool-calls', methods=['GET'], permission=Permission.RESOURCE_VIEW)
+        @self.route(
+            '/tool-calls',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
         async def get_tool_calls(request_context: RequestContext) -> str:
             """Get tool call records"""
             bot_ids = quart.request.args.getlist('botId')
@@ -153,8 +170,9 @@ class MonitoringRouterGroup(group.RouterGroup):
             session_ids = quart.request.args.getlist('sessionId')
             start_time_str = quart.request.args.get('startTime')
             end_time_str = quart.request.args.get('endTime')
-            limit = int(quart.request.args.get('limit', 100))
-            offset = int(quart.request.args.get('offset', 0))
+            limit, offset = self.ap.monitoring_service.normalize_page_window(
+                quart.request.args.get('limit', 100), quart.request.args.get('offset', 0)
+            )
 
             start_time = parse_iso_datetime(start_time_str)
             end_time = parse_iso_datetime(end_time_str)
@@ -179,15 +197,21 @@ class MonitoringRouterGroup(group.RouterGroup):
                 }
             )
 
-        @self.route('/embedding-calls', methods=['GET'], permission=Permission.RESOURCE_VIEW)
+        @self.route(
+            '/embedding-calls',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
         async def get_embedding_calls(request_context: RequestContext) -> str:
             """Get embedding call records"""
             # Parse query parameters
             start_time_str = quart.request.args.get('startTime')
             end_time_str = quart.request.args.get('endTime')
             knowledge_base_id = quart.request.args.get('knowledgeBaseId')
-            limit = int(quart.request.args.get('limit', 100))
-            offset = int(quart.request.args.get('offset', 0))
+            limit, offset = self.ap.monitoring_service.normalize_page_window(
+                quart.request.args.get('limit', 100), quart.request.args.get('offset', 0)
+            )
 
             # Parse datetime
             start_time = parse_iso_datetime(start_time_str)
@@ -211,7 +235,12 @@ class MonitoringRouterGroup(group.RouterGroup):
                 }
             )
 
-        @self.route('/sessions', methods=['GET'], permission=Permission.RESOURCE_VIEW)
+        @self.route(
+            '/sessions',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
         async def get_sessions(request_context: RequestContext) -> str:
             """Get session information"""
             # Parse query parameters
@@ -221,8 +250,9 @@ class MonitoringRouterGroup(group.RouterGroup):
             end_time_str = quart.request.args.get('endTime')
             user_query = quart.request.args.get('userQuery')
             is_active_str = quart.request.args.get('isActive')
-            limit = int(quart.request.args.get('limit', 100))
-            offset = int(quart.request.args.get('offset', 0))
+            limit, offset = self.ap.monitoring_service.normalize_page_window(
+                quart.request.args.get('limit', 100), quart.request.args.get('offset', 0)
+            )
 
             # Parse datetime
             start_time = parse_iso_datetime(start_time_str)
@@ -254,7 +284,12 @@ class MonitoringRouterGroup(group.RouterGroup):
                 }
             )
 
-        @self.route('/errors', methods=['GET'], permission=Permission.RESOURCE_VIEW)
+        @self.route(
+            '/errors',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
         async def get_errors(request_context: RequestContext) -> str:
             """Get error logs"""
             # Parse query parameters
@@ -262,8 +297,9 @@ class MonitoringRouterGroup(group.RouterGroup):
             pipeline_ids = quart.request.args.getlist('pipelineId')
             start_time_str = quart.request.args.get('startTime')
             end_time_str = quart.request.args.get('endTime')
-            limit = int(quart.request.args.get('limit', 100))
-            offset = int(quart.request.args.get('offset', 0))
+            limit, offset = self.ap.monitoring_service.normalize_page_window(
+                quart.request.args.get('limit', 100), quart.request.args.get('offset', 0)
+            )
 
             # Parse datetime
             start_time = parse_iso_datetime(start_time_str)
@@ -404,7 +440,12 @@ class MonitoringRouterGroup(group.RouterGroup):
                 }
             )
 
-        @self.route('/sessions/<session_id>/analysis', methods=['GET'], permission=Permission.RESOURCE_VIEW)
+        @self.route(
+            '/sessions/<session_id>/analysis',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
         async def get_session_analysis(session_id: str, request_context: RequestContext) -> str:
             """Get detailed analysis for a specific session"""
             start_time = parse_iso_datetime(quart.request.args.get('startTime'))
@@ -421,13 +462,18 @@ class MonitoringRouterGroup(group.RouterGroup):
             # The frontend will handle the 'found: false' case
             return self.success(data=analysis)
 
-        @self.route('/messages/<message_id>/details', methods=['GET'], permission=Permission.RESOURCE_VIEW)
+        @self.route(
+            '/messages/<message_id>/details',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
         async def get_message_details(message_id: str, request_context: RequestContext) -> str:
             """Get detailed information for a specific message"""
             details = await self.ap.monitoring_service.get_message_details(request_context, message_id)
 
             if not details.get('found'):
-                return self.error(message=f'Message {message_id} not found', code=404)
+                return self.http_status(404, 'resource_not_found', 'Message not found')
 
             return self.success(data=details)
 

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

@@ -5,6 +5,7 @@ import quart
 from ....authz import Permission, has_permission
 from ....context import RequestContext
 from ....service.secrets import redact_secrets
+from ....service.pipeline_run import run_pipeline
 from ... import group
 from ......pipeline.extension_preferences import (
     normalize_extension_preferences,
@@ -198,3 +199,18 @@ class PipelinesRouterGroup(group.RouterGroup):
             except ValueError as exc:
                 return self.http_status(400, -1, str(exc))
             return self.success()
+
+        @self.route(
+            '/<pipeline_uuid>/run',
+            methods=['POST'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RUNTIME_OPERATE,
+        )
+        async def run(pipeline_uuid: str, request_context: RequestContext):
+            body = await quart.request.get_json()
+            if not isinstance(body, dict) or set(body) != {'message'}:
+                return self.http_status(400, -1, 'Expected an object containing message')
+            text = body['message']
+            if not isinstance(text, str) or not text.strip() or len(text) > 100_000:
+                return self.http_status(400, -1, 'message must contain 1..100000 characters')
+            return self.success(data=await run_pipeline(self.ap, request_context, pipeline_uuid, text))

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

@@ -13,6 +13,24 @@ from .....workspace.invitation_delivery import InvitationDeliveryService
 
 
 SYSTEM_CAPABILITY_OPERATIONS = (
+    'knowledge_engine.list',
+    'knowledge_engine.creation_schema',
+    'knowledge_engine.retrieval_schema',
+    'knowledge_parser.list',
+    'pipeline.extensions.get',
+    'pipeline.extensions.update',
+    'pipeline.run',
+    'monitoring.messages',
+    'monitoring.llm_calls',
+    'monitoring.tool_calls',
+    'monitoring.embedding_calls',
+    'monitoring.sessions',
+    'monitoring.errors',
+    'monitoring.message_details',
+    'monitoring.session_analysis',
+    'sandbox.status',
+    'sandbox.sessions',
+    'sandbox.errors',
     'bot.list',
     'bot.get',
     'bot.create',

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

@@ -434,58 +434,34 @@ class KnowledgeService:
     # ================= Knowledge Engine Discovery =================
 
     async def list_knowledge_engines(self, context: TenantContext) -> list[dict]:
-        """List all available Knowledge Engines from plugins."""
+        """List engines; unavailable runtimes must not look like empty catalogs."""
         require_workspace_uuid(context)
-        engines = []
-
         if not self.ap.plugin_connector.is_enable_plugin:
-            return engines
+            return []
         await self.ap.plugin_connector.require_workspace_context(context)
-
-        # Get KnowledgeEngine plugins
-        try:
-            knowledge_engines = await self.ap.plugin_connector.list_knowledge_engines()
-            engines.extend(knowledge_engines)
-        except Exception as e:
-            self.ap.logger.warning(f'Failed to list Knowledge Engines from plugins: {e}')
-
-        return engines
+        return await self.ap.plugin_connector.list_knowledge_engines()
 
     async def list_parsers(self, context: TenantContext, mime_type: str | None = None) -> list[dict]:
-        """List available parsers, optionally filtered by MIME type."""
+        """List parsers, optionally filtered by MIME type."""
         require_workspace_uuid(context)
         if not self.ap.plugin_connector.is_enable_plugin:
             return []
         await self.ap.plugin_connector.require_workspace_context(context)
-        try:
-            parsers = await self.ap.plugin_connector.list_parsers()
-            if mime_type:
-                parsers = [p for p in parsers if mime_type in p.get('supported_mime_types', [])]
-            return parsers
-        except Exception as e:
-            self.ap.logger.warning(f'Failed to list parsers: {e}')
-            return []
+        parsers = await self.ap.plugin_connector.list_parsers()
+        return [p for p in parsers if mime_type in p.get('supported_mime_types', [])] if mime_type else parsers
+
+    async def _require_knowledge_engine(self, context: TenantContext, plugin_id: str) -> None:
+        engines = await self.list_knowledge_engines(context)
+        if not any(engine.get('plugin_id') == plugin_id for engine in engines):
+            raise WorkspaceNotFoundError('Knowledge engine not found')
+        await self.ap.plugin_connector.require_workspace_context(context)
 
     async def get_engine_creation_schema(self, context: TenantContext, plugin_id: str) -> dict:
-        """Get creation settings schema for a specific Knowledge Engine."""
-        require_workspace_uuid(context)
-        if not self.ap.plugin_connector.is_enable_plugin:
-            return {}
-        await self.ap.plugin_connector.require_workspace_context(context)
-        try:
-            return await self.ap.plugin_connector.get_rag_creation_schema(plugin_id)
-        except Exception as e:
-            self.ap.logger.warning(f'Failed to get creation schema for {plugin_id}: {e}')
-            return {}
+        """Get an existing engine's creation schema in this Workspace."""
+        await self._require_knowledge_engine(context, plugin_id)
+        return await self.ap.plugin_connector.get_rag_creation_schema(plugin_id)
 
     async def get_engine_retrieval_schema(self, context: TenantContext, plugin_id: str) -> dict:
-        """Get retrieval settings schema for a specific Knowledge Engine."""
-        require_workspace_uuid(context)
-        if not self.ap.plugin_connector.is_enable_plugin:
-            return {}
-        await self.ap.plugin_connector.require_workspace_context(context)
-        try:
-            return await self.ap.plugin_connector.get_rag_retrieval_schema(plugin_id)
-        except Exception as e:
-            self.ap.logger.warning(f'Failed to get retrieval schema for {plugin_id}: {e}')
-            return {}
+        """Get an existing engine's retrieval schema in this Workspace."""
+        await self._require_knowledge_engine(context, plugin_id)
+        return await self.ap.plugin_connector.get_rag_retrieval_schema(plugin_id)

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

@@ -332,6 +332,28 @@ class PipelineService:
 
         return new_uuid
 
+    async def get_pipeline_extensions(self, context: TenantContext, pipeline_uuid: str) -> dict:
+        pipeline = await self.get_pipeline(context, pipeline_uuid)
+        if pipeline is None:
+            raise WorkspaceNotFoundError('Pipeline not found')
+        if self.ap.plugin_connector.is_enable_plugin:
+            await self.ap.plugin_connector.require_workspace_context(context)
+        plugins = await self.ap.plugin_connector.list_plugins(component_kinds=['Command', 'EventListener', 'Tool'])
+        prefs = pipeline.get('extensions_preferences', {})
+        return {
+            'enable_all_plugins': prefs.get('enable_all_plugins', True),
+            'enable_all_mcp_servers': prefs.get('enable_all_mcp_servers', True),
+            'enable_all_skills': prefs.get('enable_all_skills', True),
+            'bound_plugins': prefs.get('plugins', []),
+            'available_plugins': redact_secrets(plugins),
+            'bound_mcp_servers': prefs.get('mcp_servers', []),
+            'available_mcp_servers': await self.ap.mcp_service.get_mcp_servers(context, contain_runtime_info=True),
+            'bound_mcp_resources': prefs.get('mcp_resources', []),
+            'mcp_resource_agent_read_enabled': prefs.get('mcp_resource_agent_read_enabled', True),
+            'bound_skills': prefs.get('skills', []),
+            'available_skills': await self.ap.skill_service.list_skills(context),
+        }
+
     async def update_pipeline_extensions(
         self,
         context: TenantContext,

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

@@ -0,0 +1,68 @@
+"""Single-turn diagnostic execution through the normal Pipeline scheduler."""
+
+from __future__ import annotations
+
+import asyncio
+import time
+import uuid
+
+from langbot_plugin.api.entities.builtin.platform import entities, events, message
+from langbot_plugin.api.entities.builtin.provider.session import LauncherTypes
+
+from ..authz import Permission, require_permission
+from ..context import ExecutionContext, RequestContext
+from ....workspace.errors import WorkspaceNotFoundError
+
+
+async def run_pipeline(ap, context: RequestContext, pipeline_uuid: str, text: str, *, timeout: float = 60) -> dict:
+    """Run in a fresh session; a timed-out request is never submitted again."""
+    require_permission(context, Permission.RUNTIME_OPERATE)
+    if not isinstance(text, str) or not text.strip() or len(text) > 100_000:
+        raise ValueError('message must contain 1..100000 characters')
+    if await ap.pipeline_service.get_pipeline(context, pipeline_uuid) is None:
+        raise WorkspaceNotFoundError('Pipeline not found')
+
+    bot = await ap.platform_mgr.get_websocket_proxy_bot(context)
+    adapter = bot.adapter
+    session_id = str(uuid.uuid4())
+    launcher_id = f'websocket_{pipeline_uuid}:{session_id}'
+    chain = message.MessageChain([message.Plain(text=text)])
+    event = events.FriendMessage(
+        sender=entities.Friend(id=launcher_id, nickname='CLI', remark='CLI'),
+        message_chain=chain,
+        time=time.time(),
+    )
+    execution = ExecutionContext.from_request(context, bot_uuid='websocket-proxy-bot', pipeline_uuid=pipeline_uuid)
+    query = await ap.query_pool.add_query(
+        bot_uuid='websocket-proxy-bot',
+        launcher_type=LauncherTypes.PERSON,
+        launcher_id=launcher_id,
+        sender_id=launcher_id,
+        message_event=event,
+        message_chain=chain,
+        adapter=adapter,
+        pipeline_uuid=pipeline_uuid,
+        variables={'_cli_run_status': 'failed'},
+        execution_context=execution,
+    )
+    completed = asyncio.Event()
+    # add_query returns without yielding after enqueueing, so the scheduler
+    # cannot remove this query before its completion notification is attached.
+    object.__setattr__(query, '_completion_event', completed)
+    data = {'pipeline_uuid': pipeline_uuid, 'session_id': f'person_{launcher_id}', 'query_id': query.query_uuid}
+    try:
+        await asyncio.wait_for(completed.wait(), timeout=timeout)
+    except asyncio.TimeoutError:
+        return {**data, 'status': 'unknown', 'error': 'Execution status unknown; do not automatically retry.'}
+
+    replies = adapter.get_websocket_messages(pipeline_uuid, 'person', session_id)
+    replies = [reply for reply in replies if reply.get('role') == 'assistant']
+    data.update(
+        status=query.variables['_cli_run_status'],
+        replies=replies,
+        reply='\n'.join(reply.get('content', '') for reply in replies),
+        message_id=query.variables.get('_monitoring_message_id'),
+    )
+    if data['status'] != 'completed':
+        data['error'] = 'Pipeline failed or was dropped; inspect the message and call records.'
+    return data

src/langbot/pkg/api/mcp/server.py

@@ -96,6 +96,119 @@ class LangBotMCPServer:
             }
             return _dump(data)
 
+        @mcp.tool(description='List knowledge engines and their capabilities.')
+        async def list_knowledge_engines() -> str:
+            return _dump(await ap.knowledge_service.list_knowledge_engines(_authorized(Permission.RESOURCE_VIEW)))
+
+        @mcp.tool(description='Get creation or retrieval schema for an author/name knowledge engine.')
+        async def get_knowledge_engine_schema(plugin_id: str, kind: typing.Literal['creation', 'retrieval']) -> str:
+            context = _authorized(Permission.RESOURCE_VIEW)
+            if kind == 'creation':
+                return _dump(await ap.knowledge_service.get_engine_creation_schema(context, plugin_id))
+            return _dump(await ap.knowledge_service.get_engine_retrieval_schema(context, plugin_id))
+
+        @mcp.tool(description='List document parsers, optionally filtered by MIME type.')
+        async def list_knowledge_parsers(mime_type: str | None = None) -> str:
+            return _dump(await ap.knowledge_service.list_parsers(_authorized(Permission.RESOURCE_VIEW), mime_type))
+
+        @mcp.tool(description='Read Pipeline extension bindings, enable-all switches and available extensions.')
+        async def get_pipeline_extensions(pipeline_uuid: str) -> str:
+            return _dump(
+                await ap.pipeline_service.get_pipeline_extensions(_authorized(Permission.RESOURCE_VIEW), pipeline_uuid)
+            )
+
+        @mcp.tool(description='Replace all Pipeline extension bindings. All binding lists and switches are required.')
+        async def update_pipeline_extensions(
+            pipeline_uuid: str,
+            bound_plugins: list[dict],
+            bound_mcp_servers: list[str],
+            bound_skills: list[str],
+            bound_mcp_resources: list[dict],
+            enable_all_plugins: bool,
+            enable_all_mcp_servers: bool,
+            enable_all_skills: bool,
+            mcp_resource_agent_read_enabled: bool,
+        ) -> str:
+            context = _authorized(Permission.RESOURCE_MANAGE)
+            require_permission(context, Permission.RESOURCE_VIEW)
+            await ap.pipeline_service.update_pipeline_extensions(
+                context,
+                pipeline_uuid,
+                bound_plugins,
+                bound_mcp_servers,
+                enable_all_plugins,
+                enable_all_mcp_servers,
+                bound_skills,
+                enable_all_skills,
+                bound_mcp_resources,
+                mcp_resource_agent_read_enabled,
+            )
+            return _dump(await ap.pipeline_service.get_pipeline_extensions(context, pipeline_uuid))
+
+        @mcp.tool(
+            description='Run one Pipeline turn in a fresh session. Calls configured models/tools. Never retry an unknown outcome automatically.'
+        )
+        async def run_pipeline(pipeline_uuid: str, message: str) -> str:
+            from ..http.service.pipeline_run import run_pipeline as execute
+
+            return _dump(await execute(ap, _authorized(Permission.RUNTIME_OPERATE), pipeline_uuid, message))
+
+        @mcp.tool(
+            description='Read Workspace sandbox status, sessions or recent errors. Does not create execution sessions.'
+        )
+        async def get_sandbox_diagnostics(kind: typing.Literal['status', 'sessions', 'errors'] = 'status') -> str:
+            context = _authorized(Permission.RESOURCE_VIEW if kind == 'status' else Permission.AUDIT_VIEW)
+            if kind == 'status':
+                return _dump(await ap.box_service.get_status(context))
+            if kind == 'sessions':
+                return _dump(await ap.box_service.get_sessions(context))
+            if ap.box_service.managed_admission_required:
+                await ap.box_service.require_workspace_sandbox(context)
+            return _dump(ap.box_service.get_recent_errors(context))
+
+        @mcp.tool(
+            description='Read bounded Workspace runtime records. Filters follow the monitoring service: bot_ids, pipeline_ids, session_ids, start_time, end_time, knowledge_base_id, user_query, is_active as supported by the record kind.'
+        )
+        async def get_monitoring_records(
+            kind: typing.Literal['messages', 'llm_calls', 'tool_calls', 'embedding_calls', 'sessions', 'errors'],
+            limit: int = 100,
+            offset: int = 0,
+            filters: dict | None = None,
+        ) -> str:
+            import datetime
+
+            context = _authorized(Permission.RESOURCE_VIEW)
+            allowed = {'start_time', 'end_time'}
+            if kind != 'embedding_calls':
+                allowed |= {'bot_ids', 'pipeline_ids'}
+            if kind in {'messages', 'tool_calls'}:
+                allowed.add('session_ids')
+            if kind == 'embedding_calls':
+                allowed.add('knowledge_base_id')
+            if kind == 'sessions':
+                allowed |= {'user_query', 'is_active'}
+            kwargs = dict(filters or {})
+            if set(kwargs) - allowed:
+                raise ValueError('Unsupported monitoring filter')
+            for field in ('start_time', 'end_time'):
+                if kwargs.get(field):
+                    value = datetime.datetime.fromisoformat(kwargs[field].replace('Z', '+00:00'))
+                    if value.tzinfo is not None:
+                        value = value.astimezone(datetime.timezone.utc).replace(tzinfo=None)
+                    kwargs[field] = value
+            limit, offset = ap.monitoring_service.normalize_page_window(limit, offset)
+            rows, total = await getattr(ap.monitoring_service, 'get_' + kind)(
+                context, limit=limit, offset=offset, **kwargs
+            )
+            return _dump({kind: rows, 'total': total, 'limit': limit, 'offset': offset})
+
+        @mcp.tool(description='Read message details or session analysis within the authenticated Workspace.')
+        async def get_monitoring_details(kind: typing.Literal['message', 'session'], identifier: str) -> str:
+            context = _authorized(Permission.RESOURCE_VIEW)
+            if kind == 'message':
+                return _dump(await ap.monitoring_service.get_message_details(context, identifier))
+            return _dump(await ap.monitoring_service.get_session_analysis(context, identifier))
+
         # ----- Bots ---------------------------------------------------- #
         @mcp.tool(description='List all messaging-platform bots. Secrets are redacted.')
         async def list_bots() -> str:

src/langbot/pkg/box/service.py

@@ -1373,13 +1373,14 @@ class BoxService:
             self._runtime_connector.dispose()
 
     async def get_sessions(self, context: TenantContext) -> list[dict]:
-        if not self._available:
+        if not self._enabled:
+            if self.managed_admission_required:
+                await self.require_workspace_sandbox(context)
             return []
         execution_context = await self.require_workspace_sandbox(context)
-        try:
-            return await self.client.get_sessions(action_context=self._action_context(execution_context))
-        except Exception:
-            return []
+        if not self._available:
+            raise RuntimeError('Sandbox is unavailable; inspect sandbox status')
+        return await self.client.get_sessions(action_context=self._action_context(execution_context))
 
     async def get_storage_analysis(self, context: TenantContext) -> dict:
         """Return Workspace-scoped storage measured by the Box Runtime."""

src/langbot/pkg/pipeline/pipelinemgr.py

@@ -444,6 +444,10 @@ class RuntimePipeline:
             self.ap.logger.debug(f'Processing query {query.query_id}')
 
             await self._execute_from_stage(0, query)
+            if '_cli_run_status' in query.variables:
+                query.variables['_cli_run_status'] = (
+                    'failed' if query.variables.get('_monitoring_has_error') else 'completed'
+                )
 
             # Record query success only if no error occurred during processing
             if not query.variables.get('_monitoring_has_error', False):

src/langbot/pkg/pipeline/pool.py

@@ -173,6 +173,9 @@ class QueryPool:
             else:
                 self.active_query_count_by_workspace.pop(query_workspace_uuid, None)
             plugin_diagnostics.discard_query_state(query)
+            completion = getattr(query, '_completion_event', None)
+            if completion is not None:
+                completion.set()
             counter_key = (
                 execution_context.instance_uuid,
                 execution_context.workspace_uuid,
@@ -426,6 +429,9 @@ class QueryPool:
                 self.queries.pop(index)
                 break
         plugin_diagnostics.discard_query_state(query)
+        completion = getattr(query, '_completion_event', None)
+        if completion is not None:
+            completion.set()
         return True
 
     async def __aenter__(self):

tests/unit_tests/api/service/test_cli_diagnostics.py

@@ -0,0 +1,384 @@
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+import quart
+
+from langbot.pkg.api.http.context import (
+    ExecutionContext,
+    PrincipalContext,
+    PrincipalType,
+    RequestContext,
+    WorkspaceContext,
+)
+from langbot.pkg.api.http.controller.groups.box import BoxRouterGroup
+from langbot.pkg.api.http.controller.groups.monitoring import MonitoringRouterGroup
+from langbot.pkg.api.http.controller.groups.pipelines.pipelines import PipelinesRouterGroup
+from langbot.pkg.api.http.service.pipeline import PipelineService
+from langbot.pkg.api.http.service.pipeline_run import run_pipeline
+from langbot.pkg.pipeline.pool import QueryPool
+from langbot.pkg.pipeline.pipelinemgr import RuntimePipeline, StageInstContainer
+from langbot.pkg.pipeline.entities import StageProcessResult, ResultType
+from langbot.pkg.platform.sources.websocket_adapter import WebSocketAdapter, WebSocketSession
+from langbot.pkg.cloud.entitlements import EntitlementUnavailableError
+from .test_monitoring_tenancy import service as service, WORKSPACE_A, WORKSPACE_B, _context, _record_message
+
+pytestmark = pytest.mark.asyncio
+
+
+def request_context(workspace=WORKSPACE_A, permissions=('resource.view', 'runtime.operate', 'audit.view')):
+    return RequestContext(
+        instance_uuid='instance',
+        placement_generation=3,
+        request_id='request',
+        auth_type='api_key',
+        principal=PrincipalContext(PrincipalType.API_KEY, api_key_uuid=f'key-{workspace}'),
+        workspace=WorkspaceContext(workspace, None, None, frozenset(permissions)),
+    )
+
+
+async def diagnostic_client(ap):
+    async def authenticate(key):
+        if key not in {'a', 'b', 'viewer', 'none'}:
+            raise ValueError('Invalid API key')
+        ctx = request_context(
+            WORKSPACE_B if key == 'b' else WORKSPACE_A,
+            ()
+            if key == 'none'
+            else ('resource.view',)
+            if key == 'viewer'
+            else ('resource.view', 'runtime.operate', 'resource.manage', 'audit.view'),
+        )
+        return SimpleNamespace(
+            instance_uuid=ctx.instance_uuid,
+            workspace_uuid=ctx.workspace_uuid,
+            placement_generation=ctx.placement_generation,
+            api_key_uuid=ctx.principal.api_key_uuid,
+            permissions=ctx.workspace.permissions,
+        )
+
+    ap.apikey_service = SimpleNamespace(authenticate_api_key=authenticate)
+    app = quart.Quart(__name__)
+    for cls in (MonitoringRouterGroup, BoxRouterGroup, PipelinesRouterGroup):
+        await cls(ap, app).initialize()
+    return app.test_client()
+
+
+async def test_http_api_key_diagnostics_are_scoped_bounded_and_permission_checked(service):
+    ap = service.ap
+    ap.logger = Mock()
+    ap.monitoring_service = service
+    ap.box_service = SimpleNamespace(
+        get_status=AsyncMock(return_value={'enabled': False, 'available': False}),
+        get_sessions=AsyncMock(return_value=[]),
+        get_recent_errors=Mock(return_value=[]),
+        managed_admission_required=False,
+    )
+    client = await diagnostic_client(ap)
+    message_a = await _record_message(service, _context(WORKSPACE_A), 'only-a')
+    await _record_message(service, _context(WORKSPACE_B), 'only-b')
+    for key, expected, excluded in [('a', 'only-a', 'only-b'), ('b', 'only-b', 'only-a')]:
+        response = await client.get(
+            '/api/v1/monitoring/messages?limit=1', headers={'X-API-Key': key, 'X-Workspace-Id': WORKSPACE_B}
+        )
+        assert response.status_code == 200
+        text = await response.get_data(as_text=True)
+        assert expected in text and excluded not in text
+    response = await client.get(f'/api/v1/monitoring/messages/{message_a}/details', headers={'X-API-Key': 'b'})
+    assert response.status_code == 404
+    assert (await client.get('/api/v1/monitoring/messages', headers={'X-API-Key': 'none'})).status_code == 403
+    assert (await client.get('/api/v1/monitoring/messages')).status_code == 401
+    for route in ('llm-calls', 'tool-calls', 'embedding-calls', 'sessions', 'errors'):
+        assert (await client.get('/api/v1/monitoring/' + route, headers={'X-API-Key': 'a'})).status_code == 200
+    for route in ('sessions', 'errors'):
+        assert (await client.get('/api/v1/box/' + route, headers={'X-API-Key': 'viewer'})).status_code == 403
+        response = await client.get('/api/v1/box/' + route, headers={'X-API-Key': 'a'})
+        assert response.status_code == 200 and (await response.get_json())['data'] == []
+    assert (await client.get('/api/v1/box/runtime-status', headers={'X-API-Key': 'a'})).status_code == 401
+    response = await client.get('/api/v1/box/status', headers={'X-API-Key': 'viewer'})
+    assert (await response.get_json())['data']['enabled'] is False
+    ap.box_service.get_status.side_effect = EntitlementUnavailableError('Sandbox unavailable')
+    response = await client.get('/api/v1/box/status', headers={'X-API-Key': 'a'})
+    assert response.status_code == 403
+    assert (await response.get_json())['code'] == 'managed_sandbox_unavailable'
+
+
+def runtime_app(service):
+    ap = service.ap
+    ap.logger = Mock()
+    ap.monitoring_service = service
+    ap.query_pool = QueryPool()
+    ap.workspace_service = SimpleNamespace(
+        get_execution_binding=AsyncMock(return_value=SimpleNamespace(instance_uuid='instance'))
+    )
+    ap.plugin_connector = SimpleNamespace(
+        emit_event=AsyncMock(side_effect=lambda event_obj, *_: SimpleNamespace(
+            event=event_obj, is_prevented_default=lambda: False
+        ))
+    )
+    ap.bot_service = SimpleNamespace(get_bot=AsyncMock(return_value={'name': 'CLI'}))
+    ap.pipeline_service = SimpleNamespace(get_pipeline=AsyncMock(return_value={'uuid': 'p'}))
+    adapters = {}
+
+    async def proxy(context):
+        if context.workspace_uuid not in adapters:
+            logger = Mock(execution_context=ExecutionContext.from_request(context))
+            adapters[context.workspace_uuid] = WebSocketAdapter.model_construct(ap=ap, logger=logger)
+            adapters[context.workspace_uuid].websocket_person_session = WebSocketSession(id='person')
+            adapters[context.workspace_uuid].websocket_group_session = WebSocketSession(id='group')
+        return SimpleNamespace(adapter=adapters[context.workspace_uuid])
+
+    ap.platform_mgr = SimpleNamespace(get_websocket_proxy_bot=proxy)
+    return ap
+
+
+async def process_next(ap, *, fail=False):
+    while not ap.query_pool.queries:
+        await asyncio.sleep(0)
+    query = ap.query_pool.queries[0]
+    async with ap.query_pool:
+        ap.query_pool.mark_query_running_locked(query)
+
+    class ReplyStage:
+        async def process(self, query, _name):
+            return StageProcessResult(
+                result_type=ResultType.CONTINUE,
+                new_query=query,
+                user_notice='reply:' + str(query.message_chain),
+                error_notice='model unavailable' if fail else '',
+            )
+
+    pipeline = RuntimePipeline(
+        ap,
+        SimpleNamespace(
+            uuid='p',
+            workspace_uuid=query.workspace_uuid,
+            name='Test',
+            extensions_preferences={},
+            config={'output': {'misc': {'at-sender': False, 'quote-origin': False}}},
+        ),
+        [StageInstContainer('reply', ReplyStage())],
+        ExecutionContext.from_request(request_context(query.workspace_uuid), pipeline_uuid='p'),
+    )
+    await pipeline.run(query)
+    return query
+
+
+async def test_run_route_executes_pipeline_and_isolates_sessions_and_workspaces(service):
+    ap = runtime_app(service)
+    client = await diagnostic_client(ap)
+    assert (
+        await client.post('/api/v1/pipelines/p/run', headers={'X-API-Key': 'viewer'}, json={'message': 'x'})
+    ).status_code == 403
+    sessions = set()
+    for key, text in [('a', 'first'), ('a', 'second'), ('b', 'other-workspace')]:
+        task = asyncio.create_task(process_next(ap))
+        response = await asyncio.wait_for(
+            client.post('/api/v1/pipelines/p/run', headers={'X-API-Key': key}, json={'message': text}), 3
+        )
+        if response.status_code != 200:
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+            pytest.fail(await response.get_data(as_text=True))
+        query = await task
+        assert response.status_code == 200
+        data = (await response.get_json())['data']
+        assert data['status'] == 'completed'
+        assert data['reply'] == 'reply:' + text
+        assert data['session_id'] not in sessions
+        sessions.add(data['session_id'])
+        assert query.workspace_uuid == (WORKSPACE_B if key == 'b' else WORKSPACE_A)
+        assert data['message_id']
+    a_records, _ = await service.get_messages(_context(WORKSPACE_A))
+    assert 'other-workspace' not in str(a_records)
+
+
+async def test_run_failure_timeout_and_missing_pipeline(service):
+    ap = runtime_app(service)
+    task = asyncio.create_task(process_next(ap, fail=True))
+    failed = await run_pipeline(ap, request_context(), 'p', 'fail')
+    await task
+    assert failed['status'] == 'failed' and failed['message_id']
+    errors, total = await service.get_errors(_context(WORKSPACE_A))
+    assert total == 1 and errors[0]['error_message'] == 'model unavailable'
+    unknown = await run_pipeline(ap, request_context(), 'p', 'slow', timeout=0.001)
+    assert unknown['status'] == 'unknown' and len(ap.query_pool.queries) == 1
+    await process_next(ap)
+    assert not ap.query_pool.cached_queries
+    ap.pipeline_service.get_pipeline.return_value = None
+    from langbot.pkg.workspace.errors import WorkspaceNotFoundError
+
+    with pytest.raises(WorkspaceNotFoundError):
+        await run_pipeline(ap, request_context(), 'missing', 'x')
+    assert not ap.query_pool.queries
+
+
+async def test_extension_discovery_preserves_empty_bindings_and_redacts_plugins():
+    ap = SimpleNamespace(
+        plugin_connector=SimpleNamespace(
+            is_enable_plugin=True,
+            require_workspace_context=AsyncMock(),
+            list_plugins=AsyncMock(return_value=[{'api_key': 'private'}]),
+        ),
+        mcp_service=SimpleNamespace(get_mcp_servers=AsyncMock(return_value=[])),
+        skill_service=SimpleNamespace(list_skills=AsyncMock(return_value=[])),
+    )
+    service = PipelineService(ap)
+    service.get_pipeline = AsyncMock(
+        return_value={'extensions_preferences': {'enable_all_plugins': False, 'plugins': []}}
+    )
+    data = await service.get_pipeline_extensions(request_context(), 'p')
+    assert data['bound_plugins'] == [] and data['enable_all_plugins'] is False
+    assert data['available_plugins'] == [{'api_key': '***'}]
+    ap.plugin_connector.require_workspace_context.assert_awaited_once_with(request_context())
+
+
+async def test_compiled_cli_against_core_http(service, tmp_path):
+    """Optional cross-repo check: real HTTP/SQLite/Pipeline, synthetic runtime providers."""
+    import json
+    import os
+    import socket
+    from pathlib import Path
+    from hypercorn.asyncio import serve
+    from hypercorn.config import Config
+    import sqlalchemy
+    from langbot.pkg.api.http.controller.groups.system import SystemRouterGroup
+    from langbot.pkg.api.http.controller.groups.knowledge.engines import KnowledgeEnginesRouterGroup
+    from langbot.pkg.api.http.controller.groups.knowledge.parsers import ParsersRouterGroup
+    from langbot.pkg.api.http.service.knowledge import KnowledgeService
+    from langbot.pkg.entity.persistence.pipeline import LegacyPipeline
+
+    binary = os.environ.get('LANGBOT_CLI_BIN')
+    if not binary:
+        pytest.skip('Set LANGBOT_CLI_BIN to a compiled lbctl for the cross-repo HTTP check')
+    assert Path(binary).is_file()
+    ap = runtime_app(service)
+    ap.pipeline_service = PipelineService(ap)
+    ap.pipeline_mgr = SimpleNamespace(remove_pipeline=AsyncMock(), load_pipeline=AsyncMock())
+    await ap.persistence_mgr.execute_async(
+        sqlalchemy.insert(LegacyPipeline).values(
+            uuid='p',
+            workspace_uuid=WORKSPACE_A,
+            name='CLI test',
+            description='',
+            for_version='1',
+            stages=[],
+            config={},
+        )
+    )
+    ap.plugin_connector.is_enable_plugin = True
+    ap.plugin_connector.require_workspace_context = AsyncMock()
+    ap.plugin_connector.list_plugins = AsyncMock(return_value=[])
+    ap.plugin_connector.list_knowledge_engines = AsyncMock(
+        return_value=[{'plugin_id': 'author/engine', 'capabilities': ['doc_ingestion']}]
+    )
+    ap.plugin_connector.get_rag_creation_schema = AsyncMock(return_value={'type': 'object'})
+    ap.plugin_connector.get_rag_retrieval_schema = AsyncMock(return_value={'type': 'object'})
+    ap.plugin_connector.list_parsers = AsyncMock(return_value=[{'id': 'text', 'supported_mime_types': ['text/plain']}])
+    ap.knowledge_service = KnowledgeService(ap)
+    ap.mcp_service = SimpleNamespace(get_mcp_servers=AsyncMock(return_value=[]))
+    ap.skill_service = SimpleNamespace(list_skills=AsyncMock(return_value=[]))
+    ap.box_service = SimpleNamespace(
+        get_status=AsyncMock(return_value={'enabled': False, 'available': False}),
+        get_sessions=AsyncMock(return_value=[]),
+        get_recent_errors=Mock(return_value=[]),
+        managed_admission_required=False,
+    )
+    client = await diagnostic_client(ap)
+    original_authenticate = ap.apikey_service.authenticate_api_key
+
+    async def smoke_authenticate(key):
+        return await original_authenticate('a' if key == 'lbk_synthetic_smoke_key_7f36c0' else 'invalid')
+
+    ap.apikey_service.authenticate_api_key = smoke_authenticate
+    for cls in (SystemRouterGroup, KnowledgeEnginesRouterGroup, ParsersRouterGroup):
+        await cls(ap, client.app).initialize()
+    listener = socket.socket()
+    listener.bind(('127.0.0.1', 0))
+    port = listener.getsockname()[1]
+    config = Config()
+    config.bind = [f'fd://{listener.detach()}']
+    config.accesslog = None
+    shutdown = asyncio.Event()
+    server = asyncio.create_task(serve(client.app, config, shutdown_trigger=shutdown.wait))
+    cli_config = tmp_path / 'cli.yaml'
+    env = {k: v for k, v in os.environ.items() if not k.startswith('LANGBOT_')}
+    env['SMOKE_KEY'] = 'lbk_synthetic_smoke_key_7f36c0'
+
+    async def cli(*args, code=0):
+        process = await asyncio.create_subprocess_exec(
+            binary,
+            '--config',
+            str(cli_config),
+            '-o',
+            'json',
+            *args,
+            env=env,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        out, err = await asyncio.wait_for(process.communicate(), 5)
+        assert process.returncode == code, (out.decode(), err.decode())
+        return json.loads(out)
+
+    try:
+        async with asyncio.timeout(3):
+            while True:
+                try:
+                    _, writer = await asyncio.open_connection('127.0.0.1', port)
+                    writer.close()
+                    await writer.wait_closed()
+                    break
+                except OSError:
+                    await asyncio.sleep(0.01)
+        await cli(
+            'context',
+            'add',
+            'smoke',
+            '--endpoint',
+            f'http://127.0.0.1:{port}',
+            '--api-key-env',
+            'SMOKE_KEY',
+            '--expect-workspace',
+            WORKSPACE_A,
+        )
+        await cli('context', 'use', 'smoke')
+        for args in [
+            ('knowledge-engine', 'list'),
+            ('knowledge-engine', 'creation-schema', 'author/engine'),
+            ('knowledge-engine', 'retrieval-schema', 'author/engine'),
+            ('knowledge-parser', 'list', '--mime-type', 'text/plain'),
+            ('sandbox', 'status'),
+            ('sandbox', 'sessions'),
+            ('sandbox', 'errors'),
+        ]:
+            assert (await cli(*args))['ok']
+        await cli('knowledge-engine', 'creation-schema', 'missing/engine', code=5)
+        body = {
+            'bound_plugins': [],
+            'bound_mcp_servers': [],
+            'bound_skills': [],
+            'bound_mcp_resources': [],
+            'enable_all_plugins': False,
+            'enable_all_mcp_servers': False,
+            'enable_all_skills': False,
+            'mcp_resource_agent_read_enabled': False,
+        }
+        body_path = tmp_path / 'extensions.json'
+        body_path.write_text(json.dumps(body))
+        updated = await cli('pipeline', 'extensions', 'update', 'p', '--file', str(body_path))
+        assert all(updated['data'][k] == v for k, v in body.items())
+        task = asyncio.create_task(process_next(ap))
+        result = await cli('pipeline', 'run', 'p', '--message', 'HTTP smoke')
+        await task
+        assert result['data']['status'] == 'completed' and result['data']['reply'] == 'reply:HTTP smoke'
+        await cli('monitoring', 'message', result['data']['message_id'])
+        await cli('monitoring', 'session', result['data']['session_id'])
+        for kind in ('messages', 'llm-calls', 'tool-calls', 'embedding-calls', 'sessions', 'errors'):
+            assert (await cli('monitoring', kind, '--limit', '1'))['ok']
+    finally:
+        shutdown.set()
+        await server

tests/unit_tests/api/service/test_knowledge_service.py

@@ -53,7 +53,7 @@ def _app():
             require_workspace_context=AsyncMock(side_effect=lambda context: context),
             get_rag_creation_schema=AsyncMock(return_value={}),
             get_rag_retrieval_schema=AsyncMock(return_value={}),
-            list_knowledge_engines=AsyncMock(return_value=[]),
+            list_knowledge_engines=AsyncMock(return_value=[{'plugin_id': 'author/engine'}]),
             list_parsers=AsyncMock(return_value=[]),
         ),
     )
@@ -338,14 +338,15 @@ async def test_schema_validation_refences_before_second_runtime_call():
 
 
 @pytest.mark.asyncio
-async def test_engine_schemas_are_context_gated_and_fail_soft_on_connector_error():
+async def test_engine_schemas_are_context_gated_and_report_connector_error():
     app = _app()
     app.plugin_connector.get_rag_creation_schema.return_value = {'schema': ['creation']}
     app.plugin_connector.get_rag_retrieval_schema.side_effect = RuntimeError('offline')
     service = KnowledgeService(app)
 
     assert await service.get_engine_creation_schema(CONTEXT, 'author/engine') == {'schema': ['creation']}
-    assert await service.get_engine_retrieval_schema(CONTEXT, 'author/engine') == {}
+    with pytest.raises(RuntimeError, match='offline'):
+        await service.get_engine_retrieval_schema(CONTEXT, 'author/engine')
     with pytest.raises(WorkspaceRequiredError):
         await service.get_engine_creation_schema(None, 'author/engine')
 
@@ -535,12 +536,12 @@ class TestListKnowledgeEngines:
         app.plugin_connector.list_knowledge_engines.assert_not_awaited()
 
     @pytest.mark.asyncio
-    async def test_returns_empty_on_exception(self):
+    async def test_reports_runtime_exception(self):
         app = _app()
         app.plugin_connector.list_knowledge_engines.side_effect = RuntimeError('Connection error')
 
-        assert await KnowledgeService(app).list_knowledge_engines(CONTEXT) == []
-        app.logger.warning.assert_called_once()
+        with pytest.raises(RuntimeError, match='Connection error'):
+            await KnowledgeService(app).list_knowledge_engines(CONTEXT)
 
 
 class TestListParsers:
@@ -603,17 +604,12 @@ class TestGetEngineSchemas:
         assert 'properties' in result
 
     @pytest.mark.asyncio
-    async def test_returns_empty_dict_on_exception(self):
+    async def test_reports_schema_runtime_exception(self):
         app = _app()
         app.plugin_connector.get_rag_creation_schema.side_effect = RuntimeError('Plugin error')
 
-        result = await KnowledgeService(app).get_engine_creation_schema(
-            CONTEXT,
-            'author/engine',
-        )
-
-        assert result == {}
-        app.logger.warning.assert_called_once()
+        with pytest.raises(RuntimeError, match='Plugin error'):
+            await KnowledgeService(app).get_engine_creation_schema(CONTEXT, 'author/engine')
 
 
 class TestKnowledgeBaseSecretViews:
@@ -652,3 +648,12 @@ class TestKnowledgeBaseSecretViews:
             )
 
         app.rag_mgr.create_knowledge_base.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_missing_engine_is_not_an_empty_schema():
+    app = _app()
+    app.plugin_connector.list_knowledge_engines.return_value = []
+    with pytest.raises(WorkspaceNotFoundError, match='Knowledge engine not found'):
+        await KnowledgeService(app).get_engine_creation_schema(CONTEXT, 'missing/engine')
+    app.plugin_connector.get_rag_creation_schema.assert_not_awaited()