diff --git a/astrbot/core/astr_main_agent.py b/astrbot/core/astr_main_agent.py index b013f010e1..be80b89182 100644 --- a/astrbot/core/astr_main_agent.py +++ b/astrbot/core/astr_main_agent.py @@ -1022,6 +1022,8 @@ def _apply_sandbox_tools( req.system_prompt += ( "\n[CUA Desktop Control]\n" "Use `astrbot_execute_shell` with `background=true` to launch GUI apps. " + "Do not append `&`, `nohup`, or other shell background operators; " + "the tool handles detaching when `background=true`. " 'Use Firefox for browser tasks, for example `firefox "https://example.com"`. ' "After each visible step, call `astrbot_cua_screenshot` with " "`send_to_user=true` and `return_image_to_llm=true` so the user can " diff --git a/astrbot/core/tools/computer_tools/shell.py b/astrbot/core/tools/computer_tools/shell.py index cdefe97a0e..47703252a6 100644 --- a/astrbot/core/tools/computer_tools/shell.py +++ b/astrbot/core/tools/computer_tools/shell.py @@ -15,6 +15,7 @@ _COMPUTER_RUNTIME_TOOL_CONFIG = { "provider_settings.computer_use_runtime": ("local", "sandbox"), } +_EXPLICIT_BACKGROUND_LAUNCHERS = {"nohup", "setsid", "disown", "start", "start-process"} @builtin_tool(config=_COMPUTER_RUNTIME_TOOL_CONFIG) @@ -32,7 +33,7 @@ class ExecuteShellTool(FunctionTool): }, "background": { "type": "boolean", - "description": "Whether to run the command in the background.", + "description": "Whether to run the command in the background. Do not append shell background operators such as `&`; pass the foreground command and use this flag instead.", "default": False, }, "env": { @@ -70,9 +71,22 @@ async def call( cwd = str(current_workspace_root) env = dict(env or {}) - effective_background = background and not _is_self_detached_command(command) - result = await sb.shell.exec( + prepared_command, effective_background = _prepare_shell_background( command, + background, + ) + if background and not prepared_command.strip(): + return json.dumps( + { + "success": False, + "stdout": "", + "stderr": "error: empty shell command after removing background operator.", + "exit_code": 2, + }, + ensure_ascii=False, + ) + result = await sb.shell.exec( + prepared_command, cwd=cwd, background=effective_background, env=env, @@ -83,24 +97,45 @@ async def call( return f"Error executing command: {detail}" +def _prepare_shell_background(command: str, background: bool) -> tuple[str, bool]: + if not background: + return command, False + + tokens, has_explicit_launcher, has_trailing_amp = _classify_background(command) + if has_explicit_launcher: + return command, False + if has_trailing_amp: + return " ".join(tokens[:-1]), True + return command, True + + +def _classify_background(command: str) -> tuple[list[str], bool, bool]: + tokens = _command_tokens_before_comment(command) + if not tokens: + return tokens, False, False + + has_explicit_launcher = tokens[0].lower() in _EXPLICIT_BACKGROUND_LAUNCHERS + has_trailing_amp = tokens[-1] == "&" + return tokens, has_explicit_launcher, has_trailing_amp + + def _is_self_detached_command(command: str) -> bool: + _, has_explicit_launcher, has_trailing_amp = _classify_background(command) + return has_explicit_launcher or has_trailing_amp + + +def _command_tokens_before_comment(command: str) -> list[str]: lex = shlex.shlex(command, posix=False) lex.whitespace_split = True lex.commenters = "" try: tokens = list(lex) except ValueError: - return False + return [] comment_index = next( (index for index, token in enumerate(tokens) if token.startswith("#")), None, ) if comment_index is not None: tokens = tokens[:comment_index] - if not tokens: - return False - - first = tokens[0].lower() - if first in {"nohup", "setsid", "disown", "start", "start-process"}: - return True - return tokens[-1] == "&" + return tokens diff --git a/tests/unit/test_astr_main_agent.py b/tests/unit/test_astr_main_agent.py index faae767345..1ba6c59662 100644 --- a/tests/unit/test_astr_main_agent.py +++ b/tests/unit/test_astr_main_agent.py @@ -1582,6 +1582,7 @@ def test_apply_sandbox_tools_with_cua_adds_gui_guidance(self, mock_context): assert "Firefox" in req.system_prompt assert "background=true" in req.system_prompt + assert "Do not append `&`" in req.system_prompt assert 'firefox "https://example.com"' in req.system_prompt assert "astrbot_cua_screenshot" in req.system_prompt assert "astrbot_cua_key_press" not in req.system_prompt diff --git a/tests/unit/test_func_tool_manager.py b/tests/unit/test_func_tool_manager.py index 4eae43b5ce..0d98b61ff1 100644 --- a/tests/unit/test_func_tool_manager.py +++ b/tests/unit/test_func_tool_manager.py @@ -46,6 +46,10 @@ def test_computer_tools_are_registered_as_builtin_tools(): assert tool.name == "astrbot_execute_shell" assert tool.parameters["properties"]["background"]["default"] is False + assert ( + "Do not append shell background operators" + in tool.parameters["properties"]["background"]["description"] + ) assert manager.is_builtin_tool("astrbot_execute_shell") is True @@ -222,6 +226,137 @@ async def fake_get_booter(context, session_id): assert calls == [{"command": command, "background": False}] +@pytest.mark.asyncio +async def test_execute_shell_strips_plain_trailing_ampersand_when_background_flag_is_set( + monkeypatch, +): + from astrbot.core.tools.computer_tools import shell as shell_tools + + calls = [] + + class FakeShell: + async def exec(self, command, cwd=None, background=False, env=None): + calls.append({"command": command, "background": background}) + return {"success": True, "stdout": "", "stderr": "", "exit_code": 0} + + class FakeBooter: + shell = FakeShell() + + class FakeConfig: + def get_config(self, umo): + return {"provider_settings": {"computer_use_runtime": "sandbox"}} + + class FakeEvent: + unified_msg_origin = "umo" + role = "admin" + + class FakeAstrContext: + context = FakeConfig() + event = FakeEvent() + + class FakeWrapper: + context = FakeAstrContext() + + async def fake_get_booter(context, session_id): + return FakeBooter() + + monkeypatch.setattr(shell_tools, "get_booter", fake_get_booter) + + result = await ExecuteShellTool().call( + FakeWrapper(), command="firefox &", background=True + ) + + assert json.loads(result)["success"] is True + assert calls == [{"command": "firefox", "background": True}] + + +@pytest.mark.asyncio +async def test_execute_shell_rejects_bare_background_operator(monkeypatch): + from astrbot.core.tools.computer_tools import shell as shell_tools + + calls = [] + + class FakeShell: + async def exec(self, command, cwd=None, background=False, env=None): + calls.append({"command": command, "background": background}) + return {"success": True, "stdout": "", "stderr": "", "exit_code": 0} + + class FakeBooter: + shell = FakeShell() + + class FakeConfig: + def get_config(self, umo): + return {"provider_settings": {"computer_use_runtime": "sandbox"}} + + class FakeEvent: + unified_msg_origin = "umo" + role = "admin" + + class FakeAstrContext: + context = FakeConfig() + event = FakeEvent() + + class FakeWrapper: + context = FakeAstrContext() + + async def fake_get_booter(context, session_id): + return FakeBooter() + + monkeypatch.setattr(shell_tools, "get_booter", fake_get_booter) + + result = await ExecuteShellTool().call(FakeWrapper(), command="&", background=True) + payload = json.loads(result) + + assert payload["success"] is False + assert payload["exit_code"] == 2 + assert "empty shell command" in payload["stderr"] + assert calls == [] + + +@pytest.mark.asyncio +async def test_execute_shell_keeps_quoted_ampersand_when_background_flag_is_set( + monkeypatch, +): + from astrbot.core.tools.computer_tools import shell as shell_tools + + calls = [] + + class FakeShell: + async def exec(self, command, cwd=None, background=False, env=None): + calls.append({"command": command, "background": background}) + return {"success": True, "stdout": "", "stderr": "", "exit_code": 0} + + class FakeBooter: + shell = FakeShell() + + class FakeConfig: + def get_config(self, umo): + return {"provider_settings": {"computer_use_runtime": "sandbox"}} + + class FakeEvent: + unified_msg_origin = "umo" + role = "admin" + + class FakeAstrContext: + context = FakeConfig() + event = FakeEvent() + + class FakeWrapper: + context = FakeAstrContext() + + async def fake_get_booter(context, session_id): + return FakeBooter() + + monkeypatch.setattr(shell_tools, "get_booter", fake_get_booter) + + result = await ExecuteShellTool().call( + FakeWrapper(), command="echo '&'", background=True + ) + + assert json.loads(result)["success"] is True + assert calls == [{"command": "echo '&'", "background": True}] + + @pytest.mark.asyncio async def test_execute_shell_recognizes_commented_background_command(monkeypatch): from astrbot.core.tools.computer_tools import shell as shell_tools @@ -262,7 +397,7 @@ async def fake_get_booter(context, session_id): ) assert json.loads(result)["success"] is True - assert calls == [{"command": command, "background": False}] + assert calls == [{"command": "firefox", "background": True}] @pytest.mark.parametrize(