diff --git a/app/agent/tools/impl/execute_command.py b/app/agent/tools/impl/execute_command.py index 70848f74d..203e9b2c8 100644 --- a/app/agent/tools/impl/execute_command.py +++ b/app/agent/tools/impl/execute_command.py @@ -226,7 +226,7 @@ class ExecuteCommandInput(BaseModel): ) env: Optional[dict[str, Any]] = Field( None, - description="Additional environment variables for action=start.", + description="Additional environment variables for action=start or action=run.", ) use_pty: Optional[bool] = Field( True, @@ -272,7 +272,9 @@ class ExecuteCommandTool(MoviePilotTool): "last_seq/output_until_seq. Call the same tool with action=read, wait, " "write, or kill to poll output, wait in short segments, send stdin, or " "terminate it. Use action=run only when a one-shot bounded command result " - "is preferred." + "is preferred. run returns JSON with exit_code, timed_out, execution_outcome, " + "output preview and optional output_file. Only a normal zero exit is success; " + "a timeout does not undo any side effects." ) args_schema: Type[BaseModel] = ExecuteCommandInput require_admin: bool = True @@ -428,8 +430,10 @@ class ExecuteCommandTool(MoviePilotTool): timed_out: bool, timeout_note: Optional[str], ) -> str: - """格式化 action=run 的兼容文本结果。""" - if timed_out: + """分开返回机器可判定的执行状态与有界输出，不能靠完成提示推断成功。""" + if exit_code is None: + result = "无法确认命令进程已结束，请先核对实际状态" + elif timed_out: result = f"命令执行超时 (限制: {timeout}秒，已终止进程)" else: result = f"命令执行完成 (退出码: {exit_code})" @@ -437,7 +441,7 @@ class ExecuteCommandTool(MoviePilotTool): if timeout_note: result += f"\n\n提示:\n{timeout_note}" if output.temp_file_path: - file_note = "截至命令终止前的完整输出" if timed_out else "完整输出" + file_note = "截至返回时已捕获的输出" if exit_code is None else ("截至命令终止前的完整输出" if timed_out else "完整输出") result += ( "\n\n提示:\n" f"命令输出超过 {MAX_OUTPUT_PREVIEW_BYTES // 1024}KB，" @@ -445,13 +449,19 @@ class ExecuteCommandTool(MoviePilotTool): f"{file_note}已写入临时文件: {output.temp_file_path}\n" "如需完整内容，请继续读取该文件。" ) - if output.combined_preview: - result += f"\n\n命令输出预览:\n{output.combined_preview}" if output.preview_truncated: result += "\n\n...(仅展示前后各 16KB 内容)" if not output.combined_preview: result += "\n\n(无输出内容)" - return result + succeeded = exit_code == 0 and not timed_out + outcome = "unknown" if exit_code is None else ("succeeded" if succeeded else "failed") + return ExecuteCommandTool._dump({ + "action": "run", "success": succeeded, "execution_outcome": outcome, + "status": "unknown" if exit_code is None else ("timed_out" if timed_out else "exited"), + "exit_code": exit_code, "timed_out": timed_out, "timeout": timeout, + "output_truncated": output.preview_truncated, "output_file": output.temp_file_path, + "output": output.combined_preview, "message": result, + }) async def _run_once( self, @@ -459,9 +469,10 @@ class ExecuteCommandTool(MoviePilotTool): command: str, timeout: Optional[int], cwd: Optional[str] = None, + env: Optional[dict[str, Any]] = None, confirm_dangerous: bool = False, ) -> str: - """按旧模式一次性执行命令，等待完成或超时后返回文本结果。""" + """一次性执行命令并返回结构化终态；退出路径都必须释放读取任务和归档句柄。""" self._validate_command(command, confirmed=confirm_dangerous) normalized_timeout, timeout_note = self._normalize_timeout(timeout) @@ -471,13 +482,14 @@ class ExecuteCommandTool(MoviePilotTool): process = await asyncio.create_subprocess_exec( *shell.build_argv(command), cwd=cwd, - env=build_agent_subprocess_env(), + env=build_agent_subprocess_env(env), **self._subprocess_kwargs(), ) else: process = await asyncio.create_subprocess_shell( command, cwd=cwd, + env=build_agent_subprocess_env(env), **self._subprocess_kwargs(), ) output = _CommandOutput(preview_limit_bytes=MAX_OUTPUT_PREVIEW_BYTES) @@ -499,10 +511,11 @@ class ExecuteCommandTool(MoviePilotTool): await self._cleanup_process(process, wait_task) raise - try: - await self._finish_reader_tasks(reader_tasks) finally: - output.close() + try: + await self._finish_reader_tasks(reader_tasks) + finally: + output.close() return self._format_run_result( exit_code=process.returncode, @@ -589,6 +602,7 @@ class ExecuteCommandTool(MoviePilotTool): command=self._require_command(command), timeout=timeout, cwd=cwd, + env=env, confirm_dangerous=bool(confirm_dangerous), ) diff --git a/docs/agent.md b/docs/agent.md index 41f3fbba4..95d11d00f 100644 --- a/docs/agent.md +++ b/docs/agent.md @@ -28,6 +28,8 @@ MoviePilot Agent 通过模型、Skills、工具和会话状态共同完成任务 超长结果在当前会话图中临时保存，预览带 `result_id` 和 `next_offset`。Agent 使用 `read_tool_result` 按 Unicode 字符位置续读，无需重新执行原工具。每条完整结果最多 1 MiB，每个图最多 8 条、合计 4 MiB，15 分钟后过期；容量超限、图重建和进程退出也会使编号失效。无法保存的结果明确要求缩小查询范围。不同图线程不能互读，管理员结果在降权后不能续读。结果原文只短期驻留内存，不归档到磁盘或日志。 +`execute_command(action="run")` 返回结构化 JSON：`exit_code`、`timed_out`、`execution_outcome` 和 `status` 表示实际执行结果，`output` 保存输出预览，`output_file` 指向超长输出的临时归档。仅正常退出且退出码为 0 时成功；非零退出或已停止的超时命令为失败，无法确认进程结束时为未知，不能凭“有输出”判断成功。超时和取消不会撤销命令已经产生的外部副作用。取消继续向外传播，同时回收输出读取任务、关闭归档文件；`run` 与 `start` 都支持 `env`，并保留指定工作目录。 + ## 写工具的持久执行记录 生产 Agent 在副作用之前向 `agentinvocation` 表提交原子认领。记录只包含用户、会话、工具身份、参数指纹、执行 token、状态和固定宿主文案，不保存原始参数或工具输出。无法认领时不执行写入；相同调用 ID 携带不同参数时拒绝执行。 diff --git a/docs/mcp-api.md b/docs/mcp-api.md index c2ea2da51..a5234bdaf 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -513,6 +513,13 @@ MCP、HTTP 工具管理接口、本地 CLI 和内置 Agent 都从同一严格目 `search_web` 不通过 MCP 暴露。隐藏列表只负责收敛接口暴露面，不替代各工具自身的 权限、路径和网络边界。 +内置命令工具 `execute_command(action="run")` 的返回值为 JSON 字符串，包含 +`success`、`execution_outcome`、`status`、`exit_code`、`timed_out`、`timeout`、 +`output_truncated`、`output_file`、`output` 和 `message`。正常退出码 0 才是成功， +非零退出及已停止的超时执行为失败，无法确认进程结束时为 unknown。输出预览与状态分开， +不再通过中文完成提示判断成功。后台 `start/read/wait/write/kill` 保持会话状态与游标协议； +`env` 同时适用于 `start` 和 `run`，此工具仍不通过 MCP 暴露。 + 下载器和媒体服务器的第三方原生高级能力不注册成永久 MCP 工具。内置 Agent 按需 加载 `downloader-operation` 或 `mediaserver-operation` Skill，通过固定脚本读取本机 配置、发现 provider 能力并调用受控 action；脚本不接受任意 URL、认证信息或任意 diff --git a/skills/create-moviepilot-plugin/SKILL.md b/skills/create-moviepilot-plugin/SKILL.md index 0a0134467..1911ec775 100644 --- a/skills/create-moviepilot-plugin/SKILL.md +++ b/skills/create-moviepilot-plugin/SKILL.md @@ -66,6 +66,10 @@ a local plugin source and installed into the running MoviePilot instance. - 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. +- Read the structured command result: only `execution_outcome="succeeded"` + with `exit_code=0` means a completed command passed. Inspect `output` on + failure and `output_file` for a truncated log. A timeout or unknown result + does not undo side effects and must not trigger a blind retry of a write. - 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 diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index 95105afff..44711cc37 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/tests/fixtures/architecture/ruff-baseline.json @@ -676,9 +676,6 @@ "tests/test_event_dispatch_snapshot.py": { "I001": 1 }, - "tests/test_execute_command_tool.py": { - "I001": 1 - }, "tests/test_extensible_media_source_migration.py": { "I001": 1 }, diff --git a/tests/test_agent_command_outcome.py b/tests/test_agent_command_outcome.py new file mode 100644 index 000000000..5c2e07ed5 --- /dev/null +++ b/tests/test_agent_command_outcome.py @@ -0,0 +1,237 @@ +"""真实短命令通过 Agent 图与直调入口传播退出、超时、未知及取消语义。""" + +import asyncio +import json +import os +import shlex +import subprocess +import sys +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock + +import pytest +from langchain.agents import create_agent +from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage + +from app.agent.middleware.policy import AgentPolicyMiddleware +from app.agent.policy.contracts import ( + AuthSource, + ExecutionOutcome, + PrincipalType, + ToolOrigin, + ToolPolicyContext, +) +from app.agent.policy.orchestrator import AgentToolPolicyOrchestrator +from app.agent.tools.impl import execute_command as command_module +from app.agent.tools.impl.execute_command import ExecuteCommandTool, _CommandOutput +from app.agent.tools.manager import MoviePilotToolsManager +from app.agent.tools.result import EXECUTION_OUTCOME_KEY, inspect_tool_result + + +def _command(code: str) -> str: + """固定使用当前虚拟环境解释器，并对每个 shell 参数独立转义。""" + arguments = [sys.executable, "-c", code] + return subprocess.list2cmdline(arguments) if os.name == "nt" else shlex.join(arguments) + + +def _tool() -> ExecuteCommandTool: + """工具只拥有本测试会话的显式管理员上下文，不查询真实用户或渠道。""" + tool = ExecuteCommandTool(session_id="command-outcome-test", user_id="command-test-owner") + tool.set_agent_context({"is_admin": True}) + return tool + + +class _CommandModel(FakeMessagesListChatModel): + """按固定两轮消息驱动生产 ToolNode，完全不调用真实模型。""" + + def bind_tools(self, _tools: Any, **_kwargs: Any) -> "_CommandModel": + """接收图内实际工具绑定，命令由真实执行器运行。""" + return self + + +class _RecordingPolicy(AgentToolPolicyOrchestrator): + """保留生产回执生成逻辑，只额外记录测试可核验的终态。""" + + def __init__(self) -> None: + """为每个测试创建独立回执列表。""" + super().__init__() + self.receipts: list[Any] = [] + + def finish(self, observation: Any, result: Any) -> Any: + """通过生产判定生成回执，避免测试自行推断工具状态。""" + receipt = super().finish(observation, result) + self.receipts.append(receipt) + return receipt + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("exit_code", "timed_out"), [(0, False), (7, False), (0, True)]) +async def test_real_graph_command_outcomes_reach_message_receipt_and_log( + monkeypatch: pytest.MonkeyPatch, exit_code: int, timed_out: bool, +) -> None: + """真实非零退出和已终止超时必须成为失败工具消息、失败回执及一致日志。""" + tool = _tool() + code = "import sys, time; print('stdout-probe', flush=True); print('stderr-probe', file=sys.stderr, flush=True); " + code += "time.sleep(3)" if timed_out else f"raise SystemExit({exit_code})" + arguments = {"action": "run", "command": _command(code), "timeout": 1 if timed_out else 5} + model = _CommandModel(responses=[ + AIMessage(content="", tool_calls=[{"id": "command", "name": tool.name, "args": arguments}]), + AIMessage(content="命令结果已读取。"), + ]) + policy = _RecordingPolicy() + context = ToolPolicyContext( + session_id="command-outcome-test", user_id="command-test-owner", origin=ToolOrigin.AGENT_INTERACTIVE, + principal_type=PrincipalType.HUMAN, auth_source=AuthSource.INTERNAL, agent_context={"is_admin": True}, + ) + logs: list[str] = [] + monkeypatch.setattr(command_module.logger, "info", logs.append) + graph = create_agent(model=model, tools=[tool], middleware=[AgentPolicyMiddleware(context=context, orchestrator=policy)]) + result = await graph.ainvoke({"messages": [HumanMessage(content="执行测试命令")]}) + message = next(item for item in result["messages"] if isinstance(item, ToolMessage)) + expected = ExecutionOutcome.FAILED if timed_out or exit_code else ExecutionOutcome.SUCCEEDED + assert message.status == ("error" if expected is ExecutionOutcome.FAILED else "success") + assert message.additional_kwargs[EXECUTION_OUTCOME_KEY] == expected.value + assert policy.receipts[0].outcome is expected + assert any(f"outcome={expected.value}" in line for line in logs) + payload = json.loads(message.content) + assert payload["action"] == "run" + assert payload["success"] is (expected is ExecutionOutcome.SUCCEEDED) + assert payload["execution_outcome"] == expected.value + assert payload["status"] == ("timed_out" if timed_out else "exited") + assert payload["timed_out"] is timed_out + assert payload["exit_code"] is not None + if not timed_out: + assert payload["exit_code"] == exit_code + assert "stdout-probe" in payload["output"] + assert "stderr-probe" in payload["output"] + assert "stdout-probe" not in payload["message"] + assert "stderr-probe" not in payload["message"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("exit_code", [0, 7]) +async def test_direct_tool_manager_keeps_command_outcome(exit_code: int) -> None: + """HTTP/MCP 使用的直调工具入口也必须把真实退出状态交给同一回执判定。""" + policy = _RecordingPolicy() + manager = MoviePilotToolsManager( + session_id="command-direct-test", user_id="command-test-owner", is_admin=True, policy_orchestrator=policy, + ) + tool = _tool() + manager.tools = [tool] + result = await manager.call_tool(tool.name, { + "action": "run", "command": _command(f"raise SystemExit({exit_code})"), "timeout": 5, + }) + expected = ExecutionOutcome.FAILED if exit_code else ExecutionOutcome.SUCCEEDED + assert inspect_tool_result(result) is expected + assert policy.receipts[0].outcome is expected + assert json.loads(result)["exit_code"] == exit_code + + +@pytest.mark.asyncio +async def test_run_honors_working_directory_and_environment(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """一次性命令必须在指定目录看到调用方提供的环境变量，父进程不能被污染。""" + monkeypatch.delenv("MOVIEPILOT_COMMAND_TEST_VALUE", raising=False) + result = await _tool().run( + action="run", command=_command("import os; print(os.getcwd()); print(os.environ.get('MOVIEPILOT_COMMAND_TEST_VALUE', 'MISSING'))"), + cwd=str(tmp_path), env={"MOVIEPILOT_COMMAND_TEST_VALUE": "command-env-marker"}, timeout=5, + ) + payload = json.loads(result) + assert payload["execution_outcome"] == "succeeded" + assert str(tmp_path.resolve()) in payload["output"] + assert "command-env-marker" in payload["output"] + assert "MOVIEPILOT_COMMAND_TEST_VALUE" not in os.environ + + +@pytest.mark.parametrize("timed_out", [False, True]) +def test_missing_exit_status_is_unknown_without_claiming_process_terminated(timed_out: bool) -> None: + """无法取得真实退出码时必须保留未知，超时标志也不能证明进程已结束。""" + output = _CommandOutput(preview_limit_bytes=1024) + output.append("stdout", "partial-command-output") + result = ExecuteCommandTool._format_run_result( + exit_code=None, output=output, timeout=1, timed_out=timed_out, timeout_note=None, + ) + assert inspect_tool_result(result) is ExecutionOutcome.UNKNOWN + payload = json.loads(result) + assert payload["success"] is False + assert payload["execution_outcome"] == "unknown" + assert payload["status"] == "unknown" + assert payload["exit_code"] is None + assert payload["timed_out"] is timed_out + assert "已终止" not in payload["message"] + assert "partial-command-output" in payload["output"] + + +@pytest.mark.asyncio +async def test_cancelled_run_finishes_readers_and_closes_output_archive( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + """取消真实命令必须传播取消，同时回收读取任务并关闭已经创建的输出归档。""" + captured: list[_CommandOutput] = [] + original_append = _CommandOutput.append + original_finish = ExecuteCommandTool._finish_reader_tasks + archive_ready = asyncio.Event() + reader_groups: list[list[asyncio.Task[Any]]] = [] + + def append(output: _CommandOutput, stream: str, text: str) -> None: + """在真实归档已创建时通知测试发出取消，不以固定休眠猜测子进程进度。""" + original_append(output, stream, text) + if output.temp_file_handle and not captured: + captured.append(output) + archive_ready.set() + + async def finish(readers: list[asyncio.Task[Any]]) -> None: + """记录实际读取任务，再执行原有收尾流程。""" + reader_groups.append(readers) + await original_finish(readers) + + def temporary_file(**kwargs: Any) -> Any: + """将命令归档限定在 pytest 临时目录，避免遗留系统临时文件。""" + from tempfile import NamedTemporaryFile + + return NamedTemporaryFile(dir=tmp_path, **kwargs) + + monkeypatch.setattr(_CommandOutput, "append", append) + monkeypatch.setattr(ExecuteCommandTool, "_finish_reader_tasks", staticmethod(finish)) + monkeypatch.setattr(command_module, "NamedTemporaryFile", temporary_file) + task = asyncio.create_task(_tool().run( + action="run", command=_command("import sys, time; sys.stdout.write('x' * 70000); sys.stdout.flush(); time.sleep(20)"), timeout=30, + )) + try: + await asyncio.wait_for(archive_ready.wait(), timeout=5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert reader_groups + assert all(reader.done() for readers in reader_groups for reader in readers) + assert captured[0].temp_file_handle is None or captured[0].temp_file_handle.closed + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + for output in captured: + output.close() + + +@pytest.mark.asyncio +async def test_windows_shell_branch_receives_run_environment(monkeypatch: pytest.MonkeyPatch) -> None: + """无需 Windows 宿主也验证显式 shell 分支不会丢失本次 env 覆盖值。""" + process = await asyncio.create_subprocess_exec(sys.executable, "-c", "pass", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE) + original_environment = command_module.build_agent_subprocess_env + create_process = AsyncMock(return_value=process) + + class Shell: + """提供与 Windows shell 策略相同的参数构造边界。""" + + @staticmethod + def build_argv(_command_text: str) -> list[str]: + """真实子进程已在隔离环境启动，此处只检验启动参数。""" + return [sys.executable, "-c", "pass"] + + monkeypatch.setattr(command_module, "resolve_agent_shell", lambda: Shell()) + monkeypatch.setattr(command_module.asyncio, "create_subprocess_exec", create_process) + result = await _tool().run(action="run", command="unused", env={"MOVIEPILOT_COMMAND_TEST_VALUE": "shell-env-marker"}, timeout=5) + assert create_process.call_args.kwargs["env"]["MOVIEPILOT_COMMAND_TEST_VALUE"] == "shell-env-marker" + assert create_process.call_args.kwargs["env"] == original_environment({"MOVIEPILOT_COMMAND_TEST_VALUE": "shell-env-marker"}) + assert inspect_tool_result(result) is ExecutionOutcome.SUCCEEDED diff --git a/tests/test_execute_command_tool.py b/tests/test_execute_command_tool.py index fa9babf6d..b4b065102 100644 --- a/tests/test_execute_command_tool.py +++ b/tests/test_execute_command_tool.py @@ -1,7 +1,6 @@ import asyncio import json import os -import re import shlex import subprocess import sys @@ -10,8 +9,8 @@ import unittest from unittest.mock import patch from app.agent.tools.impl.execute_command import ( - ExecuteCommandTool, MAX_OUTPUT_PREVIEW_BYTES, + ExecuteCommandTool, ) @@ -24,11 +23,13 @@ def _python_command(code: str) -> str: class TestExecuteCommandTool(unittest.TestCase): + """验证一次性命令的输出保留、超时收尾与参数安全。""" + def _temp_file_path_from_result(self, result: str) -> str: - """从工具返回文本中提取完整输出临时文件路径。""" - match = re.search(r"临时文件: (.+)", result) - self.assertIsNotNone(match) - return match.group(1).strip() + """从结构化回执读取完整归档路径，不再解析人类提示文案。""" + path = json.loads(result)["output_file"] + self.assertTrue(path) + return path def _run_command(self, command: str, timeout: int = 60) -> str: """按一次性执行模式运行命令，兼容旧测试断言。""" @@ -167,6 +168,8 @@ class TestExecuteCommandTool(unittest.TestCase): class TestExecuteCommandSessionTool(unittest.IsolatedAsyncioTestCase): + """验证后台终端会话的启动、增量读取、交互和终止。""" + async def asyncSetUp(self): """创建每个测试复用的统一命令工具。""" self.tool = ExecuteCommandTool(session_id="session-1", user_id="10001")