fix(agent): report command outcomes and clean up cancelled runs

This commit is contained in:
jxxghp
2026-09-09 23:06:26 +08:00
parent 17eaf47c34
commit c3db7531f8
7 changed files with 286 additions and 22 deletions

View File

@@ -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),
)

View File

@@ -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 携带不同参数时拒绝执行。

View File

@@ -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、认证信息或任意

View File

@@ -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

View File

@@ -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
},

View File

@@ -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

View File

@@ -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")