Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 30 additions & 11 deletions astrbot/core/astr_main_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -428,7 +428,11 @@ async def _apply_workspace_extra_prompt(
)


def _apply_local_env_tools(req: ProviderRequest, plugin_context: Context) -> None:
def _apply_local_env_tools(
req: ProviderRequest,
plugin_context: Context,
windows_shell: str | None = None,
) -> None:
if req.func_tool is None:
req.func_tool = ToolSet()
tool_mgr = plugin_context.get_llm_tool_manager()
Expand All @@ -439,18 +443,29 @@ def _apply_local_env_tools(req: ProviderRequest, plugin_context: Context) -> Non
req.func_tool.add_tool(tool_mgr.get_builtin_tool(FileWriteTool))
req.func_tool.add_tool(tool_mgr.get_builtin_tool(FileEditTool))
req.func_tool.add_tool(tool_mgr.get_builtin_tool(GrepTool))
req.system_prompt = f"{req.system_prompt or ''}\n{_build_local_mode_prompt()}\n"
req.system_prompt = (
f"{req.system_prompt or ''}\n{_build_local_mode_prompt(windows_shell)}\n"
)


def _build_local_mode_prompt() -> str:
def _build_local_mode_prompt(windows_shell: str | None = None) -> str:
system_name = platform.system() or "Unknown"
shell_hint = (
"The runtime shell is Windows PowerShell 5.1 (powershell.exe). "
"Use Windows PowerShell 5.1-compatible syntax and cmdlets; do not use "
"PowerShell 7-only syntax or assume Unix commands like cat/ls/grep are available."
if system_name.lower() == "windows"
else "The runtime shell is Unix-like. Use POSIX-compatible shell commands."
)
if system_name.lower() != "windows":
shell_hint = (
"The runtime shell is Unix-like. Use POSIX-compatible shell commands."
)
elif windows_shell and "pwsh" in windows_shell.lower():
shell_hint = (
"The runtime shell is PowerShell 7 (pwsh.exe). "
"Use PowerShell 7-compatible syntax and cmdlets; do not "
"assume Unix commands like cat/ls/grep are available."
)
else:
shell_hint = (
"The runtime shell is Windows PowerShell 5.1 (powershell.exe). "
"Use Windows PowerShell 5.1-compatible syntax and cmdlets; do not use "
"PowerShell 7-only syntax or assume Unix commands like cat/ls/grep are available."
)
return (
"You have access to the host local environment and can execute shell commands and Python code. "
f"Current operating system: {system_name}. "
Expand Down Expand Up @@ -1584,7 +1599,11 @@ async def build_main_agent(
if config.computer_use_runtime == "sandbox":
_apply_sandbox_tools(config, req, req.session_id)
elif config.computer_use_runtime == "local":
_apply_local_env_tools(req, plugin_context)
_apply_local_env_tools(
req,
plugin_context,
config.provider_settings.get("windows_shell"),
)

agent_runner = AgentRunner()
astr_agent_ctx = AstrAgentContext(
Expand Down
28 changes: 22 additions & 6 deletions astrbot/core/computer/booters/local.py
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,7 @@ async def exec(
timeout: int | None = 300,
shell: bool = True,
background: bool = False,
windows_shell: str | None = None,
) -> dict[str, Any]:
if not _is_safe_command(command):
raise PermissionError("Blocked unsafe shell command.")
Expand All @@ -146,8 +147,15 @@ def _run() -> dict[str, Any]:
popen_command: str | list[str] = command
popen_shell = shell
if sys.platform == "win32" and shell:
shell_executable = windows_shell or "powershell.exe"
if os.name == "nt" and shutil.which(shell_executable) is None:
raise RuntimeError(
f"The Windows PowerShell executable '{shell_executable}' "
"was not found on PATH. Install it or reset the "
"'Windows PowerShell version' setting."
)
popen_command = [
"powershell.exe",
shell_executable,
"-NoLogo",
"-NoProfile",
"-NonInteractive",
Expand All @@ -156,8 +164,8 @@ def _run() -> dict[str, Any]:
]
popen_shell = False
if background:
# Shell commands use PowerShell 5.1 on Windows and the platform
# shell elsewhere. Safety relies on `_is_safe_command()`.
# Shell commands use the configured PowerShell on Windows and the
# platform shell elsewhere. Safety relies on `_is_safe_command()`.
proc = subprocess.Popen( # noqa: S602 # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit
popen_command,
shell=popen_shell,
Expand All @@ -167,8 +175,8 @@ def _run() -> dict[str, Any]:
stderr=subprocess.DEVNULL,
)
return {"pid": proc.pid, "stdout": "", "stderr": "", "exit_code": None}
# Shell commands use PowerShell 5.1 on Windows and the platform shell
# elsewhere. Safety relies on `_is_safe_command()`.
# Shell commands use the configured PowerShell on Windows and the
# platform shell elsewhere. Safety relies on `_is_safe_command()`.
proc = subprocess.Popen( # noqa: S602 # nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit
popen_command,
shell=popen_shell,
Expand Down Expand Up @@ -223,6 +231,7 @@ async def exec_managed(
timeout: int | None = None,
yield_time_ms: int = 10_000,
max_output_chars: int = 10_000,
windows_shell: str | None = None,
) -> dict[str, Any]:
"""Start a locally managed shell process and briefly wait for it.

Expand Down Expand Up @@ -278,8 +287,15 @@ async def exec_managed(
try:
if sys.platform == "win32":
process_factory = asyncio.create_subprocess_exec
shell_executable = windows_shell or "powershell.exe"
if os.name == "nt" and shutil.which(shell_executable) is None:
raise RuntimeError(
f"The Windows PowerShell executable '{shell_executable}' "
"was not found on PATH. Install it or reset the "
"'Windows PowerShell version' setting."
)
process_args = (
"powershell.exe",
shell_executable,
"-NoLogo",
"-NoProfile",
"-NonInteractive",
Expand Down
11 changes: 11 additions & 0 deletions astrbot/core/config/default.py
Original file line number Diff line number Diff line change
Expand Up @@ -171,6 +171,7 @@
"add_cron_tools": True,
},
"computer_use_runtime": "none",
"windows_shell": "powershell.exe",
"computer_use_require_admin": True,
"sandbox": {
"booter": "shipyard_neo",
Expand Down Expand Up @@ -3465,6 +3466,16 @@
"labels": ["无", "本地", "沙箱"],
"hint": "选择 Computer Use 运行环境。",
},
"provider_settings.windows_shell": {
"description": "Windows PowerShell 版本",
"type": "string",
"options": ["powershell.exe", "pwsh.exe"],
"labels": ["Windows PowerShell 5.1", "PowerShell 7"],
"hint": "Windows 下本地运行时使用的 Shell。PowerShell 7 需自行安装。",
"condition": {
"provider_settings.computer_use_runtime": "local",
},
},
"provider_settings.computer_use_require_admin": {
"description": "需要 AstrBot 管理员权限",
"type": "bool",
Expand Down
5 changes: 5 additions & 0 deletions astrbot/core/tools/computer_tools/shell.py
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,10 @@ async def call(
if not creator_id:
return "Error executing command: sender identity is unavailable."
started_at = monotonic()
cfg = context.context.context.get_config(
umo=context.context.event.unified_msg_origin
)
windows_shell = cfg.get("provider_settings", {}).get("windows_shell")
result = await sb.shell.exec_managed(
command,
owner_id=context.context.event.unified_msg_origin,
Expand All @@ -134,6 +138,7 @@ async def call(
env=env,
timeout=timeout,
yield_time_ms=0 if background else yield_time_ms,
windows_shell=str(windows_shell) if windows_shell else None,
)
elapsed_seconds = monotonic() - started_at
if result.get("session_closed") and result.get("status") in {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -170,6 +170,11 @@
"description": "Computer Use Runtime",
"hint": "Environment allowed for Agent usage. `local` means the local machine environment, `sandbox` means the sandbox environment, and `none` means no environment is allowed."
},
"windows_shell": {
"description": "Windows PowerShell Version",
"hint": "Shell used by the Windows local runtime. PowerShell 7 must be installed separately.",
"labels": ["Windows PowerShell 5.1", "PowerShell 7"]
},
"computer_use_require_admin": {
"description": "Require AstrBot Admin Permission",
"hint": "When enabled, AstrBot admin permission is required to use computer capabilities. Admins can be added in Platform Config. Use the /sid command to view admin IDs."
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -170,6 +170,11 @@
"description": "Среда выполнения (Runtime)",
"hint": "Среда, к которой разрешён доступ Agent: `local` — локальная среда компьютера, `sandbox` — изолированная песочница, `none` — запретить доступ к любым средам."
},
"windows_shell": {
"description": "Версия Windows PowerShell",
"hint": "Оболочка Windows Local runtime. PowerShell 7 нужно устанавливать отдельно.",
"labels": ["Windows PowerShell 5.1", "PowerShell 7"]
},
"computer_use_require_admin": {
"description": "Требовать права администратора AstrBot",
"hint": "Если включено, только администраторы смогут использовать возможности управления компьютером. Добавить администраторов можно в конфиге платформы."
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -172,6 +172,11 @@
"description": "运行环境",
"hint": "允许 Agent 访问的环境。local 为本机环境,sandbox 为沙箱环境,none 为不允许任何环境。"
},
"windows_shell": {
"description": "Windows PowerShell 版本",
"hint": "Windows 下本地运行时使用的 Shell。PowerShell 7 需自行安装。",
"labels": ["Windows PowerShell 5.1", "PowerShell 7"]
},
"computer_use_require_admin": {
"description": "需要 AstrBot 管理员权限",
"hint": "开启后,需要 AstrBot 管理员权限才能调用使用电脑能力。在平台配置->管理员中可添加管理员。使用 /sid 指令查看管理员 ID。"
Expand Down
140 changes: 136 additions & 4 deletions tests/test_local_shell_component.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,73 @@ def fake_run(*args, **kwargs):
assert calls[0][1]["shell"] is False


def test_local_shell_component_uses_pwsh_when_configured(monkeypatch):
calls = []

def fake_run(*args, **kwargs):
calls.append((args, kwargs))
return _FakePopen(stdout=b"")

monkeypatch.setattr(subprocess, "Popen", fake_run)
monkeypatch.setattr(local_booter.sys, "platform", "win32")

result = asyncio.run(
LocalShellComponent().exec("Get-ChildItem", windows_shell="pwsh.exe")
)

assert result["exit_code"] == 0
assert calls[0][0][0] == [
"pwsh.exe",
"-NoLogo",
"-NoProfile",
"-NonInteractive",
"-Command",
"Get-ChildItem",
]
assert calls[0][1]["shell"] is False


def test_exec_raises_when_windows_shell_missing(monkeypatch):
monkeypatch.setattr(local_booter.sys, "platform", "win32")
monkeypatch.setattr(local_booter.os, "name", "nt")
monkeypatch.setattr(local_booter.shutil, "which", lambda _cmd: None)

with pytest.raises(RuntimeError, match="pwsh.exe"):
asyncio.run(
LocalShellComponent().exec("Get-ChildItem", windows_shell="pwsh.exe")
)


def test_exec_skips_windows_shell_check_outside_windows(monkeypatch):
calls = []

def fake_run(*args, **kwargs):
calls.append((args, kwargs))
return _FakePopen(stdout=b"")

def fail_if_called(_cmd):
raise AssertionError("shutil.which must not be called outside Windows")

monkeypatch.setattr(subprocess, "Popen", fake_run)
monkeypatch.setattr(local_booter.sys, "platform", "win32")
monkeypatch.setattr(local_booter.os, "name", "posix")
monkeypatch.setattr(local_booter.shutil, "which", fail_if_called)

result = asyncio.run(
LocalShellComponent().exec("Get-ChildItem", windows_shell="pwsh.exe")
)

assert result["exit_code"] == 0
assert calls[0][0][0] == [
"pwsh.exe",
"-NoLogo",
"-NoProfile",
"-NonInteractive",
"-Command",
"Get-ChildItem",
]


def test_local_shell_component_keeps_platform_shell_outside_windows(monkeypatch):
calls = []

Expand Down Expand Up @@ -156,6 +223,71 @@ async def fail_create_subprocess_shell(*_args, **_kwargs):
assert "creationflags" in calls[0][1]


@pytest.mark.asyncio
async def test_managed_shell_uses_pwsh_when_configured(monkeypatch, tmp_path):
calls = []

class FakeStdout:
def __init__(self):
self.chunks = [b"done\n", b""]

async def read(self, _limit):
return self.chunks.pop(0)

class FakeProcess:
def __init__(self):
self.pid = 12345
self.returncode = None
self.stdout = FakeStdout()
self.stdin = None

async def wait(self):
self.returncode = 0
return 0

async def fake_create_subprocess_exec(*args, **kwargs):
calls.append((args, kwargs))
return FakeProcess()

async def fail_create_subprocess_shell(*_args, **_kwargs):
raise AssertionError("Windows managed commands must not use cmd.exe.")

monkeypatch.setattr(local_booter.sys, "platform", "win32")
monkeypatch.setattr(
local_booter.asyncio,
"create_subprocess_exec",
fake_create_subprocess_exec,
)
monkeypatch.setattr(
local_booter.asyncio,
"create_subprocess_shell",
fail_create_subprocess_shell,
)

result = await LocalShellComponent().exec_managed(
"Get-ChildItem",
owner_id="owner-a",
creator_id="user-a",
creator_is_admin=False,
sandboxed=False,
cwd=str(tmp_path),
yield_time_ms=5_000,
windows_shell="pwsh.exe",
)

assert result["status"] == "completed"
assert result["stdout"] == "done\n"
assert calls[0][0] == (
"pwsh.exe",
"-NoLogo",
"-NoProfile",
"-NonInteractive",
"-Command",
"Get-ChildItem",
)
assert "creationflags" in calls[0][1]


def test_local_shell_component_prefers_utf8_before_windows_locale(
monkeypatch,
):
Expand Down Expand Up @@ -264,7 +396,7 @@ async def test_managed_shell_returns_completed_output_without_open_session():
)

assert result["status"] == "completed"
assert result["stdout"] == "hello\n"
assert result["stdout"].splitlines() == ["hello"]
assert result["exit_code"] == 0
assert result["session_closed"] is True
assert await shell.list_sessions(
Expand All @@ -288,7 +420,7 @@ async def test_managed_shell_allows_creator_and_conversation_admin():

try:
assert result["status"] == "running"
assert result["stdout"] == "ready\n"
assert result["stdout"].splitlines() == ["ready"]
session_id = result["session_id"]
assert (
await shell.list_sessions(
Expand Down Expand Up @@ -463,7 +595,7 @@ async def test_managed_shell_accepts_stdin_and_polls_incremental_output():
output += completed["stdout"]

assert completed["status"] == "completed"
assert output == "got:hello\n"
assert output.splitlines() == ["got:hello"]
assert completed["session_closed"] is True
finally:
await shell.shutdown_sessions()
Expand Down Expand Up @@ -525,7 +657,7 @@ async def test_managed_shell_keeps_completed_session_until_output_is_drained():
)
output += result["stdout"]

assert output == f"{'x' * 25000}\n"
assert output.splitlines() == ["x" * 25000]
assert result["session_closed"] is True
assert await shell.list_sessions(
owner_id="owner-a",
Expand Down
Loading