""" 执行 Shell 命令工具。 """ from __future__ import annotations import asyncio import codecs import json import os import signal import subprocess from collections import deque from dataclasses import dataclass, field from tempfile import NamedTemporaryFile from typing import Any, Literal, Optional, TextIO, Type, cast from pydantic import BaseModel, Field from app.agent.shell import ( bind_agent_project_environment, build_agent_subprocess_env, resolve_agent_cwd, resolve_agent_shell, ) from app.agent.terminal.manager import ( TERMINAL_WAIT_DEFAULT_MS, TERMINAL_YIELD_DEFAULT_MS, get_terminal_session_manager, ) from app.agent.terminal.output import TERMINAL_DEFAULT_READ_BYTES, TERMINAL_MAX_READ_BYTES, TerminalOutputError from app.agent.terminal.ownership import TerminalAccessError, TerminalScope, require_terminal_scope from app.agent.tools.base import DEFAULT_TOOL_RESULT_MAX_CHARS, MoviePilotTool from app.agent.tools.impl._command_safety import validate_command_safety from app.agent.tools.tags import ToolTag from app.runtime.log import logger from app.runtime.settings import get_runtime_setting DEFAULT_TIMEOUT_SECONDS = 60 MAX_TIMEOUT_SECONDS = 300 MAX_OUTPUT_PREVIEW_BYTES = 32 * 1024 MAX_OUTPUT_HEAD_BYTES = 16 * 1024 MAX_OUTPUT_TAIL_BYTES = 16 * 1024 READ_CHUNK_SIZE = 4096 KILL_GRACE_SECONDS = 3 COMMAND_CONCURRENCY_LIMIT = 2 _command_semaphore = asyncio.Semaphore(COMMAND_CONCURRENCY_LIMIT) @dataclass class _CommandOutput: """ 保存命令头尾预览，并在超限时将完整输出写入临时文件。 :param preview_limit_bytes: 允许缓存的头部预览上限，单位为 UTF-8 字节 :param preview_entries: 头部预览的输出流名称和文本片段，按接收顺序保存 :param tail_entries: 尾部预览的输出流名称和文本片段，超过尾部上限时淘汰旧内容 :param captured_bytes: 当前头部预览累计的 UTF-8 字节数，初始为 0 :param tail_bytes: 当前尾部预览累计的 UTF-8 字节数，初始为 0 :param preview_truncated: 完整输出是否已超过头部预览上限 :param temp_file_path: 保存完整超限输出的日志路径；尚未创建时为 None :param temp_file_handle: 完整输出日志的打开句柄；尚未创建或已关闭时为 None :param last_written_stream: 日志最后写入的输出流名称，用于切换流时插入标题；初始为 None """ preview_limit_bytes: int preview_entries: list[tuple[str, str]] = field(default_factory=list) tail_entries: deque[tuple[str, str]] = field(default_factory=deque) captured_bytes: int = 0 tail_bytes: int = 0 preview_truncated: bool = False temp_file_path: Optional[str] = None temp_file_handle: Optional[TextIO] = None last_written_stream: Optional[str] = None @staticmethod def _clip_text_to_bytes(text: str, byte_limit: int) -> str: """ 按 UTF-8 字节数截断文本，避免截断后出现非法字符。 """ if byte_limit <= 0: return "" return text.encode("utf-8")[:byte_limit].decode("utf-8", errors="ignore") def _write_chunk(self, stream_name: str, text: str) -> None: """ 把输出分片按 stdout/stderr 分段写入临时文件。 """ if not self.temp_file_handle or not text: return if self.last_written_stream != stream_name: if self.temp_file_handle.tell() > 0: self.temp_file_handle.write("\n") title = "标准输出" if stream_name == "stdout" else "错误输出" self.temp_file_handle.write(f"[{title}]\n") self.last_written_stream = stream_name self.temp_file_handle.write(text) def _ensure_temp_file(self) -> None: """ 首次超出预览上限时创建临时文件并补写已缓存预览。 """ if self.temp_file_handle: return temp_file = NamedTemporaryFile( mode="w", encoding="utf-8", suffix=".log", prefix="moviepilot-command-", delete=False, ) self.temp_file_path = temp_file.name self.temp_file_handle = cast(TextIO, temp_file) for stream_name, chunk in self.preview_entries: self._write_chunk(stream_name, chunk) def close(self) -> None: """ 关闭临时文件句柄，确保输出落盘。 """ if not self.temp_file_handle: return self.temp_file_handle.flush() self.temp_file_handle.close() self.temp_file_handle = None def append(self, stream_name: str, text: str) -> None: """ 追加一段输出，超出预览上限后保留头尾预览和完整日志文件。 """ if not text: return self._append_tail(stream_name, text) if self.temp_file_handle: self._write_chunk(stream_name, text) return chunk_bytes = len(text.encode("utf-8")) remaining = self.preview_limit_bytes - self.captured_bytes if chunk_bytes <= remaining: self.preview_entries.append((stream_name, text)) self.captured_bytes += chunk_bytes return self.preview_truncated = True self._ensure_temp_file() self._write_chunk(stream_name, text) preview = self._clip_text_to_bytes(text, remaining) if preview: self.preview_entries.append((stream_name, preview)) self.captured_bytes += len(preview.encode("utf-8")) def _append_tail(self, stream_name: str, text: str) -> None: """ 维护固定字节大小的尾部输出，方便定位测试和构建失败信息。 """ self.tail_entries.append((stream_name, text)) self.tail_bytes += len(text.encode("utf-8")) while self.tail_bytes > MAX_OUTPUT_TAIL_BYTES and self.tail_entries: old_stream, old_text = self.tail_entries.popleft() old_bytes = len(old_text.encode("utf-8")) overflow = self.tail_bytes - MAX_OUTPUT_TAIL_BYTES if old_bytes <= overflow: self.tail_bytes -= old_bytes continue kept_text = old_text.encode("utf-8")[overflow:].decode( "utf-8", errors="ignore" ) kept_bytes = len(kept_text.encode("utf-8")) self.tail_bytes -= old_bytes if kept_text: self.tail_entries.appendleft((old_stream, kept_text)) self.tail_bytes += kept_bytes @staticmethod def _format_entries(entries: list[tuple[str, str]]) -> str: """ 按 stdout/stderr 切换插入可读的输出分段标题。 """ parts: list[str] = [] last_stream: Optional[str] = None for stream_name, text in entries: if stream_name != last_stream: title = "标准输出" if stream_name == "stdout" else "错误输出" parts.append(f"\n[{title}]\n") last_stream = stream_name parts.append(text) return "".join(parts).strip() @property def combined_preview(self) -> str: """ 返回完整输出或头尾组合预览。 """ if not self.preview_truncated: return self._format_entries(self.preview_entries) head_entries: list[tuple[str, str]] = [] remaining = MAX_OUTPUT_HEAD_BYTES for stream_name, text in self.preview_entries: if remaining <= 0: break clipped = self._clip_text_to_bytes(text, remaining) if clipped: head_entries.append((stream_name, clipped)) remaining -= len(clipped.encode("utf-8")) head = self._format_entries(head_entries) tail = self._format_entries(list(self.tail_entries)) return ( f"{head}\n\n...(中间输出已省略，完整内容在临时文件中)...\n\n{tail}" ).strip() @property def stdout(self) -> str: """ 返回当前保留的 stdout 预览。 """ return "".join( text for stream_name, text in self.preview_entries if stream_name == "stdout" ).strip() @property def stderr(self) -> str: """ 返回当前保留的 stderr 预览。 """ return "".join( text for stream_name, text in self.preview_entries if stream_name == "stderr" ).strip() class ExecuteCommandInput(BaseModel): """ 执行 Shell 命令工具的输入参数模型。 """ action: Optional[Literal['start', 'read', 'wait', 'write', 'interrupt', 'kill', 'run']] = Field( "start", description=( "命令动作：start 启动受管理后台会话并返回 session_id；" "read/wait/write/interrupt/kill 操作已有会话。" "interrupt 只发送一次中断，不升级为强杀；" "kill 终止会话；" "run 执行一次并等待完成或超时" ), ) command: Optional[str] = Field( None, description="start 或 run 必填的 Shell 命令", ) session_id: Optional[str] = Field( None, description="start 返回的命令会话标识", ) input_text: Optional[str] = Field( None, description="write 写入标准输入的文本。管道模式下控制字符作为普通字节；PTY 中 Ctrl+C/EOF 的含义由终端设置决定，发送信号应使用 interrupt", ) close_stdin: Optional[bool] = Field( False, strict=True, description="write 在管道模式下写入可选末段输入后关闭标准输入，使命令收到 EOF，输出仍可读取。空文本本身不表示 EOF；不支持 PTY 半关闭", ) signal_name: Optional[str] = Field( "TERM", description="kill 使用的信号名或编号，例如 TERM、INT、KILL 或 15；无效或不支持的信号会被拒绝", ) cwd: Optional[str] = Field( None, description="start/run 的启动目录；省略或相对路径以 MoviePilot 根目录解析，支持展开 ~", ) shell: Optional[str] = Field( None, description="start/run 使用的解释器可执行文件；管道和 PTY 使用相同选择策略并报告实际选择", ) login: Optional[bool] = Field( None, strict=True, description="start/run 是否显式启用登录模式。POSIX 默认非登录，Windows 保留配置默认值；登录启动文件可能改变环境和目录", ) env: Optional[dict[str, Any]] = Field( None, description="start 或 run 追加的环境变量", ) use_pty: Optional[bool] = Field( True, description="start 是否在支持的平台上使用伪终端，默认 True", ) since_seq: Optional[int] = Field( None, ge=0, strict=True, description="read/wait/write/interrupt/kill 中最后完整交付的输出序号；续读应使用 output_until_seq，不能使用 last_seq", ) since_offset: Optional[int] = Field( None, ge=0, strict=True, description=( "since_seq 之后分片内的 UTF-8 字节偏移。" "传 0 启用分片内分页，续读须同时使用 output_until_seq 和 output_until_offset；" "省略时沿用整分片读取" ), ) max_bytes: Optional[int] = Field( TERMINAL_DEFAULT_READ_BYTES, description="start/read/wait/write/interrupt/kill 最多返回的输出字节数", ) timeout_ms: Optional[int] = Field( TERMINAL_WAIT_DEFAULT_MS, description="wait 等待未读输出或输出完成的毫秒数；0 立即返回且不停止进程", ) yield_time_ms: Optional[int] = Field( TERMINAL_YIELD_DEFAULT_MS, ge=0, strict=True, description="start 首次等待输出的毫秒预算，默认 250，上限 10000；0 立即返回", ) timeout: Optional[int] = Field( 60, description="run 的最长执行秒数，默认 60", ) confirm_dangerous: Optional[bool] = Field( False, description="是否明确确认根目录递归删除、磁盘格式化、关机重启或破坏性权限修改等高风险命令", ) class ExecuteCommandTool(MoviePilotTool): """ 统一执行和管理 Shell 命令的 Agent 工具。 """ name: str = Field( default="execute_command", description="供模型调用的稳定工具名称", ) tags: list[str] = Field( default=[ToolTag.Read, ToolTag.Command, ToolTag.Admin], description="工具的能力和权限分类标签，初始化时会统一归一化", ) description: str = Field( default=( "在服务器启动并管理 Shell 命令。" "默认 action=start 创建后台会话，短暂等待初始输出后返回 session_id 与输出游标。" "续读须同时使用 output_until_seq 和 output_until_offset；" "last_seq 不表示已消费游标。" "使用 action=read、wait、write、interrupt 或 kill 轮询输出、分段短时等待、写入标准输入、发送一次中断或终止会话。" "管道模式下 write(close_stdin=true) 在可选末段输入后发送 EOF，PTY 输入输出不能半关闭。" "start/run 共用 cwd、shell 和 login 选择规则。" "只需一次有时限的命令结果时可用 action=run，返回包含 exit_code、timed_out、execution_outcome、输出预览及可选 output_f" "ile 的 JSON。" "只有正常以零退出才算成功，超时不会撤销已有副作用。" ), description="供模型理解工具用途、输入约束和调用边界的说明", ) args_schema: Type[BaseModel] = Field( default=ExecuteCommandInput, description="工具输入参数的验证模型或 JSON Schema", ) require_admin: bool = Field( default=True, description="是否仅允许系统管理员调用", ) result_max_chars = TERMINAL_MAX_READ_BYTES + 4096 def get_tool_message(self, **kwargs: Any) -> Optional[str]: """ 根据命令动作生成友好的提示消息。 """ action = kwargs.get("action") or "start" command = kwargs.get("command") session_id = kwargs.get("session_id") if action in {"start", "run"}: return f"执行系统命令: {command or ''}" if action == "read": return f"读取命令输出: {session_id or ''}" if action == "wait": return f"等待命令会话: {session_id or ''}" if action == "write": if kwargs.get("close_stdin"): return f"关闭命令输入: {session_id or ''}" return f"写入命令输入: {session_id or ''}" if action == "interrupt": return f"中断命令会话: {session_id or ''}" if action == "kill": return f"终止命令会话: {session_id or ''}" return f"处理命令会话: {session_id or command or ''}" @staticmethod def _dump(payload: dict[str, Any]) -> str: """ 把结构化命令会话结果转换为 Agent 容易解析的 JSON 字符串。 """ return json.dumps(payload, ensure_ascii=False, indent=2) @staticmethod def _require_session_id(session_id: Optional[str]) -> str: """ 校验会话型 action 必须传入 session_id。 """ if not session_id: raise ValueError("action 需要传入 session_id") return session_id @staticmethod def _require_command(command: Optional[str]) -> str: """ 校验启动型 action 必须传入 command。 """ if not command or not command.strip(): raise ValueError("action 需要传入 command") return command @staticmethod def _validate_command(command: str, *, confirmed: bool = False) -> None: """ 复用旧工具的基础危险命令过滤，避免明显破坏性命令进入 shell。 """ validate_command_safety(command, confirmed=confirmed) @staticmethod def _normalize_timeout(timeout: Optional[int]) -> tuple[int, Optional[str]]: """ 限制一次性执行命令的最长运行时间。 """ try: normalized = int(timeout or DEFAULT_TIMEOUT_SECONDS) except (TypeError, ValueError): normalized = DEFAULT_TIMEOUT_SECONDS if normalized <= 0: return DEFAULT_TIMEOUT_SECONDS, "timeout 参数无效，已使用默认 60 秒" if normalized > MAX_TIMEOUT_SECONDS: return ( MAX_TIMEOUT_SECONDS, f"timeout 参数超过上限，已从 {normalized} 秒限制为 {MAX_TIMEOUT_SECONDS} 秒", ) return normalized, None @staticmethod def _subprocess_kwargs() -> dict[str, Any]: """ 为一次性命令创建独立进程组，便于超时清理整棵子进程。 """ kwargs = { "stdin": subprocess.DEVNULL, "stdout": asyncio.subprocess.PIPE, "stderr": asyncio.subprocess.PIPE, } if os.name == "posix": kwargs["start_new_session"] = True elif os.name == "nt": kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP") return kwargs @staticmethod async def _read_stream( stream: asyncio.StreamReader, stream_name: str, output: _CommandOutput, ) -> None: """ 每个流增量解码 UTF-8，跨读取边界的字符不能被提前替换。 """ decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") while True: chunk = await stream.read(READ_CHUNK_SIZE) if not chunk: output.append(stream_name, decoder.decode(b"", final=True)) break output.append(stream_name, decoder.decode(chunk)) @staticmethod def _terminate_process(process: Any, sig: int) -> None: """ 向进程组发送终止信号，不支持进程组的平台回退为单进程终止。 """ try: if os.name == "posix": os.killpg(process.pid, sig) elif sig == getattr(signal, "SIGKILL", None): process.kill() else: process.terminate() except ProcessLookupError: pass @classmethod async def _cleanup_process( cls, process: Any, wait_task: asyncio.Task[int], ) -> None: """ 先温和终止，失败后强杀，避免超时 shell 遗留子进程。 """ if wait_task.done(): return cls._terminate_process(process, signal.SIGTERM) try: await asyncio.wait_for( asyncio.shield(wait_task), timeout=KILL_GRACE_SECONDS ) return except asyncio.TimeoutError: pass kill_signal = getattr(signal, "SIGKILL", signal.SIGTERM) cls._terminate_process(process, kill_signal) try: await asyncio.wait_for( asyncio.shield(wait_task), timeout=KILL_GRACE_SECONDS ) except asyncio.TimeoutError: logger.warning(f"命令进程强制清理超时: pid={process.pid}") @staticmethod async def _finish_reader_tasks(reader_tasks: list[asyncio.Task[None]]) -> None: """ 等待一次性命令输出读取任务退出，异常只记录不影响工具返回。 """ if not reader_tasks: return done, pending = await asyncio.wait(reader_tasks, timeout=1) for task in pending: task.cancel() results = await asyncio.gather(*done, *pending, return_exceptions=True) for result in results: if isinstance(result, Exception) and not isinstance( result, asyncio.CancelledError ): logger.debug(f"命令输出读取任务异常: {result}") @staticmethod def _format_run_result( *, exit_code: Optional[int], output: _CommandOutput, timeout: int, timed_out: bool, timeout_note: Optional[str], cwd: Optional[str] = None, shell: Optional[str] = None, login: bool = False, scope_cancelled: bool = False, ) -> str: """ 分开返回机器可判定的执行状态与有界输出，不能靠完成提示推断成功。 """ if scope_cancelled: result = "命令因任务作用域关闭而取消，未确认业务动作是否完成" elif exit_code is None: result = "无法确认命令进程已结束，请先核对实际状态" elif timed_out: result = f"命令执行超时 (限制: {timeout}秒，已终止进程)" else: result = f"命令执行完成 (退出码: {exit_code})" if timeout_note: result += f"\n\n提示:\n{timeout_note}" if output.temp_file_path: file_note = "截至返回时已捕获的输出" if exit_code is None else ("截至命令终止前的完整输出" if timed_out else "完整输出") result += ( "\n\n提示:\n" f"命令输出超过 {MAX_OUTPUT_PREVIEW_BYTES // 1024}KB，" f"仅返回前后各 {MAX_OUTPUT_HEAD_BYTES // 1024}KB 预览。\n" f"{file_note}已写入临时文件: {output.temp_file_path}\n" "如需完整内容，请继续读取该文件。" ) if output.preview_truncated: result += "\n\n...(仅展示前后各 16KB 内容)" if not output.combined_preview: result += "\n\n(无输出内容)" succeeded = exit_code == 0 and not timed_out and not scope_cancelled outcome = "failed" if scope_cancelled else ( "unknown" if exit_code is None else ("succeeded" if succeeded else "failed") ) recovery = ( "命令结果未知；先用 action=read 或 action=wait 核对会话状态，不要直接重复可能产生副作用的命令。" if outcome == "unknown" else "检查 output 和 exit_code，修正命令或参数后再重试；不要把命令输出中的完成文字当作成功证据。" ) return ExecuteCommandTool._dump({ "action": "run", "success": succeeded, "execution_outcome": outcome, "status": "cancelled" if scope_cancelled else ( "unknown" if exit_code is None else ("timed_out" if timed_out else "exited") ), "exit_code": exit_code, "timed_out": timed_out, "timeout": timeout, "cwd": cwd, "shell": shell, "login": login, "stdin_closed": True, "output_truncated": output.preview_truncated, "output_file": output.temp_file_path, "output": output.combined_preview, "message": result, "recovery": recovery, }) async def _run_once( self, *, command: str, timeout: Optional[int], cwd: Optional[str] = None, env: Optional[dict[str, Any]] = None, shell: Optional[str] = None, login: Optional[bool] = None, confirm_dangerous: bool = False, ) -> str: """ 一次性执行命令并返回结构化终态；退出路径都必须释放读取任务和归档句柄。 """ self._validate_command(command, confirmed=confirm_dangerous) scope = require_terminal_scope() scope.begin_run() try: return await self._run_once_with_scope( scope=scope, command=command, timeout=timeout, cwd=cwd, env=env, shell=shell, login=login, confirm_dangerous=confirm_dangerous, ) finally: scope.finish_run() @staticmethod async def _acquire_command_slot(scope: TerminalScope) -> None: """ 并发槽等待期间响应作用域封口，禁止取消后迟到启动一次性进程。 """ acquire_task = asyncio.create_task(_command_semaphore.acquire()) closed_task = asyncio.create_task(scope.changed.wait()) acquired = False released = False try: done, _ = await asyncio.wait( {acquire_task, closed_task}, return_when=asyncio.FIRST_COMPLETED, ) if closed_task in done and acquire_task not in done: acquire_task.cancel() await asyncio.gather(acquire_task, return_exceptions=True) raise TerminalAccessError() await acquire_task acquired = True if scope.closed: _command_semaphore.release() released = True acquired = False raise TerminalAccessError() finally: if not acquire_task.done(): acquire_task.cancel() await asyncio.gather(acquire_task, return_exceptions=True) if acquire_task.done() and not acquire_task.cancelled() and not acquired and not released: _command_semaphore.release() if not closed_task.done(): closed_task.cancel() await asyncio.gather(closed_task, return_exceptions=True) async def _run_once_with_scope( self, *, scope: TerminalScope, command: str, timeout: Optional[int], cwd: Optional[str] = None, env: Optional[dict[str, Any]] = None, shell: Optional[str] = None, login: Optional[bool] = None, confirm_dangerous: bool = False, ) -> str: """ 在已登记作用域下运行一次命令，并对封口和进程收尾保持可观察。 """ normalized_timeout, timeout_note = self._normalize_timeout(timeout) project_root = get_runtime_setting("ROOT_PATH") normalized_cwd = resolve_agent_cwd(cwd, root_path=project_root) normalized_env = bind_agent_project_environment( build_agent_subprocess_env(env), project_root=project_root, ) shell_policy = resolve_agent_shell(executable=shell, login=login, environment=normalized_env, cwd=normalized_cwd) await self._acquire_command_slot(scope) try: require_terminal_scope() process = await asyncio.create_subprocess_exec( *shell_policy.build_argv(command), cwd=normalized_cwd, env=normalized_env, **self._subprocess_kwargs(), ) output = _CommandOutput(preview_limit_bytes=MAX_OUTPUT_PREVIEW_BYTES) wait_task = asyncio.create_task(process.wait()) # _subprocess_kwargs 固定把两个输出流设为 PIPE，创建成功后均为 StreamReader。 reader_tasks = [ asyncio.create_task(self._read_stream(cast(asyncio.StreamReader, process.stdout), "stdout", output)), asyncio.create_task(self._read_stream(cast(asyncio.StreamReader, process.stderr), "stderr", output)), ] timed_out = False scope_cancelled = False scope_task = asyncio.create_task(scope.changed.wait()) try: done, _ = await asyncio.wait( {wait_task, scope_task}, timeout=normalized_timeout, return_when=asyncio.FIRST_COMPLETED, ) # wait_task 先完成后，作用域仍可能在同一轮事件循环内被封闭。 # 再次检查所有者的权威状态，避免跨过取消边界的命令被报告为成功。 scope_cancelled = scope_task in done or scope.closed if wait_task not in done: timed_out = not scope_cancelled await self._cleanup_process(process, wait_task) except asyncio.CancelledError: await self._cleanup_process(process, wait_task) raise finally: if not scope_task.done(): scope_task.cancel() await asyncio.gather(scope_task, return_exceptions=True) try: await self._finish_reader_tasks(reader_tasks) finally: output.close() return self._format_run_result( exit_code=process.returncode, output=output, timeout=normalized_timeout, timed_out=timed_out, scope_cancelled=scope_cancelled, timeout_note=timeout_note, cwd=normalized_cwd, shell=shell_policy.executable, login=shell_policy.login, ) finally: _command_semaphore.release() async def run( # type: ignore[override] self, action: Optional[str] = "start", command: Optional[str] = None, session_id: Optional[str] = None, input_text: Optional[str] = None, close_stdin: Optional[bool] = False, signal_name: Optional[str] = "TERM", cwd: Optional[str] = None, env: Optional[dict[str, Any]] = None, shell: Optional[str] = None, login: Optional[bool] = None, use_pty: Optional[bool] = True, since_seq: Optional[int] = None, since_offset: Optional[int] = None, max_bytes: Optional[int] = TERMINAL_DEFAULT_READ_BYTES, timeout_ms: Optional[int] = TERMINAL_WAIT_DEFAULT_MS, yield_time_ms: Optional[int] = TERMINAL_YIELD_DEFAULT_MS, timeout: Optional[int] = 60, confirm_dangerous: Optional[bool] = False, **kwargs: Any, ) -> str: """ 按同一启动策略执行命令，区分写入/EOF、一次中断与会话终止。 """ normalized_action = (action or "start").strip().lower() logger.info( f"执行工具: {self.name}, action={normalized_action}, " f"command={command}, session_id={session_id}" ) try: require_terminal_scope() terminal_session_manager = get_terminal_session_manager() output_budget = DEFAULT_TOOL_RESULT_MAX_CHARS if self.result_max_chars and self.result_max_chars > 0: output_budget = min(self.result_max_chars, output_budget) if normalized_action == "start": start_command = self._require_command(command) self._validate_command( start_command, confirmed=bool(confirm_dangerous), ) payload = await terminal_session_manager.start( command=start_command, cwd=cwd, env=env, shell=shell, login=login, use_pty=use_pty, confirm_dangerous=bool(confirm_dangerous), yield_time_ms=yield_time_ms, since_offset=since_offset, max_bytes=max_bytes, max_output_chars=output_budget, ) return self._dump(payload) if normalized_action == "read": payload = await terminal_session_manager.read( session_id=self._require_session_id(session_id), since_seq=since_seq, since_offset=since_offset, max_bytes=max_bytes, max_output_chars=output_budget, ) return self._dump(payload) if normalized_action == "wait": payload = await terminal_session_manager.wait( session_id=self._require_session_id(session_id), timeout_ms=timeout_ms, since_seq=since_seq, since_offset=since_offset, max_bytes=max_bytes, max_output_chars=output_budget, ) return self._dump(payload) if normalized_action == "write": payload = await terminal_session_manager.write( session_id=self._require_session_id(session_id), input_text=input_text or "", close_stdin=False if close_stdin is None else close_stdin, since_seq=since_seq, since_offset=since_offset, max_bytes=max_bytes, max_output_chars=output_budget, ) return self._dump(payload) if normalized_action == "interrupt": payload = await terminal_session_manager.interrupt( session_id=self._require_session_id(session_id), since_seq=since_seq, since_offset=since_offset, max_bytes=max_bytes, max_output_chars=output_budget, ) return self._dump(payload) if normalized_action == "kill": payload = await terminal_session_manager.kill( session_id=self._require_session_id(session_id), sig=signal_name, since_seq=since_seq, since_offset=since_offset, max_bytes=max_bytes, max_output_chars=output_budget, ) return self._dump(payload) if normalized_action == "run": return await self._run_once( command=self._require_command(command), timeout=timeout, cwd=cwd, env=env, shell=shell, login=login, confirm_dangerous=bool(confirm_dangerous), ) raise ValueError(f"不支持的 action: {action}") except TerminalAccessError as err: return self._dump({ "error": str(err), "status": "error", "action": normalized_action, "success": False, "execution_outcome": "failed", "code": "terminal_access_denied", "recovery": "当前任务没有该终端作用域权限；使用当前任务创建的 session_id，或重新启动受控会话。", }) except TerminalOutputError as err: return self._dump({ "error": str(err), "status": "error", "action": normalized_action, "success": False, "execution_outcome": "failed", "code": err.code, "minimum_read_bytes": err.minimum_read_bytes, "recovery": "按 minimum_read_bytes 调整读取游标和页大小后重试，不要重新启动命令。", }) except Exception as err: logger.error(f"执行命令 action 失败: {err}", exc_info=True) return self._dump({"error": str(err), "status": "error", "action": normalized_action, "success": False, "execution_outcome": "failed", "recovery": "修正 action、session_id 或参数后重试；不要重复启动未确认是否结束的命令。"})