diff --git a/astrbot/core/astr_main_agent.py b/astrbot/core/astr_main_agent.py index e3bf74f59..9a26c254b 100644 --- a/astrbot/core/astr_main_agent.py +++ b/astrbot/core/astr_main_agent.py @@ -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}. " diff --git a/astrbot/core/computer/booters/local.py b/astrbot/core/computer/booters/local.py index d2ab59f2e..9846a6102 100644 --- a/astrbot/core/computer/booters/local.py +++ b/astrbot/core/computer/booters/local.py @@ -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", diff --git a/astrbot/core/tools/computer_tools/shell.py b/astrbot/core/tools/computer_tools/shell.py index 61e1f589e..69d177769 100644 --- a/astrbot/core/tools/computer_tools/shell.py +++ b/astrbot/core/tools/computer_tools/shell.py @@ -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", diff --git a/tests/test_local_shell_component.py b/tests/test_local_shell_component.py index a81e620a6..e477d7df1 100644 --- a/tests/test_local_shell_component.py +++ b/tests/test_local_shell_component.py @@ -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", diff --git a/tests/unit/test_astr_main_agent.py b/tests/unit/test_astr_main_agent.py index 73aa4419d..ee2c56fc6 100644 --- a/tests/unit/test_astr_main_agent.py +++ b/tests/unit/test_astr_main_agent.py @@ -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