From 93761fe7e4bac5eb9af7b7ad5d162c257bc62488 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 1 Aug 2026 09:02:05 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=20Agent=20=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=E7=BC=96=E8=BE=91=E5=B7=A5=E5=85=B7=20(#6218)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/agent/prompt/System Core Prompt.txt | 6 +- app/agent/tools/impl/_file_write_utils.py | 53 +++++++ app/agent/tools/impl/edit_file.py | 133 ++++++++++++++---- app/agent/tools/impl/read_file.py | 41 +++++- app/agent/tools/impl/write_file.py | 84 +++++++++-- docs/cli.md | 3 + docs/mcp-api.md | 4 + skills/create-moviepilot-plugin/SKILL.md | 31 +++- tests/test_agent_file_tools.py | 131 +++++++++++++++++ tests/test_agent_resource_flow_permissions.py | 7 +- tests/test_builtin_skill_boundaries.py | 16 +++ 11 files changed, 465 insertions(+), 44 deletions(-) create mode 100644 app/agent/tools/impl/_file_write_utils.py create mode 100644 tests/test_agent_file_tools.py diff --git a/app/agent/prompt/System Core Prompt.txt b/app/agent/prompt/System Core Prompt.txt index c8603bc5..64ea5312 100644 --- a/app/agent/prompt/System Core Prompt.txt +++ b/app/agent/prompt/System Core Prompt.txt @@ -65,7 +65,11 @@ You act as a proactive agent. Your goal is to fully resolve the user's media-rel - If `search_media` fails, fall back to `search_web` or `recognize_media`. Only ask the user when automated paths are exhausted. - If torrent search yields no useful result, check site scope, site health, and recognition quality before concluding that the resource is unavailable. - Reuse the latest torrent search cache for `get_search_results` and `add_download_tasks` instead of re-running the same search unnecessarily. -- Use `execute_command` only for diagnostics, read-only inspection, or commands the user explicitly asked to run. Its default `action=start` starts a managed background session and returns `session_id`, `status`, `last_seq`, and `output_until_seq`; call the same tool again with `action=read`, `action=wait`, `action=write`, or `action=kill` to poll output, wait in short segments, send stdin, or stop the process. +- For administrator code discovery across local files, use `execute_command(action="run")` with `rg` and narrow globs or paths. Use `list_directory` to inspect one known directory or a supported remote storage backend, and use `read_file` when the exact local file is known. +- Read the relevant file before changing it. Use `edit_file` for localized exact replacements; make `old_text` unique with enough surrounding context, and use `replace_all=true` only when every match must change. Use `write_file` for new files; set `overwrite=true` only for an intentional full rewrite, and use `read_file(include_metadata=true)` plus `expected_sha256` when preserving the previously read version matters. +- When implementation depends on a Python or Node.js API, first identify the installed or locked dependency version from environment metadata, requirements, package manifests, lockfiles, local source, and type declarations. Use `rg` against the relevant package directory, `.venv`, or `node_modules` instead of scanning the entire project without bounds. If local evidence is insufficient, use `search_web` and then `browse_webpage` to read the matching version of the official documentation. Do not guess signatures from memory, mix examples from incompatible versions, or install a package only to inspect its API. +- Use structured file tools for source edits because they enforce file access boundaries and conflict checks. Never use shell redirection, inline scripts, or another tool to bypass a file-tool permission denial. +- Use `execute_command` for administrator-only multi-file diagnostics, tests, Git, service operations, SSH, or an exact command the user requested. Use `action=run` for short bounded commands. Use `action=start` for long-running or interactive commands, including SSH; then continue with `read`, `wait`, `write`, or `kill` using the returned `session_id`. Do not start a background session for a short command that can finish within `action=run`. diff --git a/app/agent/tools/impl/_file_write_utils.py b/app/agent/tools/impl/_file_write_utils.py new file mode 100644 index 00000000..8d42729e --- /dev/null +++ b/app/agent/tools/impl/_file_write_utils.py @@ -0,0 +1,53 @@ +"""Agent 文件写入工具的共享辅助函数。""" + +import hashlib +import os +import tempfile +from pathlib import Path + + +class FileVersionConflictError(RuntimeError): + """目标文件在准备写入期间发生变化。""" + + +def calculate_file_sha256(path: Path) -> str: + """计算文件原始字节的 SHA-256，用于检测陈旧写入。""" + digest = hashlib.sha256() + with path.open("rb") as file_handle: + for chunk in iter(lambda: file_handle.read(64 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def atomic_write_text( + path: Path, + content: str, + expected_sha256: str | None = None, +) -> None: + """校验目标版本后，在同目录写入临时文件并原子替换文本。""" + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temp_name = tempfile.mkstemp( + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + ) + temp_path = Path(temp_name) + try: + with os.fdopen(descriptor, "w", encoding="utf-8", newline="") as file_handle: + file_handle.write(content) + file_handle.flush() + os.fsync(file_handle.fileno()) + + if expected_sha256: + if ( + not path.is_file() + or calculate_file_sha256(path).casefold() + != expected_sha256.casefold() + ): + raise FileVersionConflictError(str(path)) + if path.exists(): + os.chmod(temp_path, path.stat().st_mode) + os.replace(temp_path, path) + finally: + if temp_path.exists(): + temp_path.unlink() diff --git a/app/agent/tools/impl/edit_file.py b/app/agent/tools/impl/edit_file.py index ce1d0b2b..9dae896b 100644 --- a/app/agent/tools/impl/edit_file.py +++ b/app/agent/tools/impl/edit_file.py @@ -1,4 +1,4 @@ -"""文件编辑工具""" +"""文件精确编辑工具。""" from pathlib import Path from typing import Optional, Type @@ -7,6 +7,11 @@ from anyio import Path as AsyncPath from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool +from app.agent.tools.impl._file_write_utils import ( + FileVersionConflictError, + atomic_write_text, + calculate_file_sha256, +) from app.agent.tools.tags import ToolTag from app.log import logger @@ -15,18 +20,44 @@ class EditFileInput(BaseModel): """文件编辑工具的输入参数模型。""" file_path: str = Field(..., description="The absolute path of the file to edit") - old_text: str = Field(..., description="The exact old text to be replaced") + old_text: str = Field( + ..., + description=( + "The exact old text to replace. It must be non-empty and uniquely " + "identify one location unless replace_all is true." + ), + ) new_text: str = Field(..., description="The new text to replace with") + replace_all: bool = Field( + False, + description=( + "Replace every exact match. Keep false for normal code edits so an " + "ambiguous match fails instead of changing multiple locations." + ), + ) + expected_sha256: Optional[str] = Field( + None, + pattern=r"^[0-9a-fA-F]{64}$", + description=( + "Optional SHA-256 returned by read_file(include_metadata=true). The " + "edit fails if the file changed after it was read." + ), + ) class EditFileTool(MoviePilotTool): + """使用精确文本匹配安全编辑本地文件。""" + name: str = "edit_file" tags: list[str] = [ ToolTag.Write, ToolTag.File, ] description: str = ( - "Edit a local text file by replacing specific old text with new text. " + "Edit an existing local text file using an exact text match. By default " + "the match must occur exactly once; use replace_all only for intentional " + "bulk replacement. old_text cannot be empty, and new files must be " + "created with write_file. Supports an optional SHA-256 conflict check. " "Non-admin users can only edit files inside the MoviePilot Agent config " "directory." ) @@ -38,7 +69,16 @@ class EditFileTool(MoviePilotTool): file_name = Path(file_path).name if file_path else "未知文件" return f"编辑文件: {file_name}" - async def run(self, file_path: str, old_text: str, new_text: str, **kwargs) -> str: + async def run( + self, + file_path: str, + old_text: str, + new_text: str, + replace_all: bool = False, + expected_sha256: Optional[str] = None, + **kwargs, + ) -> str: + """校验精确匹配和可选文件版本后，以原子方式写入编辑结果。""" logger.info(f"执行工具: {self.name}, 参数: file_path={file_path}") try: @@ -48,37 +88,74 @@ class EditFileTool(MoviePilotTool): if access_error: return access_error - path = AsyncPath(resolved_path) - # 校验逻辑：如果要替换特定文本，文件必须存在且包含该文本 - if not await path.exists(): - # 如果 old_text 为空，可能用户想直接创建文件，但通常 edit_file 需要匹配旧内容 - if old_text: - return f"错误：文件 {resolved_path} 不存在，无法进行内容替换。" + if not old_text: + return "错误：old_text 不能为空；创建或完整写入文件请使用 write_file。" - if await path.exists() and not await path.is_file(): + path = AsyncPath(resolved_path) + if not await path.exists(): + return f"错误：文件 {resolved_path} 不存在；创建文件请使用 write_file。" + + if not await path.is_file(): return f"错误：{resolved_path} 不是一个文件" - if await path.exists(): - content = await path.read_text(encoding="utf-8", errors="replace") - if old_text not in content: - logger.warning(f"编辑文件 {resolved_path} 失败：未找到指定的旧文本块") - return f"错误：在文件 {resolved_path} 中未找到指定的旧文本。请确保包含所有的空格、缩进 and 换行符。" - occurrences = content.count(old_text) - new_content = content.replace(old_text, new_text) - else: - # 文件不存在且 old_text 为空的情形（初始化新文件） - new_content = new_text - occurrences = 1 + local_path = Path(resolved_path) + current_sha256 = await self.run_blocking( + "default", calculate_file_sha256, local_path + ) + if ( + expected_sha256 + and current_sha256.casefold() != expected_sha256.casefold() + ): + return ( + f"错误：文件 {resolved_path} 已在读取后发生变化，拒绝覆盖。" + "请重新读取文件并基于最新内容编辑。" + ) - # 自动创建父目录 - await path.parent.mkdir(parents=True, exist_ok=True) + content = await path.read_text(encoding="utf-8", errors="strict") + occurrences = content.count(old_text) + if occurrences == 0: + logger.warning(f"编辑文件 {resolved_path} 失败：未找到指定的旧文本块") + return ( + f"错误：在文件 {resolved_path} 中未找到指定的旧文本。" + "请重新读取文件并确认空格、缩进和换行。" + ) + if occurrences > 1 and not replace_all: + return ( + f"错误：old_text 在文件 {resolved_path} 中匹配到 {occurrences} 处，" + "为避免误改已拒绝编辑。请提供更多上下文使其唯一，或明确设置 " + "replace_all=true。" + ) - # 写入文件 - await path.write_text(new_content, encoding="utf-8") + replacement_count = occurrences if replace_all else 1 + new_content = content.replace( + old_text, + new_text, + -1 if replace_all else 1, + ) + await self.run_blocking( + "default", + atomic_write_text, + local_path, + new_content, + current_sha256, + ) + new_sha256 = await self.run_blocking( + "default", calculate_file_sha256, local_path + ) - logger.info(f"成功编辑文件 {resolved_path}，替换了 {occurrences} 处内容") - return f"成功编辑文件 {resolved_path} (替换了 {occurrences} 处匹配内容)" + logger.info( + f"成功编辑文件 {resolved_path}，替换了 {replacement_count} 处内容" + ) + return ( + f"成功编辑文件 {resolved_path}（替换了 {replacement_count} 处匹配内容，" + f"sha256={new_sha256}）" + ) + except FileVersionConflictError: + return ( + f"错误：文件 {file_path} 在编辑期间发生变化，拒绝覆盖。" + "请重新读取文件并再次编辑。" + ) except PermissionError: return f"错误：没有访问/修改 {file_path} 的权限" except UnicodeDecodeError: diff --git a/app/agent/tools/impl/read_file.py b/app/agent/tools/impl/read_file.py index a0a7ccb9..2ec7acbb 100644 --- a/app/agent/tools/impl/read_file.py +++ b/app/agent/tools/impl/read_file.py @@ -1,5 +1,7 @@ """文件读取工具""" +import hashlib +import json from pathlib import Path from typing import Optional, Type @@ -16,12 +18,22 @@ MAX_READ_SIZE = 50 * 1024 class ReadFileInput(BaseModel): """文件读取工具的输入参数模型。""" + file_path: str = Field(..., description="The absolute path of the file to read") start_line: Optional[int] = Field(None, description="The starting line number (1-based, inclusive). If not provided, reading starts from the beginning of the file.") end_line: Optional[int] = Field(None, description="The ending line number (1-based, inclusive). If not provided, reading goes until the end of the file.") + include_metadata: bool = Field( + False, + description=( + "Return structured JSON containing content, size, truncation state, " + "and SHA-256. Use before a guarded full-file overwrite." + ), + ) class ReadFileTool(MoviePilotTool): + """按行范围读取本地文本文件，并可返回文件版本元数据。""" + name: str = "read_file" tags: list[str] = [ ToolTag.Read, @@ -36,8 +48,15 @@ class ReadFileTool(MoviePilotTool): file_name = Path(file_path).name if file_path else "未知文件" return f"读取文件: {file_name}" - async def run(self, file_path: str, start_line: Optional[int] = None, - end_line: Optional[int] = None, **kwargs) -> str: + async def run( + self, + file_path: str, + start_line: Optional[int] = None, + end_line: Optional[int] = None, + include_metadata: bool = False, + **kwargs, + ) -> str: + """读取指定文本范围，必要时附带完整文件的 SHA-256 元数据。""" logger.info(f"执行工具: {self.name}, 参数: file_path={file_path}, start_line={start_line}, end_line={end_line}") try: @@ -55,7 +74,8 @@ class ReadFileTool(MoviePilotTool): if not await path.is_file(): return f"错误：{resolved_path} 不是一个文件" - content = await path.read_text(encoding="utf-8", errors="replace") + raw_content = await path.read_bytes() + content = raw_content.decode("utf-8", errors="replace") truncated = False if start_line is not None or end_line is not None: @@ -78,6 +98,21 @@ class ReadFileTool(MoviePilotTool): content = content_bytes[:MAX_READ_SIZE].decode("utf-8", errors="replace") truncated = True + if include_metadata: + return json.dumps( + { + "file_path": str(resolved_path), + "sha256": hashlib.sha256(raw_content).hexdigest(), + "size_bytes": len(raw_content), + "start_line": start_line, + "end_line": end_line, + "truncated": truncated, + "content": content, + }, + ensure_ascii=False, + indent=2, + ) + if truncated: return f"{content}\n\n[警告：文件内容已超过50KB限制，以上内容已被截断。请使用 start_line/end_line 参数分段读取。]" diff --git a/app/agent/tools/impl/write_file.py b/app/agent/tools/impl/write_file.py index 0e57f14f..0f8e8b1b 100644 --- a/app/agent/tools/impl/write_file.py +++ b/app/agent/tools/impl/write_file.py @@ -7,6 +7,11 @@ from anyio import Path as AsyncPath from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool +from app.agent.tools.impl._file_write_utils import ( + FileVersionConflictError, + atomic_write_text, + calculate_file_sha256, +) from app.agent.tools.tags import ToolTag from app.log import logger @@ -16,17 +21,36 @@ class WriteFileInput(BaseModel): file_path: str = Field(..., description="The absolute path of the file to write") content: str = Field(..., description="The content to write into the file") + overwrite: bool = Field( + False, + description=( + "Allow replacing an existing file in full. Keep false when creating a " + "new file; prefer edit_file for localized changes." + ), + ) + expected_sha256: Optional[str] = Field( + None, + pattern=r"^[0-9a-fA-F]{64}$", + description=( + "Optional SHA-256 returned by read_file(include_metadata=true). When " + "overwriting, fail if the existing file no longer has this hash." + ), + ) class WriteFileTool(MoviePilotTool): + """创建本地文本文件，或在显式允许后完整覆盖已有文件。""" + name: str = "write_file" tags: list[str] = [ ToolTag.Write, ToolTag.File, ] description: str = ( - "Write full content to a local text file. Non-admin users can only write " - "inside the MoviePilot Agent config directory." + "Create a local text file with complete content. Existing files are " + "protected unless overwrite=true; localized changes should use edit_file. " + "Supports an optional SHA-256 conflict check and writes atomically. " + "Non-admin users can only write inside the MoviePilot Agent config directory." ) args_schema: Type[BaseModel] = WriteFileInput @@ -36,7 +60,15 @@ class WriteFileTool(MoviePilotTool): file_name = Path(file_path).name if file_path else "未知文件" return f"写入文件: {file_name}" - async def run(self, file_path: str, content: str, **kwargs) -> str: + async def run( + self, + file_path: str, + content: str, + overwrite: bool = False, + expected_sha256: Optional[str] = None, + **kwargs, + ) -> str: + """创建或显式覆盖文件，并通过可选哈希阻止陈旧写入。""" logger.info(f"执行工具: {self.name}, 参数: file_path={file_path}") try: @@ -48,18 +80,52 @@ class WriteFileTool(MoviePilotTool): path = AsyncPath(resolved_path) - if await path.exists() and not await path.is_file(): + exists = await path.exists() + if exists and not await path.is_file(): return f"错误：{resolved_path} 路径已存在但不是一个文件" + if exists and not overwrite: + return ( + f"错误：文件 {resolved_path} 已存在，拒绝完整覆盖。" + "局部修改请使用 edit_file；确需重写时设置 overwrite=true。" + ) + if expected_sha256 and not exists: + return ( + f"错误：文件 {resolved_path} 不存在，无法校验 expected_sha256。" + "请确认路径和最新文件状态。" + ) - # 自动创建父目录 - await path.parent.mkdir(parents=True, exist_ok=True) + local_path = Path(resolved_path) + current_sha256 = None + if exists: + current_sha256 = await self.run_blocking( + "default", calculate_file_sha256, local_path + ) + if expected_sha256: + if current_sha256.casefold() != expected_sha256.casefold(): + return ( + f"错误：文件 {resolved_path} 已在读取后发生变化，拒绝覆盖。" + "请重新读取文件并基于最新内容写入。" + ) - # 写入文件 - await path.write_text(content, encoding="utf-8") + await self.run_blocking( + "default", + atomic_write_text, + local_path, + content, + current_sha256, + ) + new_sha256 = await self.run_blocking( + "default", calculate_file_sha256, local_path + ) logger.info(f"成功写入文件 {resolved_path}") - return f"成功写入文件 {resolved_path}" + return f"成功写入文件 {resolved_path}（sha256={new_sha256}）" + except FileVersionConflictError: + return ( + f"错误：文件 {file_path} 在写入期间发生变化，拒绝覆盖。" + "请重新读取文件并再次写入。" + ) except PermissionError: return f"错误：没有权限写入 {file_path}" except Exception as e: diff --git a/docs/cli.md b/docs/cli.md index d3eaf72b..f4115084 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -481,6 +481,9 @@ moviepilot tool run search_torrents media_type=movie tmdb_id=12345 - `tool list` 用于动态发现当前服务可调用的工具 - `tool show` 会输出参数名、类型和描述 - `tool run` 参数格式固定为 `key=value` +- `read_file`、`write_file`、`edit_file` 和 `execute_command` + 属于内置 Agent 的本地敏感能力，不通过 MCP/`moviepilot tool` 暴露；插件开发时 + 由 Agent 按当前用户权限直接调用这些工具。 ## Scheduler 命令 diff --git a/docs/mcp-api.md b/docs/mcp-api.md index 0e3a5140..dad58e85 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -239,6 +239,10 @@ AniList 榜单、探索、详情、人物和推荐接口优先通过 `anilist-ch 内置工具的 `inputSchema` 只包含实际执行业务所需的参数，不包含用于解释调用原因的通用 `explanation` 参数，以减少 Agent 上下文消耗。插件工具的参数结构由插件自身声明。 +内置 Agent 的本地文件与命令工具 `read_file`、`write_file`、`edit_file`、 +`execute_command` 不通过 MCP 暴露。这些工具在 Agent 运行时执行独立的 +用户权限与路径边界检查；MCP 隐藏列表只负责收敛接口暴露面，不替代权限控制。 + 媒体相关 MCP 工具（如 `query_media_detail`、`search_torrents`、`query_library_exists`、`add_subscribe`、`transfer_file`）接受 `tmdb_id`/`tmdbid`、`douban_id`/`doubanid`、`bangumi_id`/`bangumiid`、`anilist_id`/`anilistid`，也接受 `media_source` + `media_id`。工具返回的媒体、订阅、下载和整理记录会同步带回可用的四种专用 ID 及通用主身份。 `get_search_results` 可使用 `title_pattern` 对种子标题执行正则筛选，也可使用 `content_pattern` 联合匹配种子标题、简介和标签。`title_pattern` 保持仅匹配标题的兼容语义；需要在结果中查看种子简介时，传入 `include_description=true`。两种正则参数与站点、分辨率等结构化筛选条件同时传入时按 AND 关系组合。 diff --git a/skills/create-moviepilot-plugin/SKILL.md b/skills/create-moviepilot-plugin/SKILL.md index 4c5fd370..c59487d7 100644 --- a/skills/create-moviepilot-plugin/SKILL.md +++ b/skills/create-moviepilot-plugin/SKILL.md @@ -1,6 +1,6 @@ --- name: create-moviepilot-plugin -version: 2 +version: 3 description: >- Use this skill when the user asks to create, modify, debug, validate, or scaffold a MoviePilot local plugin. Covers MoviePilot V2 plugin development, @@ -11,7 +11,7 @@ description: >- sidebar pages, commands, services, workflow actions, agent tools, and local install/reload flows. Also use for Chinese requests mentioning 编写插件、本地插件源, 插件开发, V2插件, 插件市场, 本地安装插件, 插件热加载, 前端联邦, 侧栏入口, Vue插件页面. -allowed-tools: list_directory read_file write_file edit_file execute_command query_system_settings update_system_settings query_market_plugins install_plugin reload_plugin query_installed_plugins +allowed-tools: list_directory read_file write_file edit_file execute_command search_web browse_webpage query_system_settings update_system_settings query_market_plugins install_plugin reload_plugin query_installed_plugins --- # Create MoviePilot Plugin @@ -33,6 +33,33 @@ a local plugin source and installed into the running MoviePilot instance. - When working in or from `MoviePilot-Plugins`, read its `README.md`, `docs/Repository_Guide.md`, and `docs/V2_Plugin_Development.md`. For scenario-specific extensions, read the matching `docs/faq/*.md`. + +## Code Tool Workflow + +- Use `execute_command(action="run")` with `rg` and narrow globs or paths to + locate plugin classes, extension points, tests, and package entries. Use + `list_directory` only when inspecting one known folder or a configured remote + storage backend. +- Read the relevant implementation and adjacent example before editing. +- Before using a Python or Node.js dependency API, determine the exact installed + or locked version from requirements, package manifests, lockfiles, local + package source, and `.pyi`/`.d.ts` declarations. If those are insufficient, + use `search_web` with the official documentation domain and `browse_webpage` + to read the matching version. Do not guess API signatures from memory or mix + examples from different major versions. Search the relevant package directory, + `.venv`, or `node_modules` directly with `rg` instead of scanning the entire + project without bounds. +- Use `edit_file` for localized changes. Its `old_text` must identify one exact + location by default; add surrounding context instead of enabling + `replace_all` unless every match intentionally changes. +- Use `write_file` for new files. Existing files require `overwrite=true` for a + full rewrite; first call `read_file(include_metadata=true)` and pass its + `sha256` as `expected_sha256` when replacing previously read content. +- Use `execute_command(action="run")` for short validation, Git, and diagnostic + commands. Use `action="start"` only for interactive or long-running commands, + then continue through the returned session ID. +- Do not use shell redirection or inline scripts to perform source edits or to + bypass a file-tool permission error. - When the plugin uses Vue federation, also read `MoviePilot-Frontend/docs/module-federation-guide.md`, `MoviePilot-Frontend/docs/federation-troubleshooting.md`, diff --git a/tests/test_agent_file_tools.py b/tests/test_agent_file_tools.py new file mode 100644 index 00000000..54909224 --- /dev/null +++ b/tests/test_agent_file_tools.py @@ -0,0 +1,131 @@ +"""Agent 本地文件搜索与安全编辑工具测试。""" + +import asyncio +import hashlib +import json + +from app.agent.tools.impl.edit_file import EditFileTool +from app.agent.tools.impl.read_file import ReadFileTool +from app.agent.tools.impl.write_file import WriteFileTool + + +def _make_admin_tool(tool_class): + """创建带管理员上下文的文件工具实例。""" + tool = tool_class(session_id="session-1", user_id="admin") + tool.set_agent_context({"is_admin": True}) + return tool + + +def test_edit_file_rejects_ambiguous_match_by_default(tmp_path): + """精确编辑默认应拒绝多处匹配，避免静默批量修改代码。""" + file_path = tmp_path / "plugin.py" + file_path.write_text("enabled = False\nenabled = False\n", encoding="utf-8") + tool = _make_admin_tool(EditFileTool) + + result = asyncio.run( + tool.run(str(file_path), "enabled = False", "enabled = True") + ) + + assert "匹配到 2 处" in result + assert "replace_all=true" in result + assert file_path.read_text(encoding="utf-8") == ( + "enabled = False\nenabled = False\n" + ) + + +def test_edit_file_replace_all_requires_explicit_flag(tmp_path): + """显式开启 replace_all 后才应替换全部精确匹配。""" + file_path = tmp_path / "plugin.py" + file_path.write_text("old\nold\n", encoding="utf-8") + tool = _make_admin_tool(EditFileTool) + + result = asyncio.run( + tool.run(str(file_path), "old", "new", replace_all=True) + ) + + assert "替换了 2 处" in result + assert file_path.read_text(encoding="utf-8") == "new\nnew\n" + + +def test_edit_file_rejects_empty_match_and_missing_file(tmp_path): + """编辑工具不应再通过空匹配隐式创建文件。""" + file_path = tmp_path / "missing.py" + tool = _make_admin_tool(EditFileTool) + + empty_result = asyncio.run(tool.run(str(file_path), "", "content")) + missing_result = asyncio.run(tool.run(str(file_path), "old", "new")) + + assert "old_text 不能为空" in empty_result + assert "不存在" in missing_result + assert "write_file" in missing_result + assert not file_path.exists() + + +def test_edit_file_rejects_stale_sha256(tmp_path): + """文件在读取后变化时，哈希保护应拒绝基于旧版本编辑。""" + file_path = tmp_path / "plugin.py" + file_path.write_text("before", encoding="utf-8") + old_sha256 = hashlib.sha256(b"before").hexdigest() + file_path.write_text("changed elsewhere", encoding="utf-8") + tool = _make_admin_tool(EditFileTool) + + result = asyncio.run( + tool.run( + str(file_path), + "changed elsewhere", + "agent change", + expected_sha256=old_sha256, + ) + ) + + assert "已在读取后发生变化" in result + assert file_path.read_text(encoding="utf-8") == "changed elsewhere" + + +def test_write_file_protects_existing_file_and_supports_guarded_overwrite(tmp_path): + """完整写入应默认保护已有文件，并允许带版本校验的显式覆盖。""" + file_path = tmp_path / "plugin.py" + file_path.write_text("before", encoding="utf-8") + expected_sha256 = hashlib.sha256(b"before").hexdigest() + tool = _make_admin_tool(WriteFileTool) + + refused_result = asyncio.run(tool.run(str(file_path), "unexpected")) + written_result = asyncio.run( + tool.run( + str(file_path), + "after", + overwrite=True, + expected_sha256=expected_sha256, + ) + ) + stale_result = asyncio.run( + tool.run( + str(file_path), + "stale write", + overwrite=True, + expected_sha256=expected_sha256, + ) + ) + + assert "拒绝完整覆盖" in refused_result + assert "成功写入文件" in written_result + assert "sha256=" in written_result + assert "已在读取后发生变化" in stale_result + assert file_path.read_text(encoding="utf-8") == "after" + + +def test_read_file_can_return_sha256_metadata(tmp_path): + """读取工具应能返回供后续冲突检查使用的文件哈希。""" + file_path = tmp_path / "plugin.py" + file_path.write_text("插件内容", encoding="utf-8") + tool = _make_admin_tool(ReadFileTool) + + result = asyncio.run(tool.run(str(file_path), include_metadata=True)) + payload = json.loads(result) + + assert payload["content"] == "插件内容" + assert payload["size_bytes"] == len("插件内容".encode("utf-8")) + assert payload["sha256"] == hashlib.sha256( + "插件内容".encode("utf-8") + ).hexdigest() + assert payload["truncated"] is False diff --git a/tests/test_agent_resource_flow_permissions.py b/tests/test_agent_resource_flow_permissions.py index 8fea3337..5d6c6144 100644 --- a/tests/test_agent_resource_flow_permissions.py +++ b/tests/test_agent_resource_flow_permissions.py @@ -49,7 +49,12 @@ def test_non_admin_manager_exposes_restricted_file_tools(): manager = MoviePilotToolsManager(is_admin=False) tool_names = {tool.name for tool in manager.list_tools()} - assert {"read_file", "write_file", "edit_file", "list_directory"} <= tool_names + assert { + "read_file", + "write_file", + "edit_file", + "list_directory", + } <= tool_names def test_non_admin_manager_hides_admin_only_send_local_file_tool(): diff --git a/tests/test_builtin_skill_boundaries.py b/tests/test_builtin_skill_boundaries.py index 023e5acf..3a68fc13 100644 --- a/tests/test_builtin_skill_boundaries.py +++ b/tests/test_builtin_skill_boundaries.py @@ -27,6 +27,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None: "moviepilot-cli": "6", "moviepilot-update": "3", "transfer-failed-retry": "2", + "create-moviepilot-plugin": "3", } for skill_name, expected_version in expected_versions.items(): @@ -76,3 +77,18 @@ def test_agent_core_prompt_does_not_block_plugin_source_edits() -> None: assert "file editing tools, or generated patches to change code" not in core_prompt assert "write_file" in allowed_tools assert "edit_file" in allowed_tools + assert "search_web" in allowed_tools + assert "browse_webpage" in allowed_tools + + +def test_agent_core_prompt_routes_code_tools_safely() -> None: + """核心提示词应区分代码搜索、精确编辑和交互式命令场景。""" + core_prompt = CORE_PROMPT_PATH.read_text(encoding="utf-8") + + assert '`execute_command(action="run")` with `rg`' in core_prompt + assert "`replace_all=true` only when every match must change" in core_prompt + assert "Use `action=run` for short bounded commands" in core_prompt + assert "including SSH" in core_prompt + assert "Never use shell redirection" in core_prompt + assert "matching version of the official documentation" in core_prompt + assert "Do not guess signatures from memory" in core_prompt