mirror of
https://github.com/AstrBotDevs/AstrBot.git
synced 2026-08-30 17:33:24 +08:00
feat: support PowerShell 7 for Windows local shell (#9622)
* feat(config): add Windows PowerShell version option * feat(computer): honor Windows PowerShell version in local runtime * fix(computer): correct stale shell comment for configurable PowerShell * refactor(computer): inline windows shell resolution per AGENTS.md * test(computer): assert windows_shell in exec_managed call * test: make shell tests cross-platform Co-authored-by: Donoym <prober13c14@gmail.com> * refactor(computer): auto-detect Windows shell instead of config * fix(computer): drop stale cmd.exe references after shell auto-detect * style: apply ruff format and import order * Update astrbot/core/astr_main_agent.py Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> --------- Co-authored-by: Donoym <prober13c14@gmail.com> Co-authored-by: Soulter <37870767+Soulter@users.noreply.github.com> Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com>
This commit is contained in:
@@ -29,6 +29,7 @@ from astrbot.core.astr_main_agent_resources import (
|
||||
TOOL_CALL_PROMPT,
|
||||
TOOL_CALL_PROMPT_SKILLS_LIKE_MODE,
|
||||
)
|
||||
from astrbot.core.computer.booters.local import resolve_windows_shell
|
||||
from astrbot.core.conversation_mgr import Conversation
|
||||
from astrbot.core.db import BaseDatabase
|
||||
from astrbot.core.message.components import File, Image, Record, Reply, Video
|
||||
@@ -428,7 +429,10 @@ 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,
|
||||
) -> None:
|
||||
if req.func_tool is None:
|
||||
req.func_tool = ToolSet()
|
||||
tool_mgr = plugin_context.get_llm_tool_manager()
|
||||
@@ -444,13 +448,22 @@ def _apply_local_env_tools(req: ProviderRequest, plugin_context: Context) -> Non
|
||||
|
||||
def _build_local_mode_prompt() -> 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 resolve_windows_shell() == "pwsh.exe":
|
||||
shell_hint = (
|
||||
"The runtime shell is PowerShell 7 (pwsh.exe). "
|
||||
"Use PowerShell 7-compatible syntax and cmdlets, and do not "
|
||||
"assume a full Unix userland or GNU utilities 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}. "
|
||||
|
||||
@@ -53,6 +53,11 @@ def _is_safe_command(command: str) -> bool:
|
||||
return not any(pat in cmd for pat in _BLOCKED_COMMAND_PATTERNS)
|
||||
|
||||
|
||||
def resolve_windows_shell() -> str:
|
||||
"""Prefer PowerShell 7 (pwsh.exe) when on PATH, else Windows PowerShell 5.1."""
|
||||
return "pwsh.exe" if shutil.which("pwsh") else "powershell.exe"
|
||||
|
||||
|
||||
def _decode_bytes_with_fallback(
|
||||
output: bytes | None,
|
||||
*,
|
||||
@@ -146,8 +151,9 @@ class LocalShellComponent(ShellComponent):
|
||||
popen_command: str | list[str] = command
|
||||
popen_shell = shell
|
||||
if sys.platform == "win32" and shell:
|
||||
shell_executable = resolve_windows_shell()
|
||||
popen_command = [
|
||||
"powershell.exe",
|
||||
shell_executable,
|
||||
"-NoLogo",
|
||||
"-NoProfile",
|
||||
"-NonInteractive",
|
||||
@@ -156,8 +162,9 @@ class LocalShellComponent(ShellComponent):
|
||||
]
|
||||
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 PowerShell 7 if available, else Windows
|
||||
# PowerShell 5.1, 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,
|
||||
@@ -167,8 +174,9 @@ class LocalShellComponent(ShellComponent):
|
||||
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 PowerShell 7 if available, else Windows
|
||||
# PowerShell 5.1, 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,
|
||||
@@ -278,8 +286,9 @@ class LocalShellComponent(ShellComponent):
|
||||
try:
|
||||
if sys.platform == "win32":
|
||||
process_factory = asyncio.create_subprocess_exec
|
||||
shell_executable = resolve_windows_shell()
|
||||
process_args = (
|
||||
"powershell.exe",
|
||||
shell_executable,
|
||||
"-NoLogo",
|
||||
"-NoProfile",
|
||||
"-NonInteractive",
|
||||
|
||||
@@ -67,7 +67,7 @@ class ExecuteShellTool(FunctionTool):
|
||||
"properties": {
|
||||
"command": {
|
||||
"type": "string",
|
||||
"description": "The shell command to execute in the current runtime shell (for example, cmd.exe on Windows). Equal to 'cd {working_dir} && {your_command}'.",
|
||||
"description": "The shell command to execute in the current runtime shell (for example, PowerShell on Windows). Equal to 'cd {working_dir} && {your_command}'.",
|
||||
},
|
||||
"background": {
|
||||
"type": "boolean",
|
||||
|
||||
@@ -60,6 +60,61 @@ def test_local_shell_component_uses_windows_powershell(monkeypatch):
|
||||
|
||||
monkeypatch.setattr(subprocess, "Popen", fake_run)
|
||||
monkeypatch.setattr(local_booter.sys, "platform", "win32")
|
||||
monkeypatch.setattr(local_booter.shutil, "which", lambda _cmd: None)
|
||||
|
||||
result = asyncio.run(LocalShellComponent().exec("Get-ChildItem"))
|
||||
|
||||
assert result["exit_code"] == 0
|
||||
assert calls[0][0][0] == [
|
||||
"powershell.exe",
|
||||
"-NoLogo",
|
||||
"-NoProfile",
|
||||
"-NonInteractive",
|
||||
"-Command",
|
||||
"Get-ChildItem",
|
||||
]
|
||||
assert calls[0][1]["shell"] is False
|
||||
|
||||
|
||||
def test_local_shell_component_prefers_pwsh_when_available(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")
|
||||
monkeypatch.setattr(
|
||||
local_booter.shutil,
|
||||
"which",
|
||||
lambda cmd: "/opt/pwsh" if cmd == "pwsh" else None,
|
||||
)
|
||||
|
||||
result = asyncio.run(LocalShellComponent().exec("Get-ChildItem"))
|
||||
|
||||
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_falls_back_to_powershell_when_pwsh_missing(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")
|
||||
monkeypatch.setattr(local_booter.shutil, "which", lambda _cmd: None)
|
||||
|
||||
result = asyncio.run(LocalShellComponent().exec("Get-ChildItem"))
|
||||
|
||||
@@ -122,6 +177,7 @@ async def test_managed_shell_uses_windows_powershell(monkeypatch, tmp_path):
|
||||
raise AssertionError("Windows managed commands must not use cmd.exe.")
|
||||
|
||||
monkeypatch.setattr(local_booter.sys, "platform", "win32")
|
||||
monkeypatch.setattr(local_booter.shutil, "which", lambda _cmd: None)
|
||||
monkeypatch.setattr(
|
||||
local_booter.asyncio,
|
||||
"create_subprocess_exec",
|
||||
@@ -156,6 +212,75 @@ async def test_managed_shell_uses_windows_powershell(monkeypatch, tmp_path):
|
||||
assert "creationflags" in calls[0][1]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_managed_shell_prefers_pwsh_when_available(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.shutil,
|
||||
"which",
|
||||
lambda cmd: "/opt/pwsh" if cmd == "pwsh" else None,
|
||||
)
|
||||
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,
|
||||
)
|
||||
|
||||
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,
|
||||
):
|
||||
@@ -242,6 +367,7 @@ def test_local_shell_component_falls_back_when_windows_taskkill_fails(monkeypatc
|
||||
lambda *_args, **_kwargs: _FakeTaskkillResult(returncode=1),
|
||||
)
|
||||
monkeypatch.setattr(local_booter.sys, "platform", "win32")
|
||||
monkeypatch.setattr(local_booter.shutil, "which", lambda _cmd: None)
|
||||
|
||||
with pytest.raises(subprocess.TimeoutExpired):
|
||||
asyncio.run(LocalShellComponent().exec("dummy", timeout=1))
|
||||
@@ -264,7 +390,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(
|
||||
@@ -288,7 +414,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(
|
||||
@@ -463,7 +589,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()
|
||||
@@ -525,7 +651,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",
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
import datetime
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
@@ -170,7 +171,10 @@ def test_append_system_reminders_includes_weekday(mock_event):
|
||||
|
||||
|
||||
def test_local_mode_prompt_uses_windows_powershell_51():
|
||||
with patch("astrbot.core.astr_main_agent.platform.system", return_value="Windows"):
|
||||
with patch("astrbot.core.astr_main_agent.platform.system", return_value="Windows"), patch(
|
||||
"astrbot.core.astr_main_agent.resolve_windows_shell",
|
||||
return_value="powershell.exe",
|
||||
):
|
||||
prompt = ama._build_local_mode_prompt()
|
||||
|
||||
assert "Windows PowerShell 5.1 (powershell.exe)" in prompt
|
||||
@@ -178,6 +182,30 @@ def test_local_mode_prompt_uses_windows_powershell_51():
|
||||
assert "cmd.exe" not in prompt
|
||||
|
||||
|
||||
def test_local_mode_prompt_hints_pwsh_when_resolved():
|
||||
with patch("astrbot.core.astr_main_agent.platform.system", return_value="Windows"), patch(
|
||||
"astrbot.core.astr_main_agent.resolve_windows_shell",
|
||||
return_value="pwsh.exe",
|
||||
):
|
||||
prompt = ama._build_local_mode_prompt()
|
||||
|
||||
assert "PowerShell 7 (pwsh.exe)" in prompt
|
||||
assert "Windows PowerShell 5.1" not in prompt
|
||||
assert "Unix-like" not in prompt
|
||||
|
||||
|
||||
def test_local_mode_prompt_ignores_pwsh_on_non_windows():
|
||||
with patch("astrbot.core.astr_main_agent.platform.system", return_value="Linux"), patch(
|
||||
"astrbot.core.astr_main_agent.resolve_windows_shell",
|
||||
return_value="pwsh.exe",
|
||||
):
|
||||
prompt = ama._build_local_mode_prompt()
|
||||
|
||||
assert "Unix-like" in prompt
|
||||
assert "POSIX-compatible" in prompt
|
||||
assert "PowerShell" not in prompt
|
||||
|
||||
|
||||
def test_local_mode_prompt_keeps_posix_shell_guidance():
|
||||
with patch("astrbot.core.astr_main_agent.platform.system", return_value="Linux"):
|
||||
prompt = ama._build_local_mode_prompt()
|
||||
@@ -1873,6 +1901,7 @@ class TestBuildMainAgent:
|
||||
):
|
||||
"""Test building main agent with video attachments."""
|
||||
module = ama
|
||||
video_path = str(Path("/path/to/video.mp4"))
|
||||
mock_video = Video(file="file:///path/to/video.mp4")
|
||||
mock_event.message_obj.message = [mock_video]
|
||||
|
||||
@@ -1900,7 +1929,7 @@ class TestBuildMainAgent:
|
||||
assert result is not None
|
||||
assert [
|
||||
part.text for part in result.provider_request.extra_user_content_parts
|
||||
] == ["[Video Attachment: name video.mp4, path /path/to/video.mp4]"]
|
||||
] == [f"[Video Attachment: name video.mp4, path {video_path}]"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_main_agent_with_quoted_video_attachment(
|
||||
@@ -1908,6 +1937,7 @@ class TestBuildMainAgent:
|
||||
):
|
||||
"""Test building main agent with quoted video attachments."""
|
||||
module = ama
|
||||
video_path = str(Path("/path/to/quoted-video.mp4"))
|
||||
mock_video = Video(file="file:///path/to/quoted-video.mp4")
|
||||
mock_reply = Reply(
|
||||
id="reply-1",
|
||||
@@ -1941,7 +1971,7 @@ class TestBuildMainAgent:
|
||||
assert result is not None
|
||||
assert (
|
||||
"[Video Attachment in quoted message: "
|
||||
"name quoted-video.mp4, path /path/to/quoted-video.mp4]"
|
||||
f"name quoted-video.mp4, path {video_path}]"
|
||||
) in [part.text for part in result.provider_request.extra_user_content_parts]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
Reference in New Issue
Block a user