diff --git a/plugins/Runner/LocalAgent/pkg/agent_core/langbot.py b/plugins/Runner/LocalAgent/pkg/agent_core/langbot.py index c66ee512e..b903adca6 100644 --- a/plugins/Runner/LocalAgent/pkg/agent_core/langbot.py +++ b/plugins/Runner/LocalAgent/pkg/agent_core/langbot.py @@ -205,10 +205,12 @@ class LangBotToolExecutor: api: RunnerAPIProxy, allowed_tools: set[str], max_result_chars: int = DEFAULT_MAX_TOOL_RESULT_CHARS, + workspace_manager: typing.Any = None, ): self.api = api self.allowed_tools = allowed_tools self.max_result_chars = max_result_chars + self.workspace_manager = workspace_manager self.local_tools = {} def prepare(self, request: ToolCallRequest) -> PreparedToolCall: @@ -244,6 +246,20 @@ class LangBotToolExecutor: if prepared.error is not None: return self.finalize(prepared, result=None, error=prepared.error) + if self.workspace_manager is not None and self.workspace_manager.is_workspace_tool(prepared.request.name): + try: + result = await self.workspace_manager.execute( + api=self.api, + tool_name=prepared.request.name, + parameters=prepared.parameters, + max_result_chars=self.max_result_chars, + ) + return self.finalize(prepared, result=result, error=None) + except Exception as e: + if is_deadline_exceeded_error(e): + raise + return self.finalize(prepared, result=None, error=f"Workspace tool execution failed: {e}") + try: if prepared.request.name in self.local_tools: result = await self.local_tools[prepared.request.name](**prepared.parameters) diff --git a/plugins/Runner/LocalAgent/pkg/config.py b/plugins/Runner/LocalAgent/pkg/config.py index 820e4e922..d41ba180e 100644 --- a/plugins/Runner/LocalAgent/pkg/config.py +++ b/plugins/Runner/LocalAgent/pkg/config.py @@ -8,6 +8,7 @@ DEFAULT_MAX_TOOL_ITERATIONS = 100 DEFAULT_MAX_TOOL_RESULT_CHARS = 20000 DEFAULT_RUN_TIMEOUT_SECONDS = 300 DEFAULT_TOOL_EXECUTION_MODE = "parallel" +DEFAULT_WORKSPACE_DIRECT_THRESHOLD = 8 VALID_TOOL_EXECUTION_MODES = {"parallel", "serial"} @@ -183,3 +184,19 @@ def _positive_int(value: typing.Any, *, default: int) -> int: if value < 1: return default return value + + +def get_workspace_direct_threshold(config: dict[str, typing.Any] | None) -> int: + """Get the maximum number of workspace operations to expose in direct mode.""" + if not isinstance(config, dict): + return DEFAULT_WORKSPACE_DIRECT_THRESHOLD + val = config.get("workspace-direct-threshold") + if val is None: + val = config.get("workspace_direct_threshold") + if val is None: + val = config.get("workspace-tools-threshold") + if val is None: + val = config.get("workspace_tools_threshold") + if isinstance(val, bool) or not isinstance(val, int) or val < 0: + return DEFAULT_WORKSPACE_DIRECT_THRESHOLD + return val diff --git a/plugins/Runner/LocalAgent/pkg/context_pipeline.py b/plugins/Runner/LocalAgent/pkg/context_pipeline.py index 54a67e8f5..369b5ce93 100644 --- a/plugins/Runner/LocalAgent/pkg/context_pipeline.py +++ b/plugins/Runner/LocalAgent/pkg/context_pipeline.py @@ -443,12 +443,14 @@ class ContextAssembler: budget: ContextBudget | None = None, summarizer: ContextSummarizer | None = None, token_counter: ContextTokenCounter | None = None, + workspace_manager: typing.Any = None, ): self.api = api self.ctx = ctx self.budget = budget or ContextBudget.from_context(ctx) self.summarizer = summarizer self.token_counter = token_counter + self.workspace_manager = workspace_manager async def assemble(self) -> ContextAssembly: user_text = self.ctx.input.to_text() @@ -477,6 +479,10 @@ class ContextAssembler: platform_tools_message = build_platform_tools_system_message(self.ctx) if platform_tools_message is not None: system_instructions.append(platform_tools_message.content) + if self.workspace_manager is not None: + workspace_message = self.workspace_manager.build_system_message() + if workspace_message is not None: + system_instructions.append(workspace_message.content) if system_instructions: # Strict chat templates only accept a system message at position zero. guidance = "\n\n".join(system_instructions) diff --git a/plugins/Runner/LocalAgent/pkg/messages.py b/plugins/Runner/LocalAgent/pkg/messages.py index e3be04eca..5c4dfe8b6 100644 --- a/plugins/Runner/LocalAgent/pkg/messages.py +++ b/plugins/Runner/LocalAgent/pkg/messages.py @@ -145,6 +145,21 @@ def build_platform_tools_system_message(ctx: typing.Any) -> Message | None: ) +def build_workspace_tools_system_message( + domains: list[dict[str, typing.Any]] | None, + max_chars: int = 1500, +) -> Message | None: + """Build a system message with progressive disclosure instructions and domain overview.""" + from pkg.workspace_tools import PROGRESSIVE_GUIDANCE, format_domains_overview + + overview = format_domains_overview(domains or [], max_chars=max_chars) + if overview: + content = f"{PROGRESSIVE_GUIDANCE}\n\n{overview}" + else: + content = PROGRESSIVE_GUIDANCE + return Message(role="system", content=content) + + def build_user_message( user_text: str, input_contents: list[ContentElement] | None = None, diff --git a/plugins/Runner/LocalAgent/pkg/run_assembly.py b/plugins/Runner/LocalAgent/pkg/run_assembly.py index 3d9e42334..5a8cc5414 100644 --- a/plugins/Runner/LocalAgent/pkg/run_assembly.py +++ b/plugins/Runner/LocalAgent/pkg/run_assembly.py @@ -21,6 +21,7 @@ from pkg.config import ( ) from pkg.context_pipeline import ContextAssembler, ContextBudget, HostContextTokenCounter, LLMContextSummarizer from pkg.model_calling import build_llm_tools +from pkg.workspace_tools import WorkspaceToolsManager class NoAuthorizedModelError(Exception): @@ -65,7 +66,14 @@ class AgentRunAssembler: summarizer = LLMContextSummarizer( self.api, model_ids[0], remove_think=remove_think, reasoning_level=reasoning_level ) + + workspace_manager = await WorkspaceToolsManager.discover(self.api, self.ctx) + workspace_manager.resolve_conflicts(allowed_tools) + tools = await self._build_tools(allowed_tools) + tools.extend(workspace_manager.tools) + all_allowed_tools = allowed_tools | workspace_manager._tool_names + token_counter = HostContextTokenCounter(self.api, model_ids[0], tools, reasoning_level=reasoning_level) context_assembly = await ContextAssembler( @@ -74,6 +82,7 @@ class AgentRunAssembler: budget=context_budget, summarizer=summarizer, token_counter=token_counter, + workspace_manager=workspace_manager, ).assemble() return AgentRunAssembly( @@ -82,8 +91,9 @@ class AgentRunAssembler: tools=tools, tool_executor=LangBotToolExecutor( self.api, - allowed_tools, + all_allowed_tools, max_result_chars=max_tool_result_chars, + workspace_manager=workspace_manager, ), hooks=LangBotContextHooks( context_budget, diff --git a/plugins/Runner/LocalAgent/pkg/workspace_tools.py b/plugins/Runner/LocalAgent/pkg/workspace_tools.py new file mode 100644 index 000000000..73353a744 --- /dev/null +++ b/plugins/Runner/LocalAgent/pkg/workspace_tools.py @@ -0,0 +1,609 @@ +"""Workspace Application API tools and progressive disclosure for LocalAgent.""" + +from __future__ import annotations + +import collections.abc +import json +import logging +from dataclasses import dataclass +from typing import Any, Mapping + +from langbot_plugin.api.entities.builtin.provider.message import Message +from langbot_plugin.api.entities.builtin.resource.tool import LLMTool + +from pkg.config import ( + DEFAULT_MAX_TOOL_RESULT_CHARS, + get_workspace_direct_threshold, +) + +logger = logging.getLogger(__name__) + +# Default domain metadata for LangBot Application API +DOMAINS_METADATA: dict[str, dict[str, Any]] = { + "monitoring": { + "label": {"zh_Hans": "仪表盘", "en_US": "Monitoring"}, + "summary": "Inspect system metrics, service health, runtime alerts, and statistics.", + }, + "stats": { + "label": {"zh_Hans": "统计", "en_US": "Statistics"}, + "summary": "Basic usage statistics and activity summaries.", + }, + "bot": { + "label": {"zh_Hans": "机器人", "en_US": "Bot"}, + "summary": "Manage chat bots, configure message routing rules, and send outbound messages.", + }, + "adapter": { + "label": {"zh_Hans": "适配器", "en_US": "Adapter"}, + "summary": "Inspect IM platforms and chat network adapters available in the workspace.", + }, + "agent": { + "label": {"zh_Hans": "智能体", "en_US": "Agent"}, + "summary": "Create and manage autonomous AI agents, inspect execution traces, and debug agent runs.", + }, + "pipeline": { + "label": {"zh_Hans": "流水线", "en_US": "Pipeline"}, + "summary": "Create, configure, and manage message processing pipelines and node topologies.", + }, + "knowledge": { + "label": {"zh_Hans": "知识库", "en_US": "Knowledge Base"}, + "summary": "Manage knowledge bases, index documents, and retrieve relevant knowledge chunks.", + }, + "plugin": { + "label": {"zh_Hans": "插件", "en_US": "Plugin"}, + "summary": "Discover, install, configure, and inspect LangBot plugins and component manifests.", + }, + "mcp": { + "label": {"zh_Hans": "MCP 服务", "en_US": "MCP"}, + "summary": "Manage Model Context Protocol (MCP) server connections and access their resources.", + }, + "skill": { + "label": {"zh_Hans": "技能", "en_US": "Skill"}, + "summary": "Author, inspect, and update agent skills, specialized instructions, and prompts.", + }, + "extension": { + "label": {"zh_Hans": "扩展", "en_US": "Extension"}, + "summary": "List and inspect available workspace extensions and capabilities.", + }, + "tool": { + "label": {"zh_Hans": "工具", "en_US": "Tool"}, + "summary": "Manage workspace custom tools, functions, and external integrations.", + }, + "model": { + "label": {"zh_Hans": "模型", "en_US": "Model"}, + "summary": "Configure language models, fallback strategies, credentials, and parameters.", + }, + "provider": { + "label": {"zh_Hans": "供应商", "en_US": "Provider"}, + "summary": "Manage upstream AI model providers, endpoints, and credentials.", + }, + "requester": { + "label": {"zh_Hans": "请求器", "en_US": "Requester"}, + "summary": "Configure upstream HTTP clients, proxies, retry policies, and timeouts.", + }, +} + +PROGRESSIVE_GUIDANCE = ( + "For workspace capabilities, follow progressive disclosure: first use `workspace_search` to discover operations, " + "then inspect schemas with `workspace_describe`, and finally execute with `workspace_call` " + "(先 search → describe → call)." +) + +TRUNCATION_HINT = ( + "Result exceeded maximum tool result characters limit and was truncated. " + "Use pagination or filtering parameters to narrow the results. " + "(结果超过长度上限已截断，请使用分页或过滤参数缩小结果。)" +) + + +@dataclass(frozen=True) +class NormalizedOperation: + """Normalized representation of an authorized workspace operation.""" + + id: str + domain: str + label: dict[str, str] + summary: str + description: str + risk: str # 'read' | 'write' | 'dangerous' + input_schema: dict[str, Any] + output_schema: dict[str, Any] + + +def operation_id_to_tool_name(operation_id: str) -> str: + """Convert operation ID to direct tool name: 'bot.list' -> 'ws_bot__list'. + + Tool name starts with 'ws_', replaces '.' with '__', and length is bounded <= 64. + """ + tool_name = f"ws_{operation_id.replace('.', '__')}" + if len(tool_name) > 64: + return tool_name[:64] + return tool_name + + +def tool_name_to_operation_id(tool_name: str) -> str: + """Reverse tool name back to operation ID: 'ws_bot__list' -> 'bot.list'.""" + if tool_name.startswith("ws_"): + return tool_name[3:].replace("__", ".") + return tool_name.replace("__", ".") + + +def normalize_operation(op: Any) -> NormalizedOperation: + """Normalize operation from dict or SDK WorkspaceOperationResource.""" + if isinstance(op, Mapping): + op_id = str(op.get("id") or "") + domain = str(op.get("domain") or (op_id.split(".")[0] if "." in op_id else "")) + label = op.get("label") or {} + summary = str(op.get("summary") or "") + description = str(op.get("description") or summary or "") + risk = str(op.get("risk") or "read") + input_schema = op.get("input_schema") or {} + output_schema = op.get("output_schema") or {} + else: + op_id = str(getattr(op, "id", "") or "") + domain = str(getattr(op, "domain", None) or (op_id.split(".")[0] if "." in op_id else "")) + label = getattr(op, "label", None) or {} + summary = str(getattr(op, "summary", None) or "") + description = str(getattr(op, "description", None) or summary or "") + risk = str(getattr(op, "risk", None) or "read") + input_schema = getattr(op, "input_schema", None) or {} + output_schema = getattr(op, "output_schema", None) or {} + + if not isinstance(label, dict): + label = {} + if not isinstance(input_schema, dict): + input_schema = {} + if not isinstance(output_schema, dict): + output_schema = {} + + return NormalizedOperation( + id=op_id, + domain=domain, + label=label, + summary=summary, + description=description, + risk=risk, + input_schema=input_schema, + output_schema=output_schema, + ) + + +def format_domains_overview(domain_list: list[dict[str, Any]], max_chars: int = 1500) -> str: + """Format domain overview with a strict character budget (<= 1500 chars).""" + if not domain_list: + return "" + header = "Available workspace capability domains:\n" + lines: list[str] = [] + current_len = len(header) + for d in domain_list: + name = d.get("domain", "") + label = d.get("label") or {} + zh = label.get("zh_Hans") or label.get("en_US") or name + count = d.get("count", 0) + summary = d.get("summary", "") + line = f"- {name} ({zh}, {count} ops): {summary}" if summary else f"- {name} ({zh}, {count} ops)" + if current_len + len(line) + 1 > max_chars: + prefix = f"- {name} ({zh}, {count} ops): " + available = max_chars - current_len - len(prefix) - 4 + if available > 10: + lines.append(f"{prefix}{summary[:available]}...") + break + lines.append(line) + current_len += len(line) + 1 + return (header + "\n".join(lines))[:max_chars] + + +def build_domains_from_operations(operations: list[NormalizedOperation]) -> list[dict[str, Any]]: + """Build domain list summarizing operation counts and descriptions.""" + domain_counts: dict[str, int] = {} + for op in operations: + d = op.domain or (op.id.split(".")[0] if "." in op.id else "other") + domain_counts[d] = domain_counts.get(d, 0) + 1 + + result = [] + for d, count in sorted(domain_counts.items()): + meta = DOMAINS_METADATA.get(d, {}) + label = meta.get("label", {"zh_Hans": d, "en_US": d.capitalize()}) + summary = meta.get("summary", "") + result.append( + { + "domain": d, + "label": label, + "summary": summary, + "count": count, + } + ) + return result + + +class WorkspaceToolsManager: + """Manages workspace operations discovery, presentation (direct vs progressive), + system prompt injection, conflict resolution, and execution. + """ + + def __init__( + self, + operations: list[NormalizedOperation], + domains: list[dict[str, Any]], + mode: str | None, # "direct", "progressive", or None + tools: list[LLMTool], + ): + self.operations = operations + self.operations_by_id = {op.id: op for op in operations} + self.domains = domains + self.mode = mode + self.tools = tools + self._tool_names = {tool.name for tool in tools} + + @classmethod + async def discover( + cls, + api: Any, + ctx: Any, + ) -> WorkspaceToolsManager: + """Discover authorized workspace operations and build the manager. + + Rule 1: + 优先读取本次运行资源中的 workspace_operations，没有时调用 api.workspace.list()。 + 列表为空、或者 SDK / Core 不支持时，不注册任何相关工具，原有行为保持不变。 + """ + raw_ops: Any = None + resources = getattr(ctx, "resources", None) + if resources is not None: + raw_ops = getattr(resources, "workspace_operations", None) + + domains: list[dict[str, Any]] = [] + normalized_ops: list[NormalizedOperation] = [] + + if raw_ops: + normalized_ops = [normalize_operation(op) for op in raw_ops if op] + normalized_ops = [op for op in normalized_ops if op.id] + domains = build_domains_from_operations(normalized_ops) + else: + # Fallback to api.workspace.list() + workspace_proxy = getattr(api, "workspace", None) + if workspace_proxy is not None and callable(getattr(workspace_proxy, "list", None)): + try: + res = await workspace_proxy.list() + if isinstance(res, dict): + ops_list = res.get("operations") or [] + normalized_ops = [normalize_operation(op) for op in ops_list if op] + normalized_ops = [op for op in normalized_ops if op.id] + domains = res.get("domains") or [] + if not domains and normalized_ops: + domains = build_domains_from_operations(normalized_ops) + except Exception as e: + logger.warning("Failed to list workspace operations via api.workspace.list(): %s", e) + normalized_ops = [] + domains = [] + + if not normalized_ops: + return cls(operations=[], domains=[], mode=None, tools=[]) + + config = getattr(ctx, "config", None) if isinstance(getattr(ctx, "config", None), dict) else {} + threshold = get_workspace_direct_threshold(config) + + if len(normalized_ops) <= threshold: + mode = "direct" + tools = cls._build_direct_tools(normalized_ops) + else: + mode = "progressive" + tools = cls._build_progressive_tools() + + return cls( + operations=normalized_ops, + domains=domains, + mode=mode, + tools=tools, + ) + + @staticmethod + def _build_direct_tools(operations: list[NormalizedOperation]) -> list[LLMTool]: + """Build direct tools: ws_.""" + + async def _placeholder(**kwargs: Any) -> Any: + return kwargs + + tools: list[LLMTool] = [] + for op in operations: + tool_name = operation_id_to_tool_name(op.id) + desc = op.description or op.summary or f"Workspace operation {op.id}" + params = ( + op.input_schema + if isinstance(op.input_schema, dict) and op.input_schema + else { + "type": "object", + "properties": {}, + } + ) + tools.append( + LLMTool( + name=tool_name, + human_desc=desc, + description=desc, + parameters=params, + func=_placeholder, + ) + ) + return tools + + @staticmethod + def _build_progressive_tools() -> list[LLMTool]: + """Build progressive disclosure meta tools: workspace_search, workspace_describe, workspace_call.""" + + async def _placeholder(**kwargs: Any) -> Any: + return kwargs + + search_tool = LLMTool( + name="workspace_search", + human_desc="Search authorized workspace operations by keyword or domain filter", + description=( + "Search authorized workspace operations by keyword or domain filter. " + "Returns a list of matching operations with id, summary, and risk level." + ), + parameters={ + "type": "object", + "properties": { + "query": { + "type": "string", + "description": ( + "Optional keyword to match against operation ID, summary, description, domain, or label." + ), + }, + "domain": { + "type": "string", + "description": ( + "Optional domain filter (e.g. 'bot', 'pipeline', 'monitoring', 'knowledge', 'model')." + ), + }, + "limit": { + "type": "integer", + "description": "Maximum number of results to return (at most 10, default 10).", + "default": 10, + "minimum": 1, + "maximum": 10, + }, + }, + }, + func=_placeholder, + ) + + describe_tool = LLMTool( + name="workspace_describe", + human_desc="Get schema and documentation for an authorized workspace operation", + description="Get documentation, risk level, and JSON input schema for an authorized workspace operation.", + parameters={ + "type": "object", + "properties": { + "operation_id": { + "type": "string", + "description": "Exact operation ID to inspect.", + }, + }, + "required": ["operation_id"], + }, + func=_placeholder, + ) + + call_tool = LLMTool( + name="workspace_call", + human_desc="Execute an authorized workspace operation", + description="Execute an authorized workspace operation with input arguments matching its schema.", + parameters={ + "type": "object", + "properties": { + "operation_id": { + "type": "string", + "description": "Exact operation ID to execute.", + }, + "arguments": { + "type": "object", + "description": "Input arguments matching the operation schema.", + "default": {}, + }, + }, + "required": ["operation_id"], + }, + func=_placeholder, + ) + + return [search_tool, describe_tool, call_tool] + + def resolve_conflicts(self, existing_tool_names: collections.abc.Set[str]) -> None: + """Resolve conflicts with host tools, platform tools, or MCP tools. + + Rule: 工具名不能和宿主工具、平台工具、MCP 工具冲突，冲突时以原有工具为准，并记录 warning。 + """ + filtered_tools: list[LLMTool] = [] + for tool in self.tools: + if tool.name in existing_tool_names: + logger.warning( + "Workspace tool '%s' conflicts with existing tool; keeping existing tool", + tool.name, + ) + else: + filtered_tools.append(tool) + self.tools = filtered_tools + self._tool_names = {t.name for t in filtered_tools} + + def build_system_message(self) -> Message | None: + """Build system prompt message for progressive mode. + + Rule: 渐进模式下，把领域概览（来自 domains，≤ 1500 字符）和“先 search → describe → call” + 的使用说明加进 LocalAgent 的系统提示。 + """ + if self.mode != "progressive" or not self.operations: + return None + + overview = format_domains_overview(self.domains, max_chars=1500) + if overview: + content = f"{PROGRESSIVE_GUIDANCE}\n\n{overview}" + else: + content = PROGRESSIVE_GUIDANCE + + return Message(role="system", content=content) + + def is_workspace_tool(self, tool_name: str) -> bool: + """Check if tool name is handled by this workspace manager.""" + return tool_name in self._tool_names + + def execute_search( + self, + query: str | None = None, + domain: str | None = None, + limit: int = 10, + **kwargs: Any, + ) -> list[dict[str, Any]]: + """Search operations matching query and/or domain filter.""" + ops = list(self.operations) + if domain: + d_lower = str(domain).strip().lower() + ops = [op for op in ops if op.domain.lower() == d_lower] + + if query: + q = str(query).strip().lower() + + def _matches(op: NormalizedOperation) -> bool: + if q in op.id.lower(): + return True + if op.summary and q in op.summary.lower(): + return True + if op.description and q in op.description.lower(): + return True + if op.domain and q in op.domain.lower(): + return True + if op.label: + for val in op.label.values(): + if q in str(val).lower(): + return True + return False + + ops = [op for op in ops if _matches(op)] + + try: + lim = max(1, min(int(limit), 10)) + except (ValueError, TypeError): + lim = 10 + + return [{"id": op.id, "summary": op.summary, "risk": op.risk} for op in ops[:lim]] + + def execute_describe( + self, + operation_id: str, + **kwargs: Any, + ) -> dict[str, Any]: + """Describe operation documentation, risk, and input schema.""" + op = self.operations_by_id.get(operation_id) + if op is None: + return {"error": f"Operation '{operation_id}' not found or not authorized"} + return { + "id": op.id, + "description": op.description, + "risk": op.risk, + "input_schema": op.input_schema, + } + + async def execute( + self, + api: Any, + tool_name: str, + parameters: dict[str, Any], + max_result_chars: int = DEFAULT_MAX_TOOL_RESULT_CHARS, + ) -> Any: + """Execute a workspace tool call.""" + if self.mode == "progressive": + if tool_name == "workspace_search": + res = self.execute_search(**parameters) + return self._truncate_if_needed(res, max_result_chars) + elif tool_name == "workspace_describe": + op_id = parameters.get("operation_id", "") + res = self.execute_describe(operation_id=op_id) + return self._truncate_if_needed(res, max_result_chars) + elif tool_name == "workspace_call": + op_id = parameters.get("operation_id", "") + raw_args = parameters.get("arguments", {}) + if isinstance(raw_args, str): + try: + parsed = json.loads(raw_args) + if isinstance(parsed, dict): + raw_args = parsed + except Exception: + pass + if not isinstance(raw_args, dict): + raw_args = {k: v for k, v in parameters.items() if k != "operation_id"} + + return await self._call_operation( + api=api, + operation_id=op_id, + arguments=raw_args, + max_result_chars=max_result_chars, + ) + else: + return {"error": f"Unknown workspace tool '{tool_name}'"} + elif self.mode == "direct": + op_id = tool_name_to_operation_id(tool_name) + return await self._call_operation( + api=api, + operation_id=op_id, + arguments=parameters, + max_result_chars=max_result_chars, + ) + else: + return {"error": f"Workspace tools not active for '{tool_name}'"} + + async def _call_operation( + self, + api: Any, + operation_id: str, + arguments: dict[str, Any], + max_result_chars: int = DEFAULT_MAX_TOOL_RESULT_CHARS, + ) -> dict[str, Any]: + """Execute operation via api.workspace.call, enforce authorization and truncation.""" + if operation_id not in self.operations_by_id: + return { + "ok": False, + "error": { + "code": "not_authorized", + "message": f"Operation '{operation_id}' is not authorized", + }, + } + + workspace_proxy = getattr(api, "workspace", None) + if workspace_proxy is None or not callable(getattr(workspace_proxy, "call", None)): + return { + "ok": False, + "error": { + "code": "unsupported", + "message": "Workspace operations not supported by host or SDK", + }, + } + + try: + res = await workspace_proxy.call(operation_id, **arguments) + except Exception as exc: + return { + "ok": False, + "error": { + "code": "execution_failed", + "message": str(exc), + }, + } + + # Rule 3: ok: false 时，把错误原样返回给模型，不重试写操作。 + if isinstance(res, dict) and res.get("ok") is False: + return res + + # Rule 3: 返回结果超过 LocalAgent 现有工具结果的长度上限时截断，返回中给出提示：用分页或过滤参数缩小结果。 + return self._truncate_if_needed(res, max_result_chars) + + @staticmethod + def _truncate_if_needed(res: Any, max_result_chars: int) -> Any: + serialized = json.dumps(res, ensure_ascii=False, default=str) + if len(serialized) > max_result_chars: + preview_budget = max(0, max_result_chars - 300) + return { + "ok": res.get("ok", True) if isinstance(res, dict) else True, + "truncated": True, + "preview": serialized[:preview_budget], + "hint": TRUNCATION_HINT, + } + return res diff --git a/plugins/Runner/LocalAgent/tests/test_workspace_tools.py b/plugins/Runner/LocalAgent/tests/test_workspace_tools.py new file mode 100644 index 000000000..67bdd5a87 --- /dev/null +++ b/plugins/Runner/LocalAgent/tests/test_workspace_tools.py @@ -0,0 +1,568 @@ +"""Tests for workspace Application API tools and progressive disclosure.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, Mock + +import pytest +from langbot_plugin.api.entities.builtin.provider.message import Message +from langbot_plugin.api.entities.builtin.runner.context import RunnerContext +from langbot_plugin.api.entities.builtin.runner.resources import ( + AgentResources, + ModelResource, + ToolResource, + WorkspaceOperationResource, +) +from langbot_plugin.api.entities.builtin.runner.result import ( + RunnerResultType, +) + +from components.runner.default import DefaultRunner +from pkg.agent_core.langbot import LangBotToolExecutor +from pkg.agent_core.types import ToolCallRequest +from pkg.config import get_workspace_direct_threshold +from pkg.workspace_tools import ( + WorkspaceToolsManager, + operation_id_to_tool_name, + tool_name_to_operation_id, +) +from tests.test_runner import FakeRunnerAPIProxy, make_context + + +def _make_op( + op_id: str, + domain: str | None = None, + summary: str = "", + description: str = "", + risk: str = "read", + input_schema: dict[str, Any] | None = None, +) -> WorkspaceOperationResource: + d = domain or op_id.split(".")[0] + return WorkspaceOperationResource( + id=op_id, + domain=d, + label={"zh_Hans": d, "en_US": d.capitalize()}, + summary=summary or f"Summary for {op_id}", + description=description or f"Description for {op_id}", + risk=risk, + input_schema=input_schema or {"type": "object", "properties": {}}, + ) + + +def _make_ctx( + workspace_ops: list[WorkspaceOperationResource] | None = None, + tools: list[ToolResource] | None = None, + config: dict[str, Any] | None = None, + models: list[ModelResource] | None = None, +) -> RunnerContext: + if models is None: + models = [ModelResource(model_id="primary-model", model_name="Primary")] + cfg = { + "model": {"primary": "primary-model", "fallbacks": []}, + **(config or {}), + } + resources = AgentResources( + models=models, + tools=tools or [], + workspace_operations=workspace_ops if workspace_ops is not None else [], + ) + return make_context( + run_id="run-test-1", + input_text="test question", + config=cfg, + resources=resources, + delivery_supports_streaming=False, + ) + + +# ============================================================================== +# 1. 授权列表为空时，不注册工具 +# ============================================================================== + + +@pytest.mark.asyncio +async def test_empty_workspace_operations_registers_no_tools(): + ctx = _make_ctx(workspace_ops=[]) + api = Mock() + api.workspace = Mock() + api.workspace.list = AsyncMock(return_value={"operations": [], "domains": []}) + + mgr = await WorkspaceToolsManager.discover(api, ctx) + assert mgr.mode is None + assert len(mgr.tools) == 0 + assert len(mgr.operations) == 0 + assert mgr.build_system_message() is None + + +# ============================================================================== +# 2. 3 个操作时为直接模式，20 个操作时为渐进模式 (及阈值可配置) +# ============================================================================== + + +@pytest.mark.asyncio +async def test_direct_mode_for_3_operations(): + ops = [ + _make_op("bot.list", summary="List bots"), + _make_op("bot.create", summary="Create bot", risk="write"), + _make_op("stats.basic", summary="Basic statistics"), + ] + ctx = _make_ctx(workspace_ops=ops) + api = Mock() + + mgr = await WorkspaceToolsManager.discover(api, ctx) + assert mgr.mode == "direct" + assert len(mgr.tools) == 3 + tool_names = {t.name for t in mgr.tools} + assert tool_names == {"ws_bot__list", "ws_bot__create", "ws_stats__basic"} + # Direct mode does not register progressive meta tools + assert "workspace_search" not in tool_names + assert "workspace_describe" not in tool_names + assert "workspace_call" not in tool_names + # Direct mode adds no progressive system guidance + assert mgr.build_system_message() is None + + +@pytest.mark.asyncio +async def test_progressive_mode_for_20_operations(): + ops = [_make_op(f"domain_{i}.action_{i}") for i in range(20)] + ctx = _make_ctx(workspace_ops=ops) + api = Mock() + + mgr = await WorkspaceToolsManager.discover(api, ctx) + assert mgr.mode == "progressive" + assert len(mgr.tools) == 3 + tool_names = {t.name for t in mgr.tools} + assert tool_names == {"workspace_search", "workspace_describe", "workspace_call"} + + # Progressive mode injects domain overview and guidance into system prompt + msg = mgr.build_system_message() + assert msg is not None + assert msg.role == "system" + assert "先 search → describe → call" in msg.content + assert "Available workspace capability domains:" in msg.content + + +@pytest.mark.asyncio +async def test_configurable_threshold(): + ops = [_make_op(f"bot.action_{i}") for i in range(5)] + # Default threshold is 8 -> direct mode + ctx_default = _make_ctx(workspace_ops=ops) + mgr_default = await WorkspaceToolsManager.discover(Mock(), ctx_default) + assert mgr_default.mode == "direct" + + # Configured threshold = 3 -> switches to progressive mode + ctx_custom = _make_ctx(workspace_ops=ops, config={"workspace-direct-threshold": 3}) + assert get_workspace_direct_threshold(ctx_custom.config) == 3 + mgr_custom = await WorkspaceToolsManager.discover(Mock(), ctx_custom) + assert mgr_custom.mode == "progressive" + + +# ============================================================================== +# 3. 工具名可以正确还原为操作 ID +# ============================================================================== + + +def test_tool_name_bidirectional_mapping(): + cases = [ + ("bot.list", "ws_bot__list"), + ("stats.basic", "ws_stats__basic"), + ("pipeline.extensions.update", "ws_pipeline__extensions__update"), + ("monitoring.metrics.query", "ws_monitoring__metrics__query"), + ("knowledge.documents.chunks.list", "ws_knowledge__documents__chunks__list"), + ] + for op_id, expected_tool_name in cases: + tool_name = operation_id_to_tool_name(op_id) + assert tool_name == expected_tool_name + assert len(tool_name) <= 64 + reversed_op_id = tool_name_to_operation_id(tool_name) + assert reversed_op_id == op_id + + +# ============================================================================== +# 4. search 和 describe 只能看到授权列表里的操作 +# ============================================================================== + + +@pytest.mark.asyncio +async def test_search_and_describe_isolated_to_authorized(): + ops = [ + _make_op("bot.list", domain="bot", summary="List chat bots", risk="read"), + _make_op("bot.create", domain="bot", summary="Create a new bot", risk="write"), + _make_op("stats.basic", domain="stats", summary="Basic metrics and counts", risk="read"), + ] + ctx = _make_ctx(workspace_ops=ops, config={"workspace-direct-threshold": 1}) # force progressive + mgr = await WorkspaceToolsManager.discover(Mock(), ctx) + assert mgr.mode == "progressive" + + # Search by keyword in authorized operations + search_res = mgr.execute_search(query="bot") + assert len(search_res) == 2 + assert {r["id"] for r in search_res} == {"bot.list", "bot.create"} + assert all(set(r.keys()) == {"id", "summary", "risk"} for r in search_res) + + # Search for an unauthorized domain/operation returns empty + search_unauthorized = mgr.execute_search(query="pipeline") + assert search_unauthorized == [] + + # Describe authorized operation + desc_ok = mgr.execute_describe("bot.list") + assert desc_ok["id"] == "bot.list" + assert "description" in desc_ok + assert desc_ok["risk"] == "read" + assert "input_schema" in desc_ok + + # Describe unauthorized operation returns error + desc_err = mgr.execute_describe("pipeline.update") + assert "error" in desc_err + assert "not found or not authorized" in desc_err["error"] + + +# ============================================================================== +# 5. call 能把参数正确透传给 SDK (直接模式与渐进模式) +# ============================================================================== + + +@pytest.mark.asyncio +async def test_direct_mode_call_passthrough(): + ops = [_make_op("bot.list", input_schema={"type": "object", "properties": {"limit": {"type": "integer"}}})] + ctx = _make_ctx(workspace_ops=ops) + api = Mock() + api.workspace = Mock() + api.workspace.call = AsyncMock(return_value={"ok": True, "result": {"items": ["b1", "b2"]}}) + + mgr = await WorkspaceToolsManager.discover(api, ctx) + res = await mgr.execute(api, "ws_bot__list", {"limit": 10}) + + assert res["ok"] is True + assert res["result"] == {"items": ["b1", "b2"]} + api.workspace.call.assert_awaited_once_with("bot.list", limit=10) + + +@pytest.mark.asyncio +async def test_progressive_mode_call_passthrough(): + ops = [_make_op(f"op_{i}.action") for i in range(10)] + target_op = ops[0].id + ctx = _make_ctx(workspace_ops=ops) + api = Mock() + api.workspace = Mock() + api.workspace.call = AsyncMock(return_value={"ok": True, "result": {"status": "created"}}) + + mgr = await WorkspaceToolsManager.discover(api, ctx) + + # Calling via workspace_call with nested arguments + res = await mgr.execute( + api, + "workspace_call", + {"operation_id": target_op, "arguments": {"name": "test_bot"}}, + ) + assert res["ok"] is True + api.workspace.call.assert_awaited_once_with(target_op, name="test_bot") + + # Calling an unauthorized operation is blocked locally + res_unauth = await mgr.execute( + api, + "workspace_call", + {"operation_id": "unauthorized.dangerous.delete", "arguments": {}}, + ) + assert res_unauth["ok"] is False + assert res_unauth["error"]["code"] == "not_authorized" + + +# ============================================================================== +# 6. 结果超长时被截断并给出提示 +# ============================================================================== + + +@pytest.mark.asyncio +async def test_large_result_truncated_with_hint(): + ops = [_make_op("stats.large")] + ctx = _make_ctx(workspace_ops=ops) + api = Mock() + api.workspace = Mock() + # Massive payload > 20000 chars + huge_payload = {"ok": True, "data": "X" * 30000} + api.workspace.call = AsyncMock(return_value=huge_payload) + + mgr = await WorkspaceToolsManager.discover(api, ctx) + res = await mgr.execute(api, "ws_stats__large", {}, max_result_chars=1000) + + assert res["ok"] is True + assert res["truncated"] is True + assert "preview" in res + assert "hint" in res + assert "分页或过滤参数缩小结果" in res["hint"] + + +# ============================================================================== +# 7. 与原有工具同名时，以原有工具为准，并记录 warning +# ============================================================================== + + +@pytest.mark.asyncio +async def test_tool_name_conflict_resolution(caplog): + ops = [_make_op("bot.list")] # direct name: ws_bot__list + ctx = _make_ctx(workspace_ops=ops) + api = Mock() + + mgr = await WorkspaceToolsManager.discover(api, ctx) + assert len(mgr.tools) == 1 + assert mgr.tools[0].name == "ws_bot__list" + + # Suppose host already has a tool named ws_bot__list + existing_tools = {"ws_bot__list", "bash", "grep"} + with caplog.at_level("WARNING"): + mgr.resolve_conflicts(existing_tools) + + # Workspace tool should be dropped + assert len(mgr.tools) == 0 + assert any("conflicts with existing tool" in rec.message for rec in caplog.records) + + +# ============================================================================== +# 8. SDK 不支持时，自动降级为原有行为 +# ============================================================================== + + +@pytest.mark.asyncio +async def test_sdk_unsupported_graceful_degradation(): + ctx = _make_ctx(workspace_ops=None) + # API has no workspace attribute at all + api_without_workspace = Mock(spec=[]) + + mgr = await WorkspaceToolsManager.discover(api_without_workspace, ctx) + assert mgr.mode is None + assert len(mgr.tools) == 0 + assert mgr.build_system_message() is None + + +# ============================================================================== +# 9. ok: false 时把错误原样返回给模型，不重试写操作 +# ============================================================================== + + +@pytest.mark.asyncio +async def test_ok_false_returned_verbatim(): + ops = [_make_op("bot.delete", risk="dangerous")] + ctx = _make_ctx(workspace_ops=ops) + api = Mock() + api.workspace = Mock() + api.workspace.call = AsyncMock( + return_value={ + "ok": False, + "error": {"code": "not_found", "message": "Bot does not exist"}, + } + ) + + mgr = await WorkspaceToolsManager.discover(api, ctx) + res = await mgr.execute(api, "ws_bot__delete", {"uuid": "b-999"}) + + assert res["ok"] is False + assert res["error"]["code"] == "not_found" + assert res["error"]["message"] == "Bot does not exist" + + +# ============================================================================== +# 10. LangBotToolExecutor 调度 workspace tools 验证 +# ============================================================================== + + +@pytest.mark.asyncio +async def test_tool_executor_handles_workspace_tools(): + ops = [_make_op("bot.list")] + ctx = _make_ctx(workspace_ops=ops) + api = Mock() + api.workspace = Mock() + api.workspace.call = AsyncMock(return_value={"ok": True, "result": ["bot-1"]}) + + mgr = await WorkspaceToolsManager.discover(api, ctx) + executor = LangBotToolExecutor( + api=api, + allowed_tools={"ws_bot__list"}, + max_result_chars=20000, + workspace_manager=mgr, + ) + + prep = executor.prepare( + ToolCallRequest( + id="call-1", + name="ws_bot__list", + arguments='{"limit": 5}', + ) + ) + outcome = await executor.execute(prep) + + assert outcome.error is None + assert outcome.result == {"ok": True, "result": ["bot-1"]} + assert outcome.message is not None + assert outcome.message.role == "tool" + assert "bot-1" in outcome.message.content + + +# ============================================================================== +# 11. 端到端本地运行记录：直接模式与渐进模式 +# ============================================================================== + + +@pytest.mark.asyncio +async def test_e2e_direct_mode_agent_run(): + """Simulated local agent run in direct mode.""" + ops = [ + _make_op("bot.list", summary="List bots"), + _make_op("stats.basic", summary="Basic usage stats"), + ] + ctx = _make_ctx(workspace_ops=ops) + + api = FakeRunnerAPIProxy( + models=[ModelResource(model_id="primary-model", model_name="Primary")], + tools=[], + ) + api.get_prompt = AsyncMock(return_value=[{"role": "system", "content": "You are a test assistant."}]) + api.workspace = Mock() + api.workspace.call = AsyncMock(return_value={"ok": True, "result": {"bots": [{"id": "b1", "name": "HelperBot"}]}}) + + # Model mock: turn 1 calls ws_bot__list; turn 2 responds with text + turn_1_msg = Message( + role="assistant", + content="I will check the bots.", + tool_calls=[ + { + "id": "call-bot-list-1", + "type": "function", + "function": {"name": "ws_bot__list", "arguments": '{"limit": 5}'}, + } + ], + ) + turn_2_msg = Message( + role="assistant", + content="There is 1 bot available: HelperBot (id: b1).", + ) + + call_count = 0 + + async def mock_invoke(**kwargs): + nonlocal call_count + call_count += 1 + if call_count == 1: + return turn_1_msg + return turn_2_msg + + api.invoke_llm = AsyncMock(side_effect=mock_invoke) + if hasattr(api, "invoke_llm_with_usage"): + delattr(api, "invoke_llm_with_usage") + + runner = DefaultRunner() + runner.get_run_api = lambda _c: api + results = [] + async for item in runner.run(ctx): + results.append(item) + + tool_started = [r for r in results if r.type == RunnerResultType.TOOL_CALL_STARTED] + tool_completed = [r for r in results if r.type == RunnerResultType.TOOL_CALL_COMPLETED] + msg_completed = [r for r in results if r.type == RunnerResultType.MESSAGE_COMPLETED] + run_completed = [r for r in results if r.type == RunnerResultType.RUN_COMPLETED] + + assert len(tool_started) == 1 + assert tool_started[0].data["tool_name"] == "ws_bot__list" + assert tool_started[0].data["parameters"] == {"limit": 5} + + assert len(tool_completed) == 1 + assert tool_completed[0].data["tool_name"] == "ws_bot__list" + + assert len(msg_completed) == 1 + assert "HelperBot" in msg_completed[0].data["message"]["content"] + assert len(run_completed) == 1 + + +@pytest.mark.asyncio +async def test_e2e_progressive_mode_agent_run(): + """Simulated local agent run in progressive mode with search -> describe -> call.""" + ops = [_make_op(f"op_{i}.action_{i}", summary=f"Action {i} for testing") for i in range(15)] + ctx = _make_ctx(workspace_ops=ops) + + api = FakeRunnerAPIProxy( + models=[ModelResource(model_id="primary-model", model_name="Primary")], + tools=[], + ) + api.get_prompt = AsyncMock(return_value=[{"role": "system", "content": "You are a test assistant."}]) + api.workspace = Mock() + api.workspace.call = AsyncMock( + return_value={"ok": True, "result": {"success": True, "details": "Operation executed"}} + ) + + # 3-step progressive execution: + # Turn 1: workspace_search(query="action_1") + # Turn 2: workspace_describe(operation_id="op_1.action_1") + # Turn 3: workspace_call(operation_id="op_1.action_1", arguments={"param": 42}) + # Turn 4: Final response + turn_1 = Message( + role="assistant", + content="Searching operations...", + tool_calls=[ + { + "id": "c-1", + "type": "function", + "function": {"name": "workspace_search", "arguments": '{"query": "action_1"}'}, + } + ], + ) + turn_2 = Message( + role="assistant", + content="Describing schema...", + tool_calls=[ + { + "id": "c-2", + "type": "function", + "function": {"name": "workspace_describe", "arguments": '{"operation_id": "op_1.action_1"}'}, + } + ], + ) + turn_3 = Message( + role="assistant", + content="Calling operation...", + tool_calls=[ + { + "id": "c-3", + "type": "function", + "function": { + "name": "workspace_call", + "arguments": '{"operation_id": "op_1.action_1", "arguments": {"param": 42}}', + }, + } + ], + ) + turn_4 = Message( + role="assistant", + content="Operation op_1.action_1 successfully completed.", + ) + + turns = [turn_1, turn_2, turn_3, turn_4] + turn_idx = 0 + + async def mock_invoke(**kwargs): + nonlocal turn_idx + idx = min(turn_idx, len(turns) - 1) + turn_idx += 1 + return turns[idx] + + api.invoke_llm = AsyncMock(side_effect=mock_invoke) + if hasattr(api, "invoke_llm_with_usage"): + delattr(api, "invoke_llm_with_usage") + + runner = DefaultRunner() + runner.get_run_api = lambda _c: api + results = [] + async for item in runner.run(ctx): + results.append(item) + + tool_started = [r for r in results if r.type == RunnerResultType.TOOL_CALL_STARTED] + tool_completed = [r for r in results if r.type == RunnerResultType.TOOL_CALL_COMPLETED] + msg_completed = [r for r in results if r.type == RunnerResultType.MESSAGE_COMPLETED] + run_completed = [r for r in results if r.type == RunnerResultType.RUN_COMPLETED] + + called_tool_names = [t.data["tool_name"] for t in tool_started] + assert called_tool_names == ["workspace_search", "workspace_describe", "workspace_call"] + assert len(tool_completed) == 3 + assert len(msg_completed) == 1 + assert "op_1.action_1 successfully completed" in msg_completed[0].data["message"]["content"] + assert len(run_completed) == 1