From 07f3283abf9063acd29d5261af63a1be72696f29 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <60681390+huanghuoguoguo@users.noreply.github.com> Date: Sun, 27 Sep 2026 17:31:06 +0800 Subject: [PATCH] feat(api): enable CLI diagnostics on current master --- skills/skills/langbot-mcp-ops/SKILL.md | 5 + .../pkg/api/http/controller/groups/box.py | 6 +- .../api/http/controller/groups/monitoring.py | 88 +++- .../controller/groups/pipelines/pipelines.py | 16 + .../pkg/api/http/controller/groups/system.py | 18 + src/langbot/pkg/api/http/service/knowledge.py | 60 +-- src/langbot/pkg/api/http/service/pipeline.py | 22 + .../pkg/api/http/service/pipeline_run.py | 68 ++++ src/langbot/pkg/api/mcp/server.py | 113 ++++++ src/langbot/pkg/box/service.py | 11 +- src/langbot/pkg/pipeline/pipelinemgr.py | 4 + src/langbot/pkg/pipeline/pool.py | 6 + .../api/service/test_cli_diagnostics.py | 384 ++++++++++++++++++ .../api/service/test_knowledge_service.py | 33 +- 14 files changed, 749 insertions(+), 85 deletions(-) create mode 100644 src/langbot/pkg/api/http/service/pipeline_run.py create mode 100644 tests/unit_tests/api/service/test_cli_diagnostics.py diff --git a/skills/skills/langbot-mcp-ops/SKILL.md b/skills/skills/langbot-mcp-ops/SKILL.md index c793631b3..d96e56f2f 100644 --- a/skills/skills/langbot-mcp-ops/SKILL.md +++ b/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 diff --git a/src/langbot/pkg/api/http/controller/groups/box.py b/src/langbot/pkg/api/http/controller/groups/box.py index d63e9d7b5..99ba7241c 100644 --- a/src/langbot/pkg/api/http/controller/groups/box.py +++ b/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: diff --git a/src/langbot/pkg/api/http/controller/groups/monitoring.py b/src/langbot/pkg/api/http/controller/groups/monitoring.py index 854ac0ec0..01a901743 100644 --- a/src/langbot/pkg/api/http/controller/groups/monitoring.py +++ b/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//analysis', methods=['GET'], permission=Permission.RESOURCE_VIEW) + @self.route( + '/sessions//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//details', methods=['GET'], permission=Permission.RESOURCE_VIEW) + @self.route( + '/messages//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) diff --git a/src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py b/src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py index 6b8d0f61d..016616623 100644 --- a/src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py +++ b/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( + '//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)) diff --git a/src/langbot/pkg/api/http/controller/groups/system.py b/src/langbot/pkg/api/http/controller/groups/system.py index 04a7ef57b..9efed6c77 100644 --- a/src/langbot/pkg/api/http/controller/groups/system.py +++ b/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', diff --git a/src/langbot/pkg/api/http/service/knowledge.py b/src/langbot/pkg/api/http/service/knowledge.py index e8c7f1dbe..654bf2f17 100644 --- a/src/langbot/pkg/api/http/service/knowledge.py +++ b/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) diff --git a/src/langbot/pkg/api/http/service/pipeline.py b/src/langbot/pkg/api/http/service/pipeline.py index 1935e3a1f..54568433c 100644 --- a/src/langbot/pkg/api/http/service/pipeline.py +++ b/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, diff --git a/src/langbot/pkg/api/http/service/pipeline_run.py b/src/langbot/pkg/api/http/service/pipeline_run.py new file mode 100644 index 000000000..825cadcf5 --- /dev/null +++ b/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 diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py index c4ece1a9e..d07edb7a9 100644 --- a/src/langbot/pkg/api/mcp/server.py +++ b/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: diff --git a/src/langbot/pkg/box/service.py b/src/langbot/pkg/box/service.py index 25d0614c3..4f51b15aa 100644 --- a/src/langbot/pkg/box/service.py +++ b/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.""" diff --git a/src/langbot/pkg/pipeline/pipelinemgr.py b/src/langbot/pkg/pipeline/pipelinemgr.py index 45287a34c..e838d72fc 100644 --- a/src/langbot/pkg/pipeline/pipelinemgr.py +++ b/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): diff --git a/src/langbot/pkg/pipeline/pool.py b/src/langbot/pkg/pipeline/pool.py index 08971d79c..c2c05169b 100644 --- a/src/langbot/pkg/pipeline/pool.py +++ b/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): diff --git a/tests/unit_tests/api/service/test_cli_diagnostics.py b/tests/unit_tests/api/service/test_cli_diagnostics.py new file mode 100644 index 000000000..c5782a10f --- /dev/null +++ b/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 diff --git a/tests/unit_tests/api/service/test_knowledge_service.py b/tests/unit_tests/api/service/test_knowledge_service.py index 908a6f434..908774892 100644 --- a/tests/unit_tests/api/service/test_knowledge_service.py +++ b/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()