diff --git a/app/agent/__init__.py b/app/agent/__init__.py index 6a90e1c21..d24e651a8 100644 --- a/app/agent/__init__.py +++ b/app/agent/__init__.py @@ -27,6 +27,7 @@ from langgraph.checkpoint.memory import InMemorySaver from app.agent.callback import StreamingHandler from app.agent.llm import LLMHelper +from app.agent.llm.server_tools import ServerToolRegistry from app.agent.memory import memory_manager from app.agent.middleware.activity_log import ( ActivityLogMiddleware, @@ -60,10 +61,15 @@ from app.agent.policy import ( from app.agent.runtime import agent_runtime_manager from app.agent.mcp import agent_mcp_manager from app.agent.tools.factory import MoviePilotToolFactory -from app.agent.tools.impl.mcp import create_external_mcp_tools +from app.agent.tools.catalog import ToolCatalogSnapshot +from app.agent.tools.impl.mcp import ( + create_external_mcp_tools, + select_legacy_mcp_tools, +) from app.chain import ChainBase from app.core.config import settings from app.core.event import eventmanager +from app.core.plugin import PluginManager from app.db.agentchat_oper import AgentChatOper from app.db.agenttask_oper import AgentTaskOper from app.db.user_oper import UserOper @@ -192,6 +198,11 @@ class _CompiledAgentBundle: agent: Any streaming: bool created_at: datetime + tool_catalog: Optional[ToolCatalogSnapshot] = None + subagent_catalog: Optional[ToolCatalogSnapshot] = None + plugin_revision: int = -1 + mcp_config_signature: str = "" + catalog_checked_at: Optional[datetime] = None class _ThinkTagStripper: @@ -297,6 +308,8 @@ class MoviePilotAgent: MoviePilot AI智能体（基于 LangChain v1 + LangGraph） """ + TOOL_CATALOG_REFRESH_SECONDS = 30 + def __init__( self, session_id: str, @@ -1116,7 +1129,7 @@ class MoviePilotAgent: def _initialize_tools(self) -> List: """ - 初始化工具列表 + 初始化主 Agent 本地工具实例。 """ return MoviePilotToolFactory.create_tools( session_id=self.session_id, @@ -1129,6 +1142,36 @@ class MoviePilotAgent: allow_message_tools=self.allow_message_tools, ) + def _initialize_local_tool_catalogs( + self, + ) -> tuple[ToolCatalogSnapshot, ToolCatalogSnapshot]: + """在同一插件 revision 窗口内建立主图和子图工具目录。""" + plugin_manager = PluginManager() + for _attempt in range(MoviePilotToolFactory.CATALOG_BUILD_MAX_ATTEMPTS): + before_revision = plugin_manager.get_plugin_agent_tools_revision() + tools = self._initialize_tools() + subagent_tools = self._initialize_subagent_tools() + after_revision = plugin_manager.get_plugin_agent_tools_revision() + if before_revision == after_revision: + factory_revision = MoviePilotToolFactory.catalog_factory_revision() + return ( + ToolCatalogSnapshot.from_tools( + tools, + plugin_revision=after_revision, + factory_revision=factory_revision, + ), + ToolCatalogSnapshot.from_tools( + subagent_tools, + plugin_revision=after_revision, + factory_revision=factory_revision, + ), + ) + raise RuntimeError("插件工具目录持续变化，无法建立当前快照") + + def _initialize_tool_catalog(self) -> ToolCatalogSnapshot: + """兼容只需要主 Agent 工具目录的内部调用与测试。""" + return self._initialize_local_tool_catalogs()[0] + @staticmethod def _filter_local_web_search_tools(tools: List, enabled: bool) -> List: """按联网搜索策略保留或移除本地 search_web 工具。""" @@ -1168,7 +1211,12 @@ class MoviePilotAgent: runtime_config.get("web_search_mode"), ) - async def _agent_bundle_signature(self, streaming: bool) -> tuple[Any, ...]: + async def _agent_bundle_signature( + self, + streaming: bool, + tool_catalog: Optional[ToolCatalogSnapshot] = None, + subagent_catalog: Optional[ToolCatalogSnapshot] = None, + ) -> tuple[Any, ...]: """构造会话内 Agent 图缓存签名。""" runtime_config = await self._resolve_llm_runtime_config() return ( @@ -1188,6 +1236,14 @@ class MoviePilotAgent: self._public_runtime_config_signature(runtime_config), agent_runtime_manager.current_signature(), agent_mcp_manager.config_signature(), + ( + (tool_catalog.signature, subagent_catalog.signature) + if tool_catalog is not None and subagent_catalog is not None + else ( + MoviePilotToolFactory.catalog_factory_revision(), + PluginManager().get_plugin_agent_tools_revision(), + ) + ), ) def _get_cached_agent( @@ -1209,6 +1265,9 @@ class MoviePilotAgent: signature: tuple[Any, ...], agent: Any, streaming: bool, + tool_catalog: ToolCatalogSnapshot, + subagent_catalog: ToolCatalogSnapshot, + mcp_config_signature: str, ) -> Any: """保存当前会话可复用的 Agent 图。""" self._compiled_agent_bundle = _CompiledAgentBundle( @@ -1216,6 +1275,11 @@ class MoviePilotAgent: agent=agent, streaming=streaming, created_at=datetime.now(), + tool_catalog=tool_catalog, + subagent_catalog=subagent_catalog, + plugin_revision=tool_catalog.plugin_revision, + mcp_config_signature=mcp_config_signature, + catalog_checked_at=datetime.now(), ) return agent @@ -1244,7 +1308,7 @@ class MoviePilotAgent: allow_message_tools=False, ) - async def _initialize_mcp_tools(self) -> List: + async def _initialize_mcp_tools(self, specs=None) -> List: """ 初始化外部 MCP 工具列表。 """ @@ -1256,9 +1320,10 @@ class MoviePilotAgent: username=self.username, stream_handler=self.stream_handler, agent_context=self._tool_context, + specs=specs, ) - async def _initialize_subagent_mcp_tools(self) -> List: + async def _initialize_subagent_mcp_tools(self, specs=None) -> List: """ 初始化子代理可用的外部 MCP 工具列表。 """ @@ -1275,6 +1340,7 @@ class MoviePilotAgent: "should_dispatch_reply": False, "is_admin": bool(self._tool_context.get("is_admin")), }, + specs=specs, ) async def _create_agent(self, streaming: bool = False): @@ -1283,13 +1349,68 @@ class MoviePilotAgent: :param streaming: 是否启用流式输出 """ try: - bundle_signature = await self._agent_bundle_signature(streaming) - cached_agent = self._get_cached_agent(bundle_signature, streaming) - self._last_agent_cache_hit = bool(cached_agent) - if cached_agent: - logger.debug(f"复用会话内 Agent 图: session_id={self.session_id}") - return cached_agent - + runtime_config = await self._resolve_llm_runtime_config() + plugin_revision = PluginManager().get_plugin_agent_tools_revision() + mcp_config_signature = agent_mcp_manager.config_signature() + cached_bundle = self._compiled_agent_bundle + catalog_is_fresh = bool( + cached_bundle + and cached_bundle.streaming == streaming + and cached_bundle.tool_catalog is not None + and cached_bundle.subagent_catalog is not None + and cached_bundle.plugin_revision == plugin_revision + and cached_bundle.mcp_config_signature == mcp_config_signature + and cached_bundle.catalog_checked_at is not None + and ( + datetime.now() - cached_bundle.catalog_checked_at + ).total_seconds() < self.TOOL_CATALOG_REFRESH_SECONDS + ) + if catalog_is_fresh: + bundle_signature = await self._agent_bundle_signature( + streaming, + tool_catalog=cached_bundle.tool_catalog, + subagent_catalog=cached_bundle.subagent_catalog, + ) + cached_agent = self._get_cached_agent(bundle_signature, streaming) + self._last_agent_cache_hit = bool(cached_agent) + if cached_agent: + logger.debug( + f"复用会话内 Agent 图: session_id={self.session_id}" + ) + return cached_agent + web_search_resolution = ServerToolRegistry.resolve_web_search( + provider=str(runtime_config.get("provider") or ""), + model=str(runtime_config.get("model") or ""), + mode=runtime_config.get("web_search_mode"), + api_protocol=runtime_config.get("api_protocol"), + base_url=runtime_config.get("base_url"), + ) + base_tool_catalog, base_subagent_catalog = ( + self._initialize_local_tool_catalogs() + ) + mcp_specs = await agent_mcp_manager.list_enabled_tool_specs() + local_tools = self._filter_local_web_search_tools( + base_tool_catalog.tools, + enabled=web_search_resolution.use_local_web_search, + ) + mcp_tools = await self._initialize_mcp_tools(specs=mcp_specs) + tools = [*local_tools, *select_legacy_mcp_tools(mcp_tools)] + local_subagent_tools = self._filter_local_web_search_tools( + base_subagent_catalog.tools, + enabled=web_search_resolution.use_local_web_search, + ) + subagent_mcp_tools = await self._initialize_subagent_mcp_tools( + specs=mcp_specs + ) + subagent_catalog = ToolCatalogSnapshot.from_tools( + [*local_subagent_tools, *subagent_mcp_tools], + plugin_revision=base_subagent_catalog.plugin_revision, + factory_revision=base_subagent_catalog.factory_revision, + ) + subagent_tools = [ + *local_subagent_tools, + *select_legacy_mcp_tools(subagent_mcp_tools), + ] # 系统提示词 system_prompt = prompt_manager.get_agent_prompt(channel=self.channel) @@ -1298,7 +1419,6 @@ class MoviePilotAgent: self._sync_model_profile(agent_model) # 供应商原生工具不进入本地 ToolNode，宿主策略只覆盖 client-side tools。 server_tools = LLMHelper.get_server_tools(agent_model) - use_local_web_search = LLMHelper.should_use_local_web_search(agent_model) # 为内部模型调用准备非流式 LLM，避免与用户流式回复复用同一实例。 non_streaming_model = ( @@ -1306,13 +1426,6 @@ class MoviePilotAgent: if not streaming else await self._initialize_llm(streaming=False) ) - - # 工具列表 - tools = self._filter_local_web_search_tools( - self._initialize_tools(), - enabled=use_local_web_search, - ) - tools.extend(await self._initialize_mcp_tools()) skills_middleware = SkillsMiddleware( sources=[str(agent_runtime_manager.skills_dir)], bundled_skills_dir=str(settings.ROOT_PATH / "skills"), @@ -1329,11 +1442,6 @@ class MoviePilotAgent: activity_log_tools = list( getattr(activity_log_middleware, "tools", []) or [] ) - subagent_tools = self._filter_local_web_search_tools( - self._initialize_subagent_tools(), - enabled=use_local_web_search, - ) - subagent_tools.extend(await self._initialize_subagent_mcp_tools()) policy_context = self._build_policy_context() subagent_middlewares, subagent_task_tools = create_subagent_middlewares( model=non_streaming_model, @@ -1341,7 +1449,32 @@ class MoviePilotAgent: server_tools=server_tools, stream_handler=self.stream_handler, policy_context=policy_context.for_subagent(), + catalog=subagent_catalog, ) + # 严格目录必须覆盖 LangGraph ToolNode 可执行的全部 client-side 工具。 + tool_catalog = ToolCatalogSnapshot.from_tools( + [ + *local_tools, + *mcp_tools, + *skill_tools, + *activity_log_tools, + *subagent_task_tools, + ], + plugin_revision=base_tool_catalog.plugin_revision, + factory_revision=base_tool_catalog.factory_revision, + ) + bundle_signature = await self._agent_bundle_signature( + streaming, + tool_catalog=tool_catalog, + subagent_catalog=subagent_catalog, + ) + cached_agent = self._get_cached_agent(bundle_signature, streaming) + self._last_agent_cache_hit = bool(cached_agent) + if cached_agent: + # 签名相同表示已编译图中的精确工具实例仍有效；新建快照仅用于复核。 + cached_bundle.catalog_checked_at = datetime.now() + logger.debug(f"复用会话内 Agent 图: session_id={self.session_id}") + return cached_agent max_tools = settings.LLM_MAX_TOOLS always_include_tools = ( MoviePilotToolFactory.get_tool_selector_always_include_names(tools) @@ -1369,7 +1502,10 @@ class MoviePilotAgent: # 中间件 middlewares = [ # 宿主策略必须位于最外层，确保插件覆盖工具基类也不能绕过。 - AgentPolicyMiddleware(context=policy_context), + AgentPolicyMiddleware( + context=policy_context, + catalog=tool_catalog, + ), # Skills skills_middleware, # Jobs 任务管理 @@ -1412,7 +1548,12 @@ class MoviePilotAgent: agent = create_agent( model=agent_model, - tools=[*tools, *skill_tools, *activity_log_tools, *server_tools], + tools=[ + *tools, + *skill_tools, + *activity_log_tools, + *server_tools, + ], system_prompt=system_prompt, middleware=middlewares, checkpointer=InMemorySaver(), @@ -1421,6 +1562,9 @@ class MoviePilotAgent: signature=bundle_signature, agent=agent, streaming=streaming, + tool_catalog=tool_catalog, + subagent_catalog=subagent_catalog, + mcp_config_signature=mcp_config_signature, ) except Exception as e: logger.error(f"创建 Agent 失败: {e}") diff --git a/app/agent/mcp.py b/app/agent/mcp.py index cfb0ee7b7..635a60720 100644 --- a/app/agent/mcp.py +++ b/app/agent/mcp.py @@ -541,19 +541,14 @@ class AgentMcpManager: return tool_specs async def list_enabled_tool_specs(self) -> list[AgentMcpToolSpec]: - """读取所有启用 MCP 服务器暴露的工具定义。""" + """读取所有启用 MCP 服务器暴露的工具定义并保留同名冲突。""" tool_specs: list[AgentMcpToolSpec] = [] - seen_names: set[str] = set() for server in self.get_servers(): if not server.enabled: continue try: for spec in await self.list_server_tools(server): - if spec.agent_tool_name in seen_names: - logger.warning(f"跳过重复的 MCP Agent 工具名: {spec.agent_tool_name}") - continue tool_specs.append(spec) - seen_names.add(spec.agent_tool_name) except Exception as err: logger.warning(f"读取 MCP 服务器 {server.name} 工具失败: {err}") return tool_specs diff --git a/app/agent/middleware/policy.py b/app/agent/middleware/policy.py index 384885583..64dcdaa6d 100644 --- a/app/agent/middleware/policy.py +++ b/app/agent/middleware/policy.py @@ -11,6 +11,7 @@ from app.agent.policy import ( ToolPolicyContext, call_policy_hook, ) +from app.agent.tools.catalog import ToolCatalogSnapshot class AgentPolicyMiddleware(AgentMiddleware): @@ -25,10 +26,12 @@ class AgentPolicyMiddleware(AgentMiddleware): *, context: ToolPolicyContext, orchestrator: AgentToolPolicyOrchestrator = DEFAULT_TOOL_POLICY_ORCHESTRATOR, + catalog: ToolCatalogSnapshot | None = None, ) -> None: """绑定宿主可信上下文和共享策略编排器。""" self.context = context self.orchestrator = orchestrator + self.catalog = catalog async def awrap_tool_call( self, diff --git a/app/agent/middleware/subagents.py b/app/agent/middleware/subagents.py index 3bec10f91..437389c21 100644 --- a/app/agent/middleware/subagents.py +++ b/app/agent/middleware/subagents.py @@ -36,6 +36,7 @@ from app.agent.policy import ( ) from app.agent.runtime import SubAgentDefinition, agent_runtime_manager from app.agent.tools.tags import ToolTag +from app.agent.tools.catalog import ToolCatalogSnapshot from app.log import logger @@ -418,6 +419,7 @@ class _SubAgentAgentProvider: tools: list[BaseTool], server_tools: Optional[list[dict[str, Any]]] = None, policy_context: Optional[ToolPolicyContext] = None, + catalog: Optional[ToolCatalogSnapshot] = None, ) -> None: """初始化子代理执行器。""" self._model = model @@ -425,6 +427,7 @@ class _SubAgentAgentProvider: self._tools = tools self._server_tools = server_tools or [] self._policy_context = policy_context or _default_subagent_policy_context(tools) + self._catalog = catalog self._agents = {} self._default_agent_name = "general-purpose" @@ -442,6 +445,9 @@ class _SubAgentAgentProvider: return profile.name, cached_agent subagent_tools = _select_tools(self._tools, profile) + subagent_catalog = ( + self._catalog.select(subagent_tools) if self._catalog is not None else None + ) logger.info( f"创建子代理图: subagent_type={profile.name}, tools={len(subagent_tools)}" ) @@ -450,7 +456,12 @@ class _SubAgentAgentProvider: tools=[*subagent_tools, *self._server_tools], system_prompt=profile.prompt, name=profile.name, - middleware=[AgentPolicyMiddleware(context=self._policy_context)], + middleware=[ + AgentPolicyMiddleware( + context=self._policy_context, + catalog=subagent_catalog, + ) + ], ) self._agents[profile.name] = agent return profile.name, agent @@ -511,6 +522,7 @@ class MoviePilotSubAgentMiddleware(AgentMiddleware): task_description: str = SUBAGENT_TASK_DESCRIPTION, stream_handler: Any = None, policy_context: Optional[ToolPolicyContext] = None, + catalog: Optional[ToolCatalogSnapshot] = None, ) -> None: """初始化同步子代理中间件。""" self.system_prompt = system_prompt @@ -521,6 +533,7 @@ class MoviePilotSubAgentMiddleware(AgentMiddleware): tools=tools, server_tools=server_tools, policy_context=policy_context, + catalog=catalog, ) self.tools = [ StructuredTool.from_function( @@ -608,6 +621,7 @@ class SubAgentTaskControlMiddleware(AgentMiddleware): task_description: str = SUBAGENT_CONTROL_DESCRIPTION, stream_handler: Any = None, policy_context: Optional[ToolPolicyContext] = None, + catalog: Optional[ToolCatalogSnapshot] = None, ) -> None: """初始化异步子代理调度中间件。""" self.stream_handler = stream_handler @@ -617,6 +631,7 @@ class SubAgentTaskControlMiddleware(AgentMiddleware): tools=tools, server_tools=server_tools, policy_context=policy_context, + catalog=catalog, ) self._semaphore = asyncio.Semaphore(SUBAGENT_MAX_CONCURRENT_TASKS) self._tasks: dict[str, _SubAgentRuntimeTask] = {} @@ -1193,6 +1208,7 @@ def create_subagent_middlewares( server_tools: Optional[list[dict[str, Any]]] = None, stream_handler: Any = None, policy_context: Optional[ToolPolicyContext] = None, + catalog: Optional[ToolCatalogSnapshot] = None, ) -> tuple[list[AgentMiddleware], list[BaseTool]]: """创建子代理中间件列表和任务工具列表。""" runtime_signature = agent_runtime_manager.current_signature() @@ -1204,6 +1220,7 @@ def create_subagent_middlewares( server_tools=server_tools or [], stream_handler=stream_handler, policy_context=policy_context, + catalog=catalog, ) control_middleware = SubAgentTaskControlMiddleware( model=model, @@ -1212,6 +1229,7 @@ def create_subagent_middlewares( server_tools=server_tools or [], stream_handler=stream_handler, policy_context=policy_context, + catalog=catalog, ) task_tools = [ diff --git a/app/agent/policy/__init__.py b/app/agent/policy/__init__.py index 393700215..dca544ad4 100644 --- a/app/agent/policy/__init__.py +++ b/app/agent/policy/__init__.py @@ -4,9 +4,14 @@ from app.agent.policy.contracts import ( ActionEffect, ActionPolicy, AuthSource, + CanonicalInvocation, + ConversationKind, + DeliveryTarget, ConfirmationMode, ExecutionOutcome, ExecutionReceipt, + InboundEnvelope, + InboundProvenance, MigrationState, PolicyDecision, PolicyObservation, @@ -14,10 +19,14 @@ from app.agent.policy.contracts import ( PrincipalRole, PrincipalType, RecoveryMode, + RECEIPT_STATE_TRANSITIONS, + ReceiptState, ResultSensitivity, + TERMINAL_RECEIPT_STATES, ToolInvocation, ToolOrigin, ToolPolicyContext, + ToolRevision, ) from app.agent.policy.orchestrator import ( DEFAULT_TOOL_POLICY_ORCHESTRATOR, @@ -39,11 +48,16 @@ __all__ = [ "ActionPolicy", "AgentToolPolicyOrchestrator", "AuthSource", + "CanonicalInvocation", + "ConversationKind", + "DeliveryTarget", "ConfirmationMode", "DEFAULT_TOOL_POLICY_ORCHESTRATOR", "DEFAULT_TOOL_POLICY_REGISTRY", "ExecutionOutcome", "ExecutionReceipt", + "InboundEnvelope", + "InboundProvenance", "MigrationState", "PolicyDecision", "PolicyObservation", @@ -52,10 +66,14 @@ __all__ = [ "PrincipalType", "REDACTED_VALUE", "RecoveryMode", + "RECEIPT_STATE_TRANSITIONS", + "ReceiptState", "ResultSensitivity", + "TERMINAL_RECEIPT_STATES", "ToolInvocation", "ToolOrigin", "ToolPolicyContext", + "ToolRevision", "ToolPolicyRegistry", "call_policy_hook", "sanitize_for_host", diff --git a/app/agent/policy/canonical.py b/app/agent/policy/canonical.py new file mode 100644 index 000000000..d23c60488 --- /dev/null +++ b/app/agent/policy/canonical.py @@ -0,0 +1,130 @@ +"""严格工具调用规范化，不读取设置值或执行工具实现。""" + +import hashlib +import json +from typing import Any, Mapping + +from pydantic import BaseModel, ValidationError + +from app.agent.policy.contracts import ( + ActionPolicy, + CanonicalInvocation, + ToolRevision, +) +from app.agent.tools.impl._system_setting_utils import ( + list_setting_specs, + resolve_setting_spec, +) + + +class CanonicalizationError(ValueError): + """当前工具 schema 或静态前提无法产生严格调用时的稳定失败。""" + + +def _stable_json(value: Any) -> str: + """生成拒绝 NaN 且保留 Unicode 的稳定紧凑 JSON。""" + try: + return json.dumps( + value, + ensure_ascii=False, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ) + except (TypeError, ValueError, ValidationError) as error: + raise CanonicalizationError("调用参数无法规范化") from error + + +def _accepted_argument_names(args_schema: type[BaseModel]) -> set[str]: + """返回严格入口可接受的字段名与字符串别名。""" + accepted = set(args_schema.model_fields) + for field in args_schema.model_fields.values(): + for alias in (field.alias, field.validation_alias): + if isinstance(alias, str): + accepted.add(alias) + return accepted + + +def canonicalize_invocation( + *, + tool: Any, + arguments: Mapping[str, Any], + policy: ActionPolicy, + tool_revision: ToolRevision, +) -> CanonicalInvocation: + """使用当前 Pydantic schema 与静态设置定义生成不可变调用摘要。""" + tool_name = str(getattr(tool, "name", "") or "") + args_schema = getattr(tool, "args_schema", None) + if not tool_name or not isinstance(args_schema, type) or not issubclass( + args_schema, BaseModel + ): + raise CanonicalizationError("工具缺少严格 Pydantic 参数契约") + raw_arguments = dict(arguments or {}) + unknown_arguments = set(raw_arguments) - _accepted_argument_names(args_schema) + if unknown_arguments: + raise CanonicalizationError("工具参数校验失败") + try: + validated = args_schema.model_validate(raw_arguments) + normalized = validated.model_dump( + mode="json", + exclude_unset=False, + exclude_none=False, + ) + schema_json = _stable_json(args_schema.model_json_schema()) + except (TypeError, ValueError) as error: + raise CanonicalizationError("工具参数校验失败") from error + + preconditions: tuple[tuple[str, str], ...] = () + setting_key = normalized.get("setting_key") + if tool_name == "query_system_settings": + if setting_key: + spec = resolve_setting_spec(setting_key) + if spec is None: + raise CanonicalizationError("系统设置项不存在") + specs = [spec] + else: + try: + specs = list_setting_specs( + group=normalized.get("group"), + keyword=normalized.get("keyword"), + ) + except ValueError as error: + raise CanonicalizationError("系统设置选择器无效") from error + if not specs: + raise CanonicalizationError("系统设置选择器没有匹配项") + preconditions = tuple( + ( + "setting", + f"{spec.source}:{spec.key}:{spec.group}", + ) + for spec in specs + ) + + payload = { + "canonical_version": "p1-g2a2-canonical-v1", + "action_subtype": policy.effect.value, + "arguments": normalized, + "policy_version": policy.policy_version, + "preconditions": preconditions, + "schema_digest": hashlib.sha256(schema_json.encode("utf-8")).hexdigest(), + "tool_name": tool_name, + "tool_revision": { + "factory": tool_revision.factory, + "implementation": tool_revision.implementation, + "plugin": tool_revision.plugin, + }, + } + canonical_json = _stable_json(payload) + return CanonicalInvocation( + tool_name=tool_name, + arguments=normalized, + canonical_json=canonical_json, + digest=hashlib.sha256(canonical_json.encode("utf-8")).hexdigest(), + policy_version=policy.policy_version, + tool_revision=tool_revision, + schema_digest=payload["schema_digest"], + preconditions=preconditions, + ) + + +__all__ = ["CanonicalizationError", "canonicalize_invocation"] diff --git a/app/agent/policy/contracts.py b/app/agent/policy/contracts.py index 1541cd08d..f39e43a61 100644 --- a/app/agent/policy/contracts.py +++ b/app/agent/policy/contracts.py @@ -2,9 +2,23 @@ from dataclasses import dataclass, field from enum import Enum +from types import MappingProxyType from typing import Any, Mapping, MutableMapping, Optional +def _freeze_contract_value(value: Any) -> Any: + """递归冻结确认边界中的容器，避免等待期间被调用方修改。""" + if isinstance(value, Mapping): + return MappingProxyType( + {str(key): _freeze_contract_value(item) for key, item in value.items()} + ) + if isinstance(value, (list, tuple)): + return tuple(_freeze_contract_value(item) for item in value) + if isinstance(value, set): + return frozenset(_freeze_contract_value(item) for item in value) + return value + + class ToolOrigin(str, Enum): """工具调用的宿主可信入口。""" @@ -35,6 +49,24 @@ class AuthSource(str, Enum): AGENT_TOKEN = "agent_token" +class InboundProvenance(str, Enum): + """入站事实经宿主验证后的可信等级。""" + + WEB_SESSION = "web_session" + VERIFIED_ADAPTER = "verified_adapter" + ADMIN_INTEGRATION = "admin_integration" + UNTRUSTED = "untrusted" + + +class ConversationKind(str, Enum): + """目标会话的隐私边界。""" + + WEB = "web" + PRIVATE = "private" + GROUP = "group" + UNKNOWN = "unknown" + + class PrincipalRole(str, Enum): """策略授权使用的角色层级。""" @@ -99,6 +131,79 @@ class ExecutionOutcome(str, Enum): FAILED = "failed" +class ReceiptState(str, Enum): + """严格执行回执的单向生命周期状态。""" + + WAITING_CONFIRMATION = "waiting_confirmation" + VALIDATING = "validating" + EXECUTING = "executing" + DELIVERING = "delivering" + SUCCEEDED = "succeeded" + CANCELLED = "cancelled" + EXPIRED = "expired" + EXPIRED_RESTART = "expired_restart" + EXPIRED_ORPHANED = "expired_orphaned" + PREPARATION_FAILED = "preparation_failed" + PROMPT_DELIVERY_FAILED = "prompt_delivery_failed" + VALIDATION_FAILED = "validation_failed" + VALIDATION_RECORD_FAILED = "validation_record_failed" + EXECUTION_FAILED = "execution_failed" + DELIVERY_FAILED = "delivery_failed" + UNKNOWN_AFTER_RESTART = "unknown_after_restart" + UNKNOWN_ORPHANED = "unknown_orphaned" + + +TERMINAL_RECEIPT_STATES = frozenset({ + ReceiptState.SUCCEEDED, + ReceiptState.CANCELLED, + ReceiptState.EXPIRED, + ReceiptState.EXPIRED_RESTART, + ReceiptState.EXPIRED_ORPHANED, + ReceiptState.PREPARATION_FAILED, + ReceiptState.PROMPT_DELIVERY_FAILED, + ReceiptState.VALIDATION_FAILED, + ReceiptState.VALIDATION_RECORD_FAILED, + ReceiptState.EXECUTION_FAILED, + ReceiptState.DELIVERY_FAILED, + ReceiptState.UNKNOWN_AFTER_RESTART, + ReceiptState.UNKNOWN_ORPHANED, +}) + + +RECEIPT_STATE_TRANSITIONS = { + ReceiptState.WAITING_CONFIRMATION: frozenset({ + ReceiptState.VALIDATING, + ReceiptState.CANCELLED, + ReceiptState.EXPIRED, + ReceiptState.EXPIRED_RESTART, + ReceiptState.EXPIRED_ORPHANED, + ReceiptState.PREPARATION_FAILED, + ReceiptState.PROMPT_DELIVERY_FAILED, + }), + ReceiptState.VALIDATING: frozenset({ + ReceiptState.EXECUTING, + ReceiptState.CANCELLED, + ReceiptState.EXPIRED, + ReceiptState.EXPIRED_RESTART, + ReceiptState.EXPIRED_ORPHANED, + ReceiptState.VALIDATION_FAILED, + ReceiptState.VALIDATION_RECORD_FAILED, + }), + ReceiptState.EXECUTING: frozenset({ + ReceiptState.DELIVERING, + ReceiptState.EXECUTION_FAILED, + ReceiptState.UNKNOWN_AFTER_RESTART, + ReceiptState.UNKNOWN_ORPHANED, + }), + ReceiptState.DELIVERING: frozenset({ + ReceiptState.SUCCEEDED, + ReceiptState.DELIVERY_FAILED, + ReceiptState.UNKNOWN_AFTER_RESTART, + ReceiptState.UNKNOWN_ORPHANED, + }), +} + + @dataclass(frozen=True) class PolicyPrincipal: """由可信入口建立、不可由工具参数覆盖的调用主体。""" @@ -124,6 +229,62 @@ class ToolInvocation: source: Optional[str] = None +@dataclass(frozen=True) +class DeliveryTarget: + """受保护结果的精确宿主路由，actor 与 recipient 不得互相替代。""" + + channel: str + source_instance_id: str + tenant_or_account_id: str + conversation_kind: ConversationKind + conversation_id: str + recipient_id: str + actor_id: str + server_session_id: str + + +@dataclass(frozen=True) +class InboundEnvelope: + """由可信宿主入口创建的不可变入站事实。""" + + provenance: InboundProvenance + target: DeliveryTarget + inbound_event_id: str + raw_text: str = field(repr=False) + normalized_text: str = field(repr=False) + has_images: bool = False + has_audio: bool = False + has_files: bool = False + is_callback: bool = False + + +@dataclass(frozen=True) +class ToolRevision: + """严格调用绑定的工具实现、工厂和插件目录版本。""" + + implementation: str + factory: str + plugin: str + + +@dataclass(frozen=True) +class CanonicalInvocation: + """经当前工具 schema 校验、可绑定确认与版本前提的调用。""" + + tool_name: str + arguments: Mapping[str, Any] = field(repr=False) + canonical_json: str = field(repr=False) + digest: str + policy_version: str + tool_revision: ToolRevision + schema_digest: str + preconditions: tuple[tuple[str, str], ...] = () + + def __post_init__(self) -> None: + """阻止调用方在确认等待期间修改规范化参数。""" + object.__setattr__(self, "arguments", _freeze_contract_value(self.arguments)) + + @dataclass(frozen=True) class ActionPolicy: """参数级动作策略及其兼容迁移状态。""" @@ -230,9 +391,14 @@ __all__ = [ "ActionEffect", "ActionPolicy", "AuthSource", + "CanonicalInvocation", "ConfirmationMode", + "ConversationKind", + "DeliveryTarget", "ExecutionOutcome", "ExecutionReceipt", + "InboundEnvelope", + "InboundProvenance", "MigrationState", "PolicyDecision", "PolicyObservation", @@ -240,8 +406,12 @@ __all__ = [ "PrincipalRole", "PrincipalType", "RecoveryMode", + "RECEIPT_STATE_TRANSITIONS", + "ReceiptState", "ResultSensitivity", + "TERMINAL_RECEIPT_STATES", "ToolInvocation", "ToolOrigin", "ToolPolicyContext", + "ToolRevision", ] diff --git a/app/agent/policy/orchestrator.py b/app/agent/policy/orchestrator.py index 2756a18ce..8ff8868d6 100644 --- a/app/agent/policy/orchestrator.py +++ b/app/agent/policy/orchestrator.py @@ -9,6 +9,7 @@ from langchain_core.messages import ToolMessage from pydantic import ValidationError from app.agent.policy.contracts import ( + ConfirmationMode, ExecutionOutcome, ExecutionReceipt, MigrationState, @@ -101,6 +102,14 @@ class AgentToolPolicyOrchestrator: shadow=True, reason_code="legacy_shadow_allow", ) + elif policy.confirmation is ConfirmationMode.REQUIRED: + # 严格运行时接管前保持既有调用能力，但不得把敏感动作记为安全读取。 + decision = PolicyDecision( + allowed=True, + confirmation_required=False, + shadow=True, + reason_code="strict_runtime_pending", + ) else: decision = PolicyDecision( allowed=True, diff --git a/app/agent/policy/registry.py b/app/agent/policy/registry.py index a43d8396e..9e06e4de8 100644 --- a/app/agent/policy/registry.py +++ b/app/agent/policy/registry.py @@ -140,10 +140,25 @@ class ToolPolicyRegistry: requires_admin: bool, ) -> ActionPolicy: """根据工具名和宿主权限声明解析当前迁移策略。""" - del arguments # 参数级迁移由后续领域 Goal 逐项加入。 required_role = ( PrincipalRole.SYSTEM_ADMIN if requires_admin else PrincipalRole.USER ) + if ( + tool_name == "query_system_settings" + and arguments.get("show_secrets") is True + ): + return ActionPolicy( + effect=ActionEffect.SENSITIVE_READ, + required_role=PrincipalRole.SYSTEM_ADMIN, + confirmation=ConfirmationMode.REQUIRED, + recovery=RecoveryMode.NONE, + result_sensitivity=ResultSensitivity.SECRET, + migration_state=MigrationState.ENFORCED, + policy_version="p1-g2a2-v1", + machine_allowed=True, + background_allowed=False, + subagent_allowed=False, + ) if tool_name in self._safe_read_tool_names: return ActionPolicy( effect=ActionEffect.SAFE_READ, diff --git a/app/agent/tools/catalog.py b/app/agent/tools/catalog.py new file mode 100644 index 000000000..2b5588454 --- /dev/null +++ b/app/agent/tools/catalog.py @@ -0,0 +1,189 @@ +"""Agent 本地工具目录的不可变快照与严格解析。""" + +import hashlib +import json +from dataclasses import dataclass, field +from types import MappingProxyType +from typing import Any, Mapping, Optional + +from pydantic import BaseModel + +from app.agent.policy import ToolRevision + + +class ToolCatalogError(RuntimeError): + """工具目录无法建立可信当前视图时的稳定失败。""" + + +class ToolIdentityAmbiguousError(ToolCatalogError): + """同一工具名对应多个实现，无法进行严格解析。""" + + +def _stable_json(value: Any) -> str: + """生成工具身份摘要使用的稳定 JSON。""" + return json.dumps( + value, + ensure_ascii=False, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ) + + +def _schema_digest(tool: Any) -> str: + """计算工具当前 Pydantic 参数契约摘要。""" + args_schema = getattr(tool, "args_schema", None) + if isinstance(args_schema, type) and issubclass(args_schema, BaseModel): + schema = args_schema.model_json_schema() + elif isinstance(args_schema, Mapping): + schema = dict(args_schema) + else: + schema = {"type": "object", "properties": {}} + return hashlib.sha256(_stable_json(schema).encode("utf-8")).hexdigest() + + +def _implementation_identity(tool: Any) -> str: + """返回不依赖对象地址、可区分动态绑定的工具实现身份。""" + tool_class = type(tool) + implementation = f"{tool_class.__module__}.{tool_class.__qualname__}" + binding = str(getattr(tool, "_agent_tool_binding", "") or "") + return f"{implementation}:{binding}" if binding else implementation + + +@dataclass(frozen=True) +class ToolCatalogEntry: + """绑定一次目录构造中精确工具实例的身份记录。""" + + name: str + source: str + identity: str + description_digest: str + schema_digest: str + revision: ToolRevision + tool: Any = field(repr=False, compare=False) + + +@dataclass(frozen=True) +class ToolCatalogSnapshot: + """同一图构造与严格执行共享的本地工具事实源。""" + + entries: tuple[ToolCatalogEntry, ...] + plugin_revision: int + factory_revision: str + _by_name: Mapping[str, tuple[ToolCatalogEntry, ...]] = field( + init=False, + repr=False, + compare=False, + ) + + def __post_init__(self) -> None: + """建立只读名称索引并保留所有冲突项。""" + by_name: dict[str, list[ToolCatalogEntry]] = {} + for entry in self.entries: + by_name.setdefault(entry.name, []).append(entry) + object.__setattr__( + self, + "_by_name", + MappingProxyType( + {name: tuple(matches) for name, matches in by_name.items()} + ), + ) + + @classmethod + def from_tools( + cls, + tools: list[Any], + *, + plugin_revision: int, + factory_revision: str, + ) -> "ToolCatalogSnapshot": + """从已完成上下文注入的精确实例建立不可变目录。""" + entries = [] + for tool in tools: + name = str(getattr(tool, "name", "") or "") + if not name: + raise ToolCatalogError("工具缺少稳定名称") + source = str(getattr(tool, "_agent_tool_source", "builtin")) + schema_digest = _schema_digest(tool) + description_digest = hashlib.sha256( + str(getattr(tool, "description", "") or "").encode("utf-8") + ).hexdigest() + implementation = _implementation_identity(tool) + revision = ToolRevision( + implementation=implementation, + factory=factory_revision, + plugin=str(plugin_revision), + ) + entries.append( + ToolCatalogEntry( + name=name, + source=source, + identity=f"{source}:{implementation}:{schema_digest}", + description_digest=description_digest, + schema_digest=schema_digest, + revision=revision, + tool=tool, + ) + ) + return cls( + entries=tuple(entries), + plugin_revision=plugin_revision, + factory_revision=factory_revision, + ) + + @property + def tools(self) -> list[Any]: + """按目录顺序返回精确工具实例。""" + return [entry.tool for entry in self.entries] + + @property + def collisions(self) -> Mapping[str, tuple[ToolCatalogEntry, ...]]: + """返回所有同名工具，不按注册顺序隐式选胜者。""" + return MappingProxyType( + {name: entries for name, entries in self._by_name.items() if len(entries) > 1} + ) + + @property + def signature(self) -> tuple[Any, ...]: + """返回可参与 Agent 图缓存的完整目录签名。""" + return ( + self.factory_revision, + self.plugin_revision, + tuple( + ( + entry.name, + entry.identity, + entry.description_digest, + entry.schema_digest, + ) + for entry in self.entries + ), + ) + + def resolve_unique(self, name: str) -> Optional[ToolCatalogEntry]: + """严格解析当前唯一实现；重名时拒绝继承 first/last-wins。""" + entries = self._by_name.get(name, ()) + if len(entries) > 1: + raise ToolIdentityAmbiguousError("TOOL_IDENTITY_AMBIGUOUS") + return entries[0] if entries else None + + def select(self, tools: list[Any]) -> "ToolCatalogSnapshot": + """为子图保留所选工具名的全部候选身份与 revision 语义。""" + selected_names = { + str(getattr(tool, "name", "") or "") for tool in tools + } + return ToolCatalogSnapshot( + entries=tuple( + entry for entry in self.entries if entry.name in selected_names + ), + plugin_revision=self.plugin_revision, + factory_revision=self.factory_revision, + ) + + +__all__ = [ + "ToolCatalogEntry", + "ToolCatalogError", + "ToolCatalogSnapshot", + "ToolIdentityAmbiguousError", +] diff --git a/app/agent/tools/factory.py b/app/agent/tools/factory.py index 338e6fb7a..520a62403 100644 --- a/app/agent/tools/factory.py +++ b/app/agent/tools/factory.py @@ -1,3 +1,5 @@ +import hashlib + from typing import Callable, List, Optional, Type from app.agent.tools.impl.add_download_tasks import AddDownloadTasksTool @@ -90,6 +92,7 @@ from app.log import logger from app.schemas.message import ChannelCapabilityManager from app.schemas.types import MessageChannel from .base import MoviePilotTool +from .catalog import ToolCatalogError, ToolCatalogSnapshot class MoviePilotToolFactory: @@ -195,6 +198,17 @@ class MoviePilotToolFactory: "query_agent_tasks", ) + CATALOG_BUILD_MAX_ATTEMPTS = 3 + + @classmethod + def catalog_factory_revision(cls) -> str: + """返回当前内置工具工厂定义的稳定摘要。""" + identities = ( + f"{tool_class.__module__}.{tool_class.__qualname__}" + for tool_class in cls.BUILTIN_TOOL_CLASSES + ) + return hashlib.sha256("\n".join(identities).encode("utf-8")).hexdigest() + @staticmethod def _should_enable_choice_tool(channel: Optional[str] = None) -> bool: if not channel: @@ -266,6 +280,7 @@ class MoviePilotToolFactory: tool.set_message_attr(channel=channel, source=source, username=username) tool.set_stream_handler(stream_handler=stream_handler) tool.set_agent_context(agent_context=agent_context) + object.__setattr__(tool, "_agent_tool_source", "builtin") tools.append(tool) # 加载插件提供的工具 @@ -292,6 +307,11 @@ class MoviePilotToolFactory: ) tool.set_stream_handler(stream_handler=stream_handler) tool.set_agent_context(agent_context=agent_context) + object.__setattr__( + tool, + "_agent_tool_source", + f"plugin:{plugin_id or 'unknown'}", + ) tools.append(tool) plugin_tools_count += 1 logger.debug( @@ -310,3 +330,19 @@ class MoviePilotToolFactory: else: logger.debug(f"成功创建 {len(tools)} 个MoviePilot工具") return tools + + @classmethod + def create_catalog(cls, **tool_kwargs) -> ToolCatalogSnapshot: + """在插件目录稳定窗口内构造一份完整本地工具快照。""" + plugin_manager = PluginManager() + for _attempt in range(cls.CATALOG_BUILD_MAX_ATTEMPTS): + before_revision = plugin_manager.get_plugin_agent_tools_revision() + tools = cls.create_tools(**tool_kwargs) + after_revision = plugin_manager.get_plugin_agent_tools_revision() + if before_revision == after_revision: + return ToolCatalogSnapshot.from_tools( + tools, + plugin_revision=after_revision, + factory_revision=cls.catalog_factory_revision(), + ) + raise ToolCatalogError("插件工具目录持续变化，无法建立当前快照") diff --git a/app/agent/tools/impl/mcp.py b/app/agent/tools/impl/mcp.py index 2fdf617ad..4df0eac98 100644 --- a/app/agent/tools/impl/mcp.py +++ b/app/agent/tools/impl/mcp.py @@ -1,5 +1,6 @@ """外部 MCP 工具适配器。""" +import hashlib import json from typing import Any, Optional @@ -77,6 +78,22 @@ class McpExternalTool(MoviePilotTool): return json.dumps(result, ensure_ascii=False, indent=2, default=str) +def _mcp_binding_identity(spec: AgentMcpToolSpec) -> str: + """生成不暴露端点配置的稳定 MCP 工具绑定身份。""" + payload = json.dumps( + { + "agent_tool_name": spec.agent_tool_name, + "server_id": spec.server.id, + "tool_name": spec.name, + }, + ensure_ascii=False, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + async def create_external_mcp_tools( *, session_id: str, @@ -86,13 +103,41 @@ async def create_external_mcp_tools( username: Optional[str] = None, stream_handler=None, agent_context: Optional[dict] = None, + specs: Optional[list[AgentMcpToolSpec]] = None, ) -> list[McpExternalTool]: """创建当前已启用的外部 MCP Agent 工具列表。""" tools = [] - for spec in await agent_mcp_manager.list_enabled_tool_specs(): + current_specs = specs + if current_specs is None: + current_specs = await agent_mcp_manager.list_enabled_tool_specs() + for spec in current_specs: tool = McpExternalTool(spec=spec, session_id=session_id, user_id=user_id) tool.set_message_attr(channel=channel, source=source, username=username) tool.set_stream_handler(stream_handler=stream_handler) tool.set_agent_context(agent_context=agent_context) + object.__setattr__( + tool, + "_agent_tool_source", + f"mcp:{spec.server.id}", + ) + object.__setattr__( + tool, + "_agent_tool_binding", + _mcp_binding_identity(spec), + ) tools.append(tool) return tools + + +def select_legacy_mcp_tools( + tools: list[McpExternalTool], +) -> list[McpExternalTool]: + """保留跨服务器同名工具历史上的 first-wins 执行顺序。""" + selected = [] + seen_names = set() + for tool in tools: + if tool.name in seen_names: + continue + selected.append(tool) + seen_names.add(tool.name) + return selected diff --git a/app/agent/tools/manager.py b/app/agent/tools/manager.py index 3249ff64b..98fc68ecb 100644 --- a/app/agent/tools/manager.py +++ b/app/agent/tools/manager.py @@ -15,6 +15,7 @@ from app.agent.policy import ( ) from app.agent.tools.base import ToolExecutionTimeoutError, format_tool_result_for_agent from app.agent.tools.factory import MoviePilotToolFactory +from app.agent.tools.catalog import ToolCatalogSnapshot from app.core.plugin import PluginManager from app.log import logger @@ -66,6 +67,7 @@ class MoviePilotToolsManager: agent_context={"is_admin": is_admin}, ) self.tools: List[Any] = [] + self.catalog: Optional[ToolCatalogSnapshot] = None self._tools_lock = threading.Lock() self._plugin_agent_tools_revision = -1 self._load_tools() @@ -75,31 +77,23 @@ class MoviePilotToolsManager: 加载所有MoviePilot工具 """ try: - plugin_manager = PluginManager() - while True: - plugin_tools_revision = ( - plugin_manager.get_plugin_agent_tools_revision() - ) - tools = MoviePilotToolFactory.create_tools( - session_id=self.session_id, - user_id=self.user_id, - channel=None, - source="api", - username="API Client", - stream_handler=None, - agent_context={"is_admin": self.is_admin}, - ) - if ( - plugin_tools_revision - == plugin_manager.get_plugin_agent_tools_revision() - ): - break - self.tools = tools - self._plugin_agent_tools_revision = plugin_tools_revision + catalog = MoviePilotToolFactory.create_catalog( + session_id=self.session_id, + user_id=self.user_id, + channel=None, + source="api", + username="API Client", + stream_handler=None, + agent_context={"is_admin": self.is_admin}, + ) + self.catalog = catalog + self.tools = catalog.tools + self._plugin_agent_tools_revision = catalog.plugin_revision logger.info(f"成功加载 {len(self.tools)} 个工具") except Exception as e: logger.error(f"加载工具失败: {summarize_error(e)}") self.tools = [] + self.catalog = None self._plugin_agent_tools_revision = -1 def _ensure_tools_current(self) -> None: @@ -162,10 +156,26 @@ class MoviePilotToolsManager: 工具实例，如果未找到返回None """ self._ensure_tools_current() - for tool in self.tools: - if tool.name == tool_name: - return tool - return None + return next( + (tool for tool in self.tools if tool.name == tool_name), + None, + ) + + def get_strict_tool(self, tool_name: str) -> Optional[Any]: + """按当前目录唯一身份解析严格调用，重名时稳定失败。""" + self._ensure_tools_current() + if self.catalog is None or [ + id(tool) for tool in self.catalog.tools + ] != [id(tool) for tool in self.tools]: + self.catalog = ToolCatalogSnapshot.from_tools( + self.tools, + plugin_revision=self._plugin_agent_tools_revision, + factory_revision=MoviePilotToolFactory.catalog_factory_revision(), + ) + if self.catalog is None: + return None + entry = self.catalog.resolve_unique(tool_name) + return entry.tool if entry else None @staticmethod def _resolve_field_schema(field_info: Dict[str, Any]) -> Dict[str, Any]: diff --git a/app/core/plugin.py b/app/core/plugin.py index ff5101a4c..f15bcc4fc 100644 --- a/app/core/plugin.py +++ b/app/core/plugin.py @@ -41,6 +41,7 @@ from app.utils.system import SystemUtils class PluginManager(ConfigReloadMixin, metaclass=Singleton): """插件管理器""" CONFIG_WATCH = {"DEV", "PLUGIN_AUTO_RELOAD", "PLUGIN_LOCAL_REPO_PATHS"} + AGENT_TOOLS_BUILD_MAX_ATTEMPTS = 3 def __init__(self): # 插件列表 @@ -1011,7 +1012,7 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): }] """ cache_key = pid or "__all__" - while True: + for _attempt in range(self.AGENT_TOOLS_BUILD_MAX_ATTEMPTS): with self._plugin_agent_tools_cache_lock: cache_revision = self._plugin_agent_tools_revision cached_tools = self._plugin_agent_tools_cache.get(cache_key) @@ -1047,6 +1048,7 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton): ret_tools ) return ret_tools + raise RuntimeError("插件工具注册表持续变化，无法建立当前快照") @staticmethod def get_plugin_remote_entry(plugin_id: str, dist_path: str) -> str: diff --git a/tests/test_agent_graph_cache.py b/tests/test_agent_graph_cache.py index 88062cfb2..725cbf8d7 100644 --- a/tests/test_agent_graph_cache.py +++ b/tests/test_agent_graph_cache.py @@ -1,6 +1,7 @@ """Agent 图缓存行为测试。""" -from datetime import datetime +from contextlib import ExitStack +from datetime import datetime, timedelta from types import SimpleNamespace from unittest.mock import AsyncMock, patch @@ -8,7 +9,15 @@ import pytest from langchain_core.messages import AIMessage, HumanMessage from app.agent import MoviePilotAgent, ReplyMode, _CompiledAgentBundle +from app.agent.mcp import AgentMcpToolSpec +from app.agent.middleware.policy import AgentPolicyMiddleware +from app.agent.tools.catalog import ( + ToolCatalogSnapshot, + ToolIdentityAmbiguousError, +) +from app.agent.tools.impl.mcp import create_external_mcp_tools from app.core.config import settings +from app.schemas.agent import AgentMcpServerConfig @pytest.fixture @@ -45,18 +54,32 @@ class _CapturingAgent: async def test_create_agent_reuses_cached_graph_when_signature_matches(): """构造签名一致时应直接复用已编译 Agent 图。""" cached_graph = object() + catalog = ToolCatalogSnapshot.from_tools( + [], plugin_revision=0, factory_revision="factory-v1" + ) agent = MoviePilotAgent(session_id="cache-hit", user_id="user-1") agent._compiled_agent_bundle = _CompiledAgentBundle( signature=("sig",), agent=cached_graph, streaming=False, created_at=datetime.now(), + tool_catalog=catalog, + subagent_catalog=catalog, + plugin_revision=0, + mcp_config_signature="mcp-config", + catalog_checked_at=datetime.now(), ) with patch.object( agent, "_agent_bundle_signature", new=AsyncMock(return_value=("sig",)), + ), patch( + "app.agent.PluginManager.get_plugin_agent_tools_revision", + return_value=0, + ), patch( + "app.agent.agent_mcp_manager.config_signature", + return_value="mcp-config", ), patch("app.agent.create_agent") as create_agent: graph = await agent._create_agent(streaming=False) @@ -65,6 +88,130 @@ async def test_create_agent_reuses_cached_graph_when_signature_matches(): create_agent.assert_not_called() +@pytest.mark.anyio +async def test_fresh_catalog_cache_hit_skips_tool_and_mcp_discovery() -> None: + """目录仍在 freshness 窗口内时，缓存命中不得重建工具或访问 MCP。""" + cached_graph = object() + catalog = ToolCatalogSnapshot.from_tools( + [], plugin_revision=0, factory_revision="factory-v1" + ) + agent = MoviePilotAgent(session_id="catalog-cache-hit", user_id="user-1") + agent._compiled_agent_bundle = _CompiledAgentBundle( + signature=("sig",), + agent=cached_graph, + streaming=False, + created_at=datetime.now(), + tool_catalog=catalog, + subagent_catalog=catalog, + plugin_revision=0, + mcp_config_signature="mcp-config", + catalog_checked_at=datetime.now(), + ) + + with patch.object( + agent, + "_agent_bundle_signature", + new=AsyncMock(return_value=("sig",)), + ), patch.object( + agent, + "_initialize_local_tool_catalogs", + side_effect=AssertionError("tool catalog rebuilt"), + ), patch( + "app.agent.PluginManager.get_plugin_agent_tools_revision", + return_value=0, + ), patch( + "app.agent.agent_mcp_manager.config_signature", + return_value="mcp-config", + ), patch( + "app.agent.agent_mcp_manager.list_enabled_tool_specs", + new=AsyncMock(side_effect=AssertionError("MCP discovery called")), + ): + graph = await agent._create_agent(streaming=False) + + assert graph is cached_graph + assert agent._last_agent_cache_hit is True + + +@pytest.mark.anyio +async def test_expired_unchanged_catalog_renews_freshness() -> None: + """过期目录复核后若签名未变，应续期缓存而不是每轮重复 discovery。""" + cached_graph = object() + catalog = ToolCatalogSnapshot.from_tools( + [], plugin_revision=0, factory_revision="factory-v1" + ) + expired_at = datetime.now() - timedelta(minutes=5) + agent = MoviePilotAgent(session_id="catalog-refresh", user_id="user-1") + agent._compiled_agent_bundle = _CompiledAgentBundle( + signature=("sig",), + agent=cached_graph, + streaming=False, + created_at=datetime.now(), + tool_catalog=catalog, + subagent_catalog=catalog, + plugin_revision=0, + mcp_config_signature="mcp-config", + catalog_checked_at=expired_at, + ) + fake_llm = SimpleNamespace( + _llm_type="openai-chat", + model="fake", + profile={"max_input_tokens": 64000}, + ) + + with patch.object( + agent, + "_agent_bundle_signature", + new=AsyncMock(return_value=("sig",)), + ), patch.object( + agent, + "_initialize_local_tool_catalogs", + return_value=(catalog, catalog), + ), patch.object( + agent, + "_initialize_mcp_tools", + new=AsyncMock(return_value=[]), + ), patch.object( + agent, + "_initialize_subagent_mcp_tools", + new=AsyncMock(return_value=[]), + ), patch.object( + agent, + "_initialize_llm", + new=AsyncMock(return_value=fake_llm), + ), patch.object( + agent, + "_sync_model_profile", + ), patch( + "app.agent.ServerToolRegistry.resolve_web_search", + return_value=SimpleNamespace(use_local_web_search=True), + ), patch( + "app.agent.LLMHelper.get_server_tools", + return_value=[], + ), patch( + "app.agent.prompt_manager.get_agent_prompt", + return_value="prompt", + ), patch( + "app.agent.SkillsMiddleware", + return_value=SimpleNamespace(name="skills", tools=[]), + ), patch( + "app.agent.create_subagent_middlewares", + return_value=([], []), + ), patch( + "app.agent.PluginManager.get_plugin_agent_tools_revision", + return_value=0, + ), patch( + "app.agent.agent_mcp_manager.config_signature", + return_value="mcp-config", + ), patch( + "app.agent.agent_mcp_manager.list_enabled_tool_specs", + new=AsyncMock(return_value=[]), + ): + graph = await agent._create_agent(streaming=False) + + assert graph is cached_graph + assert agent._compiled_agent_bundle.catalog_checked_at > expired_at + + @pytest.mark.anyio async def test_agent_bundle_signature_changes_with_temperature(monkeypatch) -> None: """温度配置变化时应使会话内 Agent 图缓存失效。""" @@ -94,6 +241,256 @@ async def test_agent_bundle_signature_changes_with_temperature(monkeypatch) -> N assert updated_signature != initial_signature +@pytest.mark.anyio +async def test_agent_bundle_signature_changes_with_tool_catalog() -> None: + """工具目录 revision 必须参与会话内 Agent 图缓存签名。""" + agent = MoviePilotAgent(session_id="tool-revision", user_id="user-1") + runtime_config = {"provider": "openai", "model": "gpt-test"} + first_catalog = ToolCatalogSnapshot.from_tools( + [], plugin_revision=1, factory_revision="factory-v1" + ) + second_catalog = ToolCatalogSnapshot.from_tools( + [], plugin_revision=2, factory_revision="factory-v1" + ) + + with patch.object( + agent, + "_resolve_llm_runtime_config", + new=AsyncMock(return_value=runtime_config), + ): + first_signature = await agent._agent_bundle_signature( + streaming=False, + tool_catalog=first_catalog, + subagent_catalog=first_catalog, + ) + second_signature = await agent._agent_bundle_signature( + streaming=False, + tool_catalog=second_catalog, + subagent_catalog=second_catalog, + ) + + assert second_signature != first_signature + + +@pytest.mark.anyio +@pytest.mark.parametrize("max_tools", [0, 5]) +async def test_graph_keeps_mcp_first_winner_and_catalogs_all_collisions( + max_tools: int, +) -> None: + """主图与子图保持 MCP first-wins，严格目录覆盖全部客户端工具。""" + servers = [ + AgentMcpServerConfig( + id=server_id, + name="Shared Name", + transport="stdio", + command=server_id, + ) + for server_id in ("one", "two") + ] + specs = [ + AgentMcpToolSpec( + server=server, + name="echo", + agent_tool_name="shared_echo", + description="echo", + input_schema={"type": "object", "properties": {}}, + ) + for server in servers + ] + main_tools = await create_external_mcp_tools( + session_id="session", + user_id="user", + specs=specs, + ) + subagent_tools = await create_external_mcp_tools( + session_id="session", + user_id="user", + specs=specs, + ) + empty_catalog = ToolCatalogSnapshot.from_tools( + [], plugin_revision=0, factory_revision="factory-v1" + ) + fake_llm = SimpleNamespace( + _llm_type="openai-chat", + model="fake", + profile={"max_input_tokens": 64000}, + ) + skill_tool = SimpleNamespace( + name="shared_echo", + description="skill collision", + args_schema={"type": "object", "properties": {}}, + _agent_tool_source="middleware:skills", + ) + activity_tool = SimpleNamespace( + name="query_activity_log", + description="activity log", + args_schema={"type": "object", "properties": {}}, + _agent_tool_source="middleware:activity_log", + ) + subagent_task_tool = SimpleNamespace( + name="task", + description="subagent task", + args_schema={"type": "object", "properties": {}}, + _agent_tool_source="middleware:subagents", + ) + captured = {} + agent = MoviePilotAgent( + session_id="mcp-collision", + user_id="user", + channel="web", + source="test", + ) + + def _capture_subagents(**kwargs): + captured["subagent_tools"] = kwargs["tools"] + captured["subagent_catalog"] = kwargs["catalog"] + return [], [subagent_task_tool] + + def _capture_selector(**kwargs): + captured["selection_tools"] = kwargs["selection_tools"] + return SimpleNamespace(name="selector") + + def _capture_agent(**kwargs): + captured["agent_tools"] = kwargs["tools"] + captured["middlewares"] = kwargs["middleware"] + return object() + + patchers = [ + patch.object( + agent, + "_resolve_llm_runtime_config", + new=AsyncMock(return_value={"provider": "openai", "model": "fake"}), + ), + patch.object( + agent, + "_initialize_local_tool_catalogs", + return_value=(empty_catalog, empty_catalog), + ), + patch.object( + agent, + "_initialize_mcp_tools", + new=AsyncMock(return_value=main_tools), + ), + patch.object( + agent, + "_initialize_subagent_mcp_tools", + new=AsyncMock(return_value=subagent_tools), + ), + patch.object( + agent, + "_agent_bundle_signature", + new=AsyncMock(return_value=("mcp-collision", max_tools)), + ), + patch.object( + agent, + "_initialize_llm", + new=AsyncMock(return_value=fake_llm), + ), + patch.object(agent, "_sync_model_profile"), + patch( + "app.agent.PluginManager.get_plugin_agent_tools_revision", + return_value=0, + ), + patch( + "app.agent.agent_mcp_manager.config_signature", + return_value="mcp-config", + ), + patch( + "app.agent.agent_mcp_manager.list_enabled_tool_specs", + new=AsyncMock(return_value=specs), + ), + patch( + "app.agent.ServerToolRegistry.resolve_web_search", + return_value=SimpleNamespace(use_local_web_search=True), + ), + patch("app.agent.LLMHelper.get_server_tools", return_value=[]), + patch("app.agent.prompt_manager.get_agent_prompt", return_value="prompt"), + patch( + "app.agent.create_subagent_middlewares", + side_effect=_capture_subagents, + ), + patch( + "app.agent.MoviePilotToolFactory.get_tool_selector_always_include_names", + return_value=[], + ), + patch( + "app.agent.SkillsMiddleware", + return_value=SimpleNamespace(name="skills", tools=[skill_tool]), + ), + patch( + "app.agent.ActivityLogMiddleware", + return_value=SimpleNamespace(name="activity", tools=[activity_tool]), + ), + patch( + "app.agent.JobsMiddleware", + return_value=SimpleNamespace(name="jobs"), + ), + patch( + "app.agent.RuntimeConfigMiddleware", + return_value=SimpleNamespace(name="runtime"), + ), + patch( + "app.agent.MemoryMiddleware", + return_value=SimpleNamespace(name="memory"), + ), + patch( + "app.agent.SummarizationMiddleware", + return_value=SimpleNamespace(name="summary"), + ), + patch( + "app.agent.PatchToolCallsMiddleware", + return_value=SimpleNamespace(name="patch"), + ), + patch( + "app.agent.UsageMiddleware", + return_value=SimpleNamespace(name="usage"), + ), + patch( + "app.agent.ToolSelectorMiddleware", + side_effect=_capture_selector, + ), + patch("app.agent.InMemorySaver", return_value=object()), + patch("app.agent.create_agent", side_effect=_capture_agent), + patch.object(settings, "LLM_MAX_TOOLS", max_tools), + ] + with ExitStack() as stack: + for patcher in patchers: + stack.enter_context(patcher) + await agent._create_agent(streaming=False) + + assert captured["agent_tools"] == [main_tools[0], skill_tool, activity_tool] + assert captured["subagent_tools"] == [subagent_tools[0]] + if max_tools: + assert captured["selection_tools"] == [ + main_tools[0], + skill_tool, + activity_tool, + subagent_task_tool, + ] + else: + assert "selection_tools" not in captured + policy_middleware = next( + middleware + for middleware in captured["middlewares"] + if isinstance(middleware, AgentPolicyMiddleware) + ) + assert [ + entry.source + for entry in policy_middleware.catalog.collisions["shared_echo"] + ] == ["mcp:one", "mcp:two", "middleware:skills"] + assert ( + policy_middleware.catalog.resolve_unique("query_activity_log").tool + is activity_tool + ) + assert policy_middleware.catalog.resolve_unique("task").tool is subagent_task_tool + with pytest.raises(ToolIdentityAmbiguousError, match="TOOL_IDENTITY_AMBIGUOUS"): + policy_middleware.catalog.resolve_unique("shared_echo") + assert [ + entry.source + for entry in captured["subagent_catalog"].collisions["shared_echo"] + ] == ["mcp:one", "mcp:two"] + + @pytest.mark.anyio async def test_execute_agent_sends_only_latest_message_on_cache_hit(): """缓存命中时只把本轮新消息交给 LangGraph，避免重复提交历史。""" diff --git a/tests/test_agent_mcp.py b/tests/test_agent_mcp.py index 1c6b6c9c3..1d8064f37 100644 --- a/tests/test_agent_mcp.py +++ b/tests/test_agent_mcp.py @@ -1,10 +1,16 @@ import sys import textwrap +from unittest.mock import AsyncMock, patch import pytest from app.agent.mcp import AgentMcpManager, AgentMcpToolSpec -from app.agent.tools.impl.mcp import McpExternalTool +from app.agent.tools.catalog import ToolCatalogSnapshot +from app.agent.tools.impl.mcp import ( + McpExternalTool, + create_external_mcp_tools, + select_legacy_mcp_tools, +) from app.schemas.agent import AgentMcpServerConfig @@ -80,6 +86,122 @@ async def test_stdio_mcp_server_lists_tools(tmp_path): assert tools[0].input_schema["properties"]["text"]["type"] == "string" +@pytest.mark.anyio +async def test_enabled_specs_preserve_cross_server_name_collisions() -> None: + """跨 MCP 服务器生成相同 Agent 名时必须把全部身份交给目录判断。""" + manager = AgentMcpManager() + first = AgentMcpServerConfig(id="one", name="one", transport="stdio", command="one") + second = AgentMcpServerConfig(id="two", name="two", transport="stdio", command="two") + + def _spec(server): + return AgentMcpToolSpec( + server=server, + name="echo", + agent_tool_name="shared_echo", + description="echo", + input_schema={"type": "object", "properties": {}}, + ) + + with patch.object(manager, "get_servers", return_value=[first, second]), patch.object( + manager, + "list_server_tools", + new=AsyncMock(side_effect=lambda server: [_spec(server)]), + ): + specs = await manager.list_enabled_tool_specs() + + assert [spec.server.id for spec in specs] == ["one", "two"] + + +@pytest.mark.anyio +async def test_mcp_catalog_uses_server_id_and_legacy_execution_keeps_first() -> None: + """目录应区分同显示名服务器，普通执行仍保留首个同名工具。""" + first_server = AgentMcpServerConfig( + id="one", + name="Shared Name", + transport="stdio", + command="one", + ) + second_server = AgentMcpServerConfig( + id="two", + name="Shared Name", + transport="stdio", + command="two", + ) + + def _spec(server): + return AgentMcpToolSpec( + server=server, + name="echo", + agent_tool_name="shared_echo", + description="echo", + input_schema={"type": "object", "properties": {}}, + ) + + tools = await create_external_mcp_tools( + session_id="session", + user_id="user", + specs=[_spec(first_server), _spec(second_server)], + ) + catalog = ToolCatalogSnapshot.from_tools( + tools, + plugin_revision=0, + factory_revision="factory-v1", + ) + + assert [entry.source for entry in catalog.collisions["shared_echo"]] == [ + "mcp:one", + "mcp:two", + ] + assert len({entry.identity for entry in catalog.collisions["shared_echo"]}) == 2 + assert select_legacy_mcp_tools(tools) == [tools[0]] + + +@pytest.mark.anyio +async def test_mcp_catalog_distinguishes_normalized_names_within_server() -> None: + """同一服务内规范化重名的原始工具仍应具有不同绑定身份。""" + server = AgentMcpServerConfig( + id="shared", + name="Shared", + transport="stdio", + command="shared", + ) + + def _spec(name: str) -> AgentMcpToolSpec: + return AgentMcpToolSpec( + server=server, + name=name, + agent_tool_name="mcp_shared_foo_bar", + description="same description", + input_schema={"type": "object", "properties": {}}, + ) + + tools = await create_external_mcp_tools( + session_id="session", + user_id="user", + specs=[_spec("foo-bar"), _spec("foo_bar")], + ) + catalog = ToolCatalogSnapshot.from_tools( + tools, + plugin_revision=0, + factory_revision="factory-v1", + ) + first_only = ToolCatalogSnapshot.from_tools( + [tools[0]], + plugin_revision=0, + factory_revision="factory-v1", + ) + second_only = ToolCatalogSnapshot.from_tools( + [tools[1]], + plugin_revision=0, + factory_revision="factory-v1", + ) + + collisions = catalog.collisions["mcp_shared_foo_bar"] + assert len({entry.identity for entry in collisions}) == 2 + assert len({entry.revision.implementation for entry in collisions}) == 2 + assert first_only.signature != second_only.signature + + @pytest.mark.anyio async def test_stdio_mcp_server_calls_tool(tmp_path): """stdio MCP 工具应能通过 tools/call 返回内容。""" diff --git a/tests/test_agent_policy_canonical.py b/tests/test_agent_policy_canonical.py new file mode 100644 index 000000000..c6ead3d40 --- /dev/null +++ b/tests/test_agent_policy_canonical.py @@ -0,0 +1,228 @@ +import math +from dataclasses import FrozenInstanceError + +import pytest +from pydantic import BaseModel, ConfigDict, Field + +from app.agent.policy.canonical import ( + CanonicalizationError, + canonicalize_invocation, +) +from app.agent.policy.contracts import ( + ConversationKind, + DeliveryTarget, + InboundEnvelope, + InboundProvenance, + ToolRevision, +) +from app.agent.policy.registry import DEFAULT_TOOL_POLICY_REGISTRY +from app.agent.tools.impl.query_system_settings import QuerySystemSettingsTool + + +def _secret_policy(): + return DEFAULT_TOOL_POLICY_REGISTRY.resolve( + tool_name="query_system_settings", + arguments={"show_secrets": True}, + requires_admin=True, + ) + + +def _tool_revision() -> ToolRevision: + return ToolRevision( + implementation="query-system-settings:1", + factory="builtin-factory:1", + plugin="plugin-catalog:1", + ) + + +def test_canonical_invocation_includes_defaults_and_is_stable() -> None: + """省略的默认值必须进入相同规范化参数与摘要。""" + tool = QuerySystemSettingsTool(session_id="session-1", user_id="admin") + + omitted = canonicalize_invocation( + tool=tool, + arguments={"setting_key": "COOKIECLOUD_KEY", "show_secrets": True}, + policy=_secret_policy(), + tool_revision=_tool_revision(), + ) + explicit = canonicalize_invocation( + tool=tool, + arguments={ + "setting_key": "COOKIECLOUD_KEY", + "group": "all", + "keyword": None, + "include_values": None, + "show_secrets": True, + }, + policy=_secret_policy(), + tool_revision=_tool_revision(), + ) + + assert omitted.digest == explicit.digest + assert omitted.arguments == explicit.arguments + assert omitted.preconditions == ( + ("setting", "settings:COOKIECLOUD_KEY:settings"), + ) + + +def test_canonical_invocation_repr_hides_arguments_and_json() -> None: + """调用对象的 repr 不得泄露确认参数或完整规范化 JSON。""" + marker = "canonical-private-marker" + + class _Input(BaseModel): + model_config = ConfigDict(extra="forbid") + value: str + + class _Tool: + name = "private_tool" + args_schema = _Input + + invocation = canonicalize_invocation( + tool=_Tool(), + arguments={"value": marker}, + policy=_secret_policy(), + tool_revision=_tool_revision(), + ) + + assert marker not in repr(invocation) + assert "canonical_json" not in repr(invocation) + + +def test_canonical_arguments_are_recursively_immutable() -> None: + """确认等待期间不能修改嵌套参数后复用旧 digest。""" + + class _Input(BaseModel): + payload: dict[str, list[str]] + + class _Tool: + name = "nested_tool" + args_schema = _Input + + invocation = canonicalize_invocation( + tool=_Tool(), + arguments={"payload": {"items": ["one"]}}, + policy=_secret_policy(), + tool_revision=_tool_revision(), + ) + + with pytest.raises(TypeError): + invocation.arguments["payload"]["items"] += ("two",) + + +def test_inbound_contract_hides_raw_text_and_keeps_target_identities_separate() -> None: + """入站原文不可进入 repr，actor、recipient 和 conversation 独立绑定。""" + target = DeliveryTarget( + channel="telegram", + source_instance_id="source-1", + tenant_or_account_id="account-1", + conversation_kind=ConversationKind.PRIVATE, + conversation_id="chat-1", + recipient_id="recipient-1", + actor_id="actor-1", + server_session_id="session-1", + ) + envelope = InboundEnvelope( + provenance=InboundProvenance.VERIFIED_ADAPTER, + target=target, + inbound_event_id="event-1", + raw_text=" confirm K7P4-M2Q8 ", + normalized_text="confirm K7P4-M2Q8", + ) + + rendered = repr(envelope) + assert "K7P4-M2Q8" not in rendered + assert target.actor_id != target.recipient_id + assert target.recipient_id != target.conversation_id + with pytest.raises(FrozenInstanceError): + envelope.inbound_event_id = "forged" + + +def test_canonical_invocation_rejects_unknown_arguments() -> None: + """严格确认不能把 schema 外参数降级为原始字典。""" + tool = QuerySystemSettingsTool(session_id="session-1", user_id="admin") + + with pytest.raises(CanonicalizationError, match="参数校验失败"): + canonicalize_invocation( + tool=tool, + arguments={"show_secrets": True, "forged": "value"}, + policy=_secret_policy(), + tool_revision=_tool_revision(), + ) + + +def test_canonical_invocation_rejects_missing_pydantic_schema() -> None: + """动态 dict schema 不能进入严格确认路径。""" + + class _Tool: + name = "dynamic_tool" + args_schema = {"type": "object"} + + with pytest.raises(CanonicalizationError, match="Pydantic"): + canonicalize_invocation( + tool=_Tool(), + arguments={}, + policy=_secret_policy(), + tool_revision=_tool_revision(), + ) + + +def test_canonical_invocation_preserves_unicode_and_rejects_nan() -> None: + """稳定 JSON 保留 Unicode，同时禁止非标准 NaN。""" + + class _Input(BaseModel): + label: str + value: float = Field(allow_inf_nan=True) + + class _Tool: + name = "unicode_tool" + args_schema = _Input + + with pytest.raises(CanonicalizationError, match="无法规范化"): + canonicalize_invocation( + tool=_Tool(), + arguments={"label": "中文", "value": math.nan}, + policy=_secret_policy(), + tool_revision=_tool_revision(), + ) + + +def test_canonicalization_reads_no_setting_value(monkeypatch) -> None: + """确认前只能读取静态 SettingSpec，不能访问实际设置值。""" + tool = QuerySystemSettingsTool(session_id="session-1", user_id="admin") + load_value = monkeypatch.setattr( + QuerySystemSettingsTool, + "_load_setting_value", + lambda *_: pytest.fail("canonicalization must not read setting value"), + ) + + invocation = canonicalize_invocation( + tool=tool, + arguments={"setting_key": "COOKIECLOUD_KEY", "show_secrets": True}, + policy=_secret_policy(), + tool_revision=_tool_revision(), + ) + + assert invocation.digest + assert load_value is None + + +def test_group_selector_binds_static_setting_set_without_loading_values( + monkeypatch, +) -> None: + """列表型密钥读取必须绑定匹配的静态设置集合。""" + tool = QuerySystemSettingsTool(session_id="session-1", user_id="admin") + monkeypatch.setattr( + QuerySystemSettingsTool, + "_load_setting_value", + lambda *_: pytest.fail("canonicalization must not read setting value"), + ) + + invocation = canonicalize_invocation( + tool=tool, + arguments={"group": "ai_agent", "show_secrets": True}, + policy=_secret_policy(), + tool_revision=_tool_revision(), + ) + + assert len(invocation.preconditions) > 1 + assert all(kind == "setting" for kind, _identity in invocation.preconditions) diff --git a/tests/test_agent_tool_catalog.py b/tests/test_agent_tool_catalog.py new file mode 100644 index 000000000..105d0dba0 --- /dev/null +++ b/tests/test_agent_tool_catalog.py @@ -0,0 +1,159 @@ +"""Agent 本地工具目录身份、冲突与版本窗口测试。""" + +from types import SimpleNamespace +from unittest.mock import patch + +import pytest +from pydantic import BaseModel + +from app.agent.tools.catalog import ( + ToolCatalogSnapshot, + ToolIdentityAmbiguousError, +) +from app.agent.tools.factory import MoviePilotToolFactory +from app.core.plugin import PluginManager + + +class _Arguments(BaseModel): + """目录测试工具的参数契约。""" + + query: str = "" + + +def _tool(name: str, source: str = "builtin") -> SimpleNamespace: + """构造带稳定 schema 与来源的最小工具替身。""" + return SimpleNamespace( + name=name, + args_schema=_Arguments, + _agent_tool_source=source, + ) + + +def test_catalog_preserves_order_and_resolves_exact_instance() -> None: + """目录应保持构造顺序，并返回同一次构造的精确实例。""" + first = _tool("first") + second = _tool("second", "plugin:demo") + + catalog = ToolCatalogSnapshot.from_tools( + [first, second], + plugin_revision=7, + factory_revision="factory-v1", + ) + + assert catalog.tools == [first, second] + assert catalog.resolve_unique("second").tool is second + assert catalog.resolve_unique("missing") is None + assert catalog.signature[0:2] == ("factory-v1", 7) + + +def test_catalog_records_all_duplicate_names_and_strict_lookup_fails() -> None: + """内置与插件同名时必须保留双方身份并拒绝隐式选胜者。""" + builtin = _tool("query_system_settings") + plugin = _tool("query_system_settings", "plugin:demo") + catalog = ToolCatalogSnapshot.from_tools( + [builtin, plugin], + plugin_revision=3, + factory_revision="factory-v1", + ) + + assert [entry.tool for entry in catalog.collisions["query_system_settings"]] == [ + builtin, + plugin, + ] + with pytest.raises( + ToolIdentityAmbiguousError, + match="TOOL_IDENTITY_AMBIGUOUS", + ): + catalog.resolve_unique("query_system_settings") + + +def test_catalog_subset_keeps_all_identities_for_selected_name() -> None: + """子图执行只选首个实例时，严格目录仍必须保留全部同名身份。""" + first = _tool("shared", "mcp:one") + second = _tool("shared", "mcp:two") + catalog = ToolCatalogSnapshot.from_tools( + [first, second], + plugin_revision=0, + factory_revision="factory-v1", + ) + + selected = catalog.select([first]) + + assert [entry.tool for entry in selected.collisions["shared"]] == [first, second] + + +def test_catalog_signature_changes_with_schema_and_plugin_revision() -> None: + """schema 或插件目录 revision 变化必须使图缓存签名失效。""" + tool = _tool("demo") + first = ToolCatalogSnapshot.from_tools( + [tool], plugin_revision=1, factory_revision="factory-v1" + ) + + class _UpdatedArguments(BaseModel): + """模拟热加载后的参数契约。""" + + query: str = "" + limit: int = 10 + + tool.args_schema = _UpdatedArguments + schema_changed = ToolCatalogSnapshot.from_tools( + [tool], plugin_revision=1, factory_revision="factory-v1" + ) + revision_changed = ToolCatalogSnapshot.from_tools( + [tool], plugin_revision=2, factory_revision="factory-v1" + ) + + assert schema_changed.signature != first.signature + assert revision_changed.signature != schema_changed.signature + + +def test_catalog_signature_changes_with_json_schema_mapping() -> None: + """MCP 使用的 dict JSON Schema 变化必须使目录签名失效。""" + tool = _tool("mcp_demo", "mcp:demo") + tool.args_schema = {"type": "object", "properties": {"query": {"type": "string"}}} + first = ToolCatalogSnapshot.from_tools( + [tool], plugin_revision=1, factory_revision="factory-v1" + ) + tool.args_schema = { + "type": "object", + "properties": { + "query": {"type": "string"}, + "limit": {"type": "integer"}, + }, + } + second = ToolCatalogSnapshot.from_tools( + [tool], plugin_revision=1, factory_revision="factory-v1" + ) + + assert second.signature != first.signature + + +def test_catalog_signature_changes_with_tool_description() -> None: + """影响模型选择的工具描述变化必须使目录签名失效。""" + tool = _tool("demo") + tool.description = "first description" + first = ToolCatalogSnapshot.from_tools( + [tool], plugin_revision=1, factory_revision="factory-v1" + ) + tool.description = "updated description" + second = ToolCatalogSnapshot.from_tools( + [tool], plugin_revision=1, factory_revision="factory-v1" + ) + + assert second.signature != first.signature + + +def test_factory_catalog_retries_plugin_revision_churn_with_bound() -> None: + """插件构造期间持续 reload 时只能有界重试并失败关闭。""" + revisions = iter([1, 2, 3, 4, 5, 6]) + plugin_manager = PluginManager() + with patch.object( + plugin_manager, + "get_plugin_agent_tools_revision", + side_effect=lambda: next(revisions), + ), patch.object(MoviePilotToolFactory, "create_tools", return_value=[]): + with pytest.raises(RuntimeError, match="持续变化"): + MoviePilotToolFactory.create_catalog( + session_id="session", + user_id="user", + ) diff --git a/tests/test_agent_tool_factory_cache.py b/tests/test_agent_tool_factory_cache.py index 8f2aa9304..d8108cfde 100644 --- a/tests/test_agent_tool_factory_cache.py +++ b/tests/test_agent_tool_factory_cache.py @@ -1,5 +1,6 @@ from types import SimpleNamespace from typing import Iterator, Optional, Type +from unittest.mock import patch import pytest from pydantic import BaseModel @@ -147,6 +148,24 @@ def test_plugin_agent_tools_cache_can_be_cleared( assert len(calls) == 2 +def test_plugin_agent_tools_revision_churn_is_bounded( + plugin_manager: PluginManager, +) -> None: + """插件状态持续变化时注册表构造必须有界失败，不能卡住调用线程。""" + def _changing_tools() -> list[type[MoviePilotTool]]: + plugin_manager.clear_plugin_agent_tools_cache() + return [DemoAgentTool] + + plugin_manager.running_plugins["DemoPlugin"] = SimpleNamespace( + plugin_name="Demo Plugin", + get_state=lambda: True, + get_agent_tools=_changing_tools, + ) + + with pytest.raises(RuntimeError, match="持续变化"): + plugin_manager.get_plugin_agent_tools() + + def test_factory_reuses_plugin_registry_but_creates_new_tool_instances( plugin_manager: PluginManager, ) -> None: @@ -191,3 +210,25 @@ def test_factory_suppresses_plugin_message_tools_for_subagents( assert "demo_agent_tool" in tool_names assert "demo_message_agent_tool" not in tool_names + + +def test_factory_catalog_records_two_plugin_duplicate_names( + plugin_manager: PluginManager, +) -> None: + """两个插件声明同名工具时，目录必须保留两个插件身份。""" + plugin_manager.running_plugins["PluginOne"] = _build_plugin([DemoAgentTool]) + plugin_manager.running_plugins["PluginTwo"] = _build_plugin([DemoAgentTool]) + + with patch.object( + MoviePilotToolFactory, + "_get_builtin_tool_classes", + return_value=[], + ): + catalog = MoviePilotToolFactory.create_catalog( + session_id="session-1", + user_id="10001", + ) + + assert [ + entry.source for entry in catalog.collisions["demo_agent_tool"] + ] == ["plugin:PluginOne", "plugin:PluginTwo"] diff --git a/tests/test_agent_tool_policy.py b/tests/test_agent_tool_policy.py index 32ec5fbed..774625561 100644 --- a/tests/test_agent_tool_policy.py +++ b/tests/test_agent_tool_policy.py @@ -171,6 +171,38 @@ def test_registry_separates_safe_read_from_legacy_shadow() -> None: assert dynamic_policy.result_sensitivity is ResultSensitivity.UNKNOWN +@pytest.mark.parametrize("show_secrets", [None, False]) +def test_system_settings_without_secret_values_stays_legacy_shadow( + show_secrets, +) -> None: + """普通设置读取维持既有兼容路径，不增加确认。""" + policy = DEFAULT_TOOL_POLICY_REGISTRY.resolve( + tool_name="query_system_settings", + arguments={"show_secrets": show_secrets}, + requires_admin=True, + ) + + assert policy.migration_state is MigrationState.LEGACY_SHADOW + assert policy.effect is ActionEffect.UNKNOWN + + +def test_system_settings_secret_read_has_enforced_sensitive_policy() -> None: + """显式读取密钥只能进入宿主强制确认策略。""" + policy = DEFAULT_TOOL_POLICY_REGISTRY.resolve( + tool_name="query_system_settings", + arguments={"show_secrets": True}, + requires_admin=True, + ) + + assert policy.migration_state is MigrationState.ENFORCED + assert policy.effect is ActionEffect.SENSITIVE_READ + assert policy.required_role is PrincipalRole.SYSTEM_ADMIN + assert policy.confirmation.value == "required" + assert policy.result_sensitivity is ResultSensitivity.SECRET + assert policy.background_allowed is False + assert policy.subagent_allowed is False + + def test_legacy_shadow_decision_allows_without_claiming_enforcement() -> None: """G1 的 shadow 决策只能观测，不能拒绝或要求确认。""" context = _interactive_context() @@ -187,6 +219,24 @@ def test_legacy_shadow_decision_allows_without_claiming_enforcement() -> None: assert observation.decision.reason_code == "legacy_shadow_allow" +def test_sensitive_policy_does_not_claim_safe_read_before_strict_runtime() -> None: + """严格运行时接管前，敏感读取只能以明确的兼容 shadow 语义通过。""" + tool = QuerySystemSettingsTool(session_id="session-1", user_id="admin") + tool.set_agent_context({"is_admin": True}) + + observation = DEFAULT_TOOL_POLICY_ORCHESTRATOR.start( + context=_interactive_context(), + tool=tool, + arguments={"setting_key": "COOKIECLOUD_KEY", "show_secrets": True}, + ) + + assert observation.policy.migration_state is MigrationState.ENFORCED + assert observation.decision.allowed is True + assert observation.decision.shadow is True + assert observation.decision.confirmation_required is False + assert observation.decision.reason_code == "strict_runtime_pending" + + def test_policy_context_reads_mutable_admin_state_without_model_fields() -> None: """缓存图复用时权限取当前宿主状态，模型参数不能伪造 principal。""" context = _interactive_context(is_admin=False) diff --git a/tests/test_mcp_plugin_tools.py b/tests/test_mcp_plugin_tools.py index c28facd3d..21b21598b 100644 --- a/tests/test_mcp_plugin_tools.py +++ b/tests/test_mcp_plugin_tools.py @@ -9,6 +9,7 @@ import pytest from app.agent.tools.base import MoviePilotTool from app.agent.tools.factory import MoviePilotToolFactory from app.agent.tools.manager import MoviePilotToolsManager +from app.agent.tools.catalog import ToolCatalogSnapshot from app.api.endpoints import mcp from app.core.plugin import PluginManager from app.utils.singleton import Singleton @@ -96,3 +97,23 @@ def test_mcp_refreshes_tools_after_plugin_lifecycle_change( ) missing_payload = json.loads(missing_result["content"][0]["text"]) assert "未找到" in missing_payload["error"] + + +def test_direct_manager_preserves_legacy_lookup_and_exposes_strict_resolution() -> None: + """普通 direct 调用保留 first-wins，严格调用可拒绝同名身份。""" + first = DemoPluginTool(session_id="session", user_id="user") + second = DemoPluginTool(session_id="session", user_id="user") + manager = MoviePilotToolsManager(session_id="session", user_id="user") + manager.tools = [first, second] + manager.catalog = ToolCatalogSnapshot.from_tools( + manager.tools, + plugin_revision=manager._plugin_agent_tools_revision, + factory_revision=MoviePilotToolFactory.catalog_factory_revision(), + ) + + assert manager.get_tool("demo_plugin_tool") is first + with pytest.raises(RuntimeError, match="TOOL_IDENTITY_AMBIGUOUS"): + manager.get_strict_tool("demo_plugin_tool") + + result = asyncio.run(manager.call_tool("demo_plugin_tool", {})) + assert result == "plugin-ok"