From 587e9ec229a5a206c8db64b6ef3d694fd0eeb521 Mon Sep 17 00:00:00 2001 From: ahmad-ajmal Date: Wed, 7 Oct 2026 11:54:13 +0100 Subject: [PATCH 1/5] fix #475: guide Windows shell execution and report quoting hazards Signed-off-by: ahmad-ajmal --- .github/workflows/windows-shell.yml | 26 ++ agent_core/core/prompts/context.py | 7 + app/data/action/run_shell.py | 272 +++++++++++------ app/data/agent_file_system_template/AGENT.md | 4 +- tests/windows_shell/test_run_shell_windows.py | 283 ++++++++++++++++++ 5 files changed, 502 insertions(+), 90 deletions(-) create mode 100644 .github/workflows/windows-shell.yml create mode 100644 tests/windows_shell/test_run_shell_windows.py diff --git a/.github/workflows/windows-shell.yml b/.github/workflows/windows-shell.yml new file mode 100644 index 00000000..b2690c52 --- /dev/null +++ b/.github/workflows/windows-shell.yml @@ -0,0 +1,26 @@ +name: Windows shell regressions + +on: + push: + paths: + - 'app/data/action/run_shell.py' + - 'agent_core/core/prompts/context.py' + - 'tests/windows_shell/**' + - '.github/workflows/windows-shell.yml' + pull_request: + paths: + - 'app/data/action/run_shell.py' + - 'agent_core/core/prompts/context.py' + - 'tests/windows_shell/**' + - '.github/workflows/windows-shell.yml' + +jobs: + shell: + runs-on: windows-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.10' + - name: Verify real cmd, Windows PowerShell and pwsh execution + run: python -m unittest discover -s tests/windows_shell -v diff --git a/agent_core/core/prompts/context.py b/agent_core/core/prompts/context.py index deacc08d..1da54626 100644 --- a/agent_core/core/prompts/context.py +++ b/agent_core/core/prompts/context.py @@ -171,6 +171,13 @@ - User Location: {user_location} - Current Working Directory: {working_directory} - Operating System: {operating_system} {os_version} ({os_platform}) +On Windows, run_shell defaults to cmd. For PowerShell source select shell="powershell" +or "pwsh" and submit the raw script; run_shell launches the interpreter. For example, +command="Get-CimInstance Win32_LogicalDisk -Filter 'DriveType=3'" with shell="powershell". +Omit powershell -Command and cmd /c wrappers unless nesting is intentional. Read +run_shell's guidance and verify stdout even on success; nested quoting can silently +change the result. After a quoting failure, check the selected shell and partial +side effects before correcting the source. Avoid repeated escape-character guesses. """ diff --git a/app/data/action/run_shell.py b/app/data/action/run_shell.py index 689b095f..f6eb671f 100644 --- a/app/data/action/run_shell.py +++ b/app/data/action/run_shell.py @@ -218,20 +218,28 @@ def shell_exec(input_data: dict) -> dict: @action( name="run_shell", - description="Executes a shell command using the appropriate OS shell, capturing stdout, stderr, and exit code. Stdin is closed (EOF) by default. IMPORTANT: For long-running commands that don't terminate (e.g., 'npm run dev', 'npm start', 'python -m http.server', 'flask run', watch processes, dev servers), you MUST set background=true. Otherwise, the command will block the entire task until timeout and may not capture any output.", + description=( + "Executes a command using the selected Windows shell, capturing stdout, " + "stderr and exit code. For PowerShell scripts set shell='powershell' " + "or 'pwsh' and pass raw source; the tool launches the interpreter, so " + "omit powershell -Command and cmd /c wrappers. Default/auto remains cmd. " + "Check guidance and verify stdout: nested quoting can return exit code 0 " + "with the wrong result. Stdin is closed (EOF). Set background=true for " + "dev servers, watchers and other commands that do not terminate." + ), platforms=["windows"], default=True, action_sets=["core"], input_schema={ "command": { "type": "string", - "example": "dir C:\\\\Windows\\\\System32", - "description": "The shell command to execute.", + "example": "Get-CimInstance Win32_LogicalDisk -Filter 'DriveType=3'", + "description": "Source for the selected shell. For shell='powershell'/'pwsh', send a raw script without an interpreter wrapper. JSON escaping is decoded once; do not add another escaping layer.", }, "shell": { "type": "string", - "example": "auto", - "description": "Shell to use. Windows: 'cmd' (default), 'powershell', or 'pwsh' — bash/zsh are NOT available, and an unsupported value returns an error. macOS: 'bash' (default) or 'zsh'. Linux: ignored (runs via the system shell).", + "example": "powershell", + "description": "Windows: 'cmd' (default/auto), 'powershell', or 'pwsh'. Choose powershell/pwsh for PowerShell source; the tool supplies -Command. Choose cmd for cmd syntax and batch workflows. bash/zsh/sh are unavailable.", }, "timeout": { "type": "integer", @@ -256,6 +264,21 @@ def shell_exec(input_data: dict) -> dict: }, }, output_schema={ + "shell": { + "type": "string", + "example": "powershell", + "description": "Normalized shell selected for this execution.", + }, + "shell_executable": { + "type": "string", + "example": "powershell.exe", + "description": "Resolved interpreter path, or executable name if unavailable. Null when no launch was prepared.", + }, + "guidance": { + "type": "string", + "example": "", + "description": "Advisory for nested interpreters or recognized parser failures. Commands are never rewritten or retried automatically.", + }, "status": {"type": "string", "example": "success"}, "stdout": {"type": "string", "example": "Command output text"}, "stderr": {"type": "string", "example": ""}, @@ -279,45 +302,98 @@ def shell_exec(input_data: dict) -> dict: ) def shell_exec_windows(input_data: dict) -> dict: import os + import re + import shutil import subprocess + shell_choice = str(input_data.get("shell", "cmd")).strip().lower() + if shell_choice in ("", "auto"): + shell_choice = "cmd" + shell_executable = None + command = str(input_data.get("command", "")).strip() + # Advisory only: intentional nested shells remain supported. Never rewrite + # or re-execute a command that might already have had side effects. + nested_shell = re.match( + r"^(?:cmd(?:\.exe)?\s+(?:/[dsk]\s+)*/c\s+)?" + r"(?:powershell|pwsh)(?:\.exe)?\s+.*?-(?:command|c)\b", + command, + re.IGNORECASE | re.DOTALL, + ) + script_guidance = ( + "For PowerShell source, set shell='powershell' (or 'pwsh') and pass " + "the raw script as command, without powershell -Command or cmd /c " + "wrappers. The tool launches the interpreter." + ) + + def result(output): + output["shell"] = shell_choice + output["shell_executable"] = shell_executable + guidance = [] + if nested_shell: + guidance.append( + "A nested PowerShell -Command wrapper was detected. Outer " + "shells can expand variables or interpret quotes before the " + "script reaches PowerShell, even with exit code 0. " + + script_guidance + + " If nesting is intentional, verify stdout matches the intended result." + ) + stderr = output.get("stderr", "").lower() + if output.get("return_code", 0) != 0 and any( + marker in stderr + for marker in ( + "parsererror", + "unexpectedtoken", + "terminatorexpectedatendofstring", + "missingclosing", + "was unexpected at this time", + ) + ): + guidance.append( + f"The command failed with a possible syntax/quoting error under {shell_choice}. " + + script_guidance + + " For cmd syntax, keep shell='cmd'. Check for partial side effects " + "before issuing a corrected command; this tool has not retried it." + ) + output["guidance"] = " ".join(guidance) + return output + simulated_mode = input_data.get("simulated_mode", False) if simulated_mode: # Return mock result for testing - return { - "status": "success", - "stdout": "Simulated command output", - "stderr": "", - "return_code": 0, - "message": "", - "pid": None, - } + return result( + { + "status": "success", + "stdout": "Simulated command output", + "stderr": "", + "return_code": 0, + "message": "", + "pid": None, + } + ) - command = str(input_data.get("command", "")).strip() - shell_choice = str(input_data.get("shell", "cmd")).strip().lower() - if shell_choice in ("", "auto"): - shell_choice = "cmd" if shell_choice not in ("cmd", "powershell", "pwsh"): # Previously any unsupported value (e.g. "bash", "sh", "zsh") was # silently coerced to cmd, so a bash heredoc would run under cmd and # fail with a cryptic "<< was unexpected at this time." Return an # explicit error instead so the caller knows its shell choice was # rejected and why. - return { - "status": "error", - "stdout": "", - "stderr": "", - "return_code": -1, - "message": ( - f"Shell '{shell_choice}' is not available on Windows. " - "Supported shells: cmd, powershell, pwsh. " - "bash/zsh/sh syntax (e.g. heredocs) will NOT run here — " - "use PowerShell for scripting, or write files via a file action " - "rather than shell redirection." - ), - "pid": None, - } + return result( + { + "status": "error", + "stdout": "", + "stderr": "", + "return_code": -1, + "message": ( + f"Shell '{shell_choice}' is not available on Windows. " + "Supported shells: cmd, powershell, pwsh. " + "bash/zsh/sh syntax (e.g. heredocs) will NOT run here — " + "use PowerShell for scripting, or write files via a file action " + "rather than shell redirection." + ), + "pid": None, + } + ) timeout_val = input_data.get("timeout") cwd = input_data.get("cwd") env_input = input_data.get("env") or {} @@ -326,24 +402,28 @@ def shell_exec_windows(input_data: dict) -> dict: timeout_seconds = float(timeout_val) if timeout_val is not None else 600.0 if not command: - return { - "status": "error", - "stdout": "", - "stderr": "", - "return_code": -1, - "message": "command is required.", - "pid": None, - } + return result( + { + "status": "error", + "stdout": "", + "stderr": "", + "return_code": -1, + "message": "command is required.", + "pid": None, + } + ) if cwd and not os.path.isdir(cwd): - return { - "status": "error", - "stdout": "", - "stderr": "", - "return_code": -1, - "message": "Working directory does not exist.", - "pid": None, - } + return result( + { + "status": "error", + "stdout": "", + "stderr": "", + "return_code": -1, + "message": "Working directory does not exist.", + "pid": None, + } + ) # Resolved Node runtime leads PATH: the agent is instructed to run the # agent-app CLI (TypeScript, needs node >= 24) via bare `node` through this @@ -354,9 +434,13 @@ def shell_exec_windows(input_data: dict) -> dict: for k, v in env_input.items(): env[str(k)] = str(v) + shell_executable = ( + shutil.which(shell_choice + ".exe", path=env.get("PATH")) + or shell_choice + ".exe" + ) if shell_choice == "powershell": args = [ - "powershell.exe", + shell_executable, "-NoLogo", "-NonInteractive", "-NoProfile", @@ -367,7 +451,7 @@ def shell_exec_windows(input_data: dict) -> dict: ] elif shell_choice == "pwsh": args = [ - "pwsh.exe", + shell_executable, "-NoLogo", "-NonInteractive", "-NoProfile", @@ -379,7 +463,7 @@ def shell_exec_windows(input_data: dict) -> dict: # escape embedded quotes as \" (MSVCRT rules), which cmd.exe does not # understand, mangling any command containing a quoted path. With # /s /c, cmd strips the outer quotes and runs the command verbatim. - args = 'cmd.exe /d /s /c "' + command + '"' + args = '"' + shell_executable + '" /d /s /c "' + command + '"' creation_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0) @@ -397,23 +481,27 @@ def shell_exec_windows(input_data: dict) -> dict: env=env, creationflags=bg_flags, ) - return { - "status": "background", - "stdout": "", - "stderr": "", - "return_code": 0, - "message": f"Process started in background with PID {process.pid}", - "pid": process.pid, - } + return result( + { + "status": "background", + "stdout": "", + "stderr": "", + "return_code": 0, + "message": f"Process started in background with PID {process.pid}", + "pid": process.pid, + } + ) except Exception as e: - return { - "status": "error", - "stdout": "", - "stderr": str(e), - "return_code": -1, - "message": str(e), - "pid": None, - } + return result( + { + "status": "error", + "stdout": "", + "stderr": str(e), + "return_code": -1, + "message": str(e), + "pid": None, + } + ) # Foreground mode with proper timeout handling try: @@ -443,14 +531,16 @@ def shell_exec_windows(input_data: dict) -> dict: try: stdout, stderr = process.communicate(timeout=timeout_seconds) - return { - "status": "success" if process.returncode == 0 else "error", - "stdout": stdout.strip() if stdout else "", - "stderr": stderr.strip() if stderr else "", - "return_code": process.returncode, - "message": "", - "pid": None, - } + return result( + { + "status": "success" if process.returncode == 0 else "error", + "stdout": stdout.strip() if stdout else "", + "stderr": stderr.strip() if stderr else "", + "return_code": process.returncode, + "message": "", + "pid": None, + } + ) except subprocess.TimeoutExpired: # Kill the entire process tree on Windows using taskkill try: @@ -463,25 +553,29 @@ def shell_exec_windows(input_data: dict) -> dict: pass process.kill() stdout, stderr = process.communicate() - return { + return result( + { + "status": "error", + "stdout": (stdout or "").strip(), + "stderr": (stderr or "").strip(), + "return_code": -1, + "message": f"Timed out after {timeout_seconds}s.", + "pid": None, + } + ) + finally: + unregister_process(run_session_id, process) + except Exception as e: + return result( + { "status": "error", - "stdout": (stdout or "").strip(), - "stderr": (stderr or "").strip(), + "stdout": "", + "stderr": str(e), "return_code": -1, - "message": f"Timed out after {timeout_seconds}s.", + "message": str(e), "pid": None, } - finally: - unregister_process(run_session_id, process) - except Exception as e: - return { - "status": "error", - "stdout": "", - "stderr": str(e), - "return_code": -1, - "message": str(e), - "pid": None, - } + ) @action( diff --git a/app/data/agent_file_system_template/AGENT.md b/app/data/agent_file_system_template/AGENT.md index 11674af6..67997209 100644 --- a/app/data/agent_file_system_template/AGENT.md +++ b/app/data/agent_file_system_template/AGENT.md @@ -1,5 +1,5 @@ --- -version: 9 +version: 10 purpose: agent operations manual --- @@ -520,6 +520,8 @@ There are four failure types. Identify which one you are in, then follow the mat **File / shell / Python action returns `status=error`** - Read the `message` field. It often points at the fix (file not found, permission, syntax error, missing dep). +- On Windows, read `shell`, `shell_executable`, and `guidance` from `run_shell`. For PowerShell source select `shell="powershell"` (or `"pwsh"`) and submit raw source: `command="Get-CimInstance Win32_LogicalDisk -Filter 'DriveType=3'"`. The tool launches the interpreter; omit `powershell -Command` and `cmd /c` wrappers unless nesting is intentional. Keep `shell="cmd"` for cmd syntax and batch workflows. +- For quoting/parser failures, correct the shell choice and source instead of guessing escapes or adding wrappers. Check partial side effects before retrying. Verify stdout even with exit code 0: nested shells can expand variables or turn source into a literal string. `guidance` is advisory; the tool never rewrites or retries a command automatically. - If the message says a missing dependency while running a script via `run_shell` (e.g. a Python `ModuleNotFoundError`), install it with `pip install`/`npm install` in a follow-up `run_shell` call. - If it says path not found, `find_files` or `list_folder` to locate before retry. diff --git a/tests/windows_shell/test_run_shell_windows.py b/tests/windows_shell/test_run_shell_windows.py new file mode 100644 index 00000000..b7546aed --- /dev/null +++ b/tests/windows_shell/test_run_shell_windows.py @@ -0,0 +1,283 @@ +"""Real Windows shell regressions, runnable with only the Python standard library. + +Run: python -m unittest discover -s tests/windows_shell -v +Like the action registry, load the function's source without its decorator. +Only unrelated Node PATH setup and cancellation registration are stubbed; +commands execute through the production launcher using real OS processes. +""" + +import ast +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import tempfile +import time +import types +import unittest +from unittest.mock import Mock, patch + + +ROOT = Path(__file__).resolve().parents[2] +SOURCE = ROOT / "app/data/action/run_shell.py" + + +def load_action(): + tree = ast.parse(SOURCE.read_text(encoding="utf-8")) + function = next( + node + for node in tree.body + if isinstance(node, ast.FunctionDef) and node.name == "shell_exec_windows" + ) + metadata = { + keyword.arg: ast.literal_eval(keyword.value) + for keyword in function.decorator_list[0].keywords + } + function.decorator_list = [] + namespace = {} + exec( + compile(ast.Module(body=[function], type_ignores=[]), str(SOURCE), "exec"), + namespace, + ) + return namespace["shell_exec_windows"], metadata + + +class ShellGuidanceTests(unittest.TestCase): + def setUp(self): + self.run_shell, self.metadata = load_action() + + def test_schema_teaches_raw_powershell(self): + self.assertIn("raw source", self.metadata["description"]) + self.assertEqual( + self.metadata["input_schema"]["shell"]["example"], "powershell" + ) + self.assertIn( + "Get-CimInstance", self.metadata["input_schema"]["command"]["example"] + ) + self.assertIn("guidance", self.metadata["output_schema"]) + + def test_invalid_shell_does_not_launch(self): + with patch("subprocess.Popen") as spawn: + result = self.run_shell({"command": "echo x", "shell": "bash"}) + spawn.assert_not_called() + self.assertEqual(result["status"], "error") + self.assertEqual(result["shell"], "bash") + self.assertIsNone(result["shell_executable"]) + + def test_empty_command_reports_default_without_launch(self): + result = self.run_shell({"command": "", "shell": "auto"}) + self.assertEqual(result["shell"], "cmd") + self.assertIsNone(result["shell_executable"]) + self.assertEqual(result["message"], "command is required.") + + def test_environment_prompt_can_still_be_formatted(self): + tree = ast.parse( + (ROOT / "agent_core/core/prompts/context.py").read_text("utf-8") + ) + assignment = next( + node + for node in tree.body + if isinstance(node, ast.Assign) + and any( + isinstance(target, ast.Name) + and target.id == "ENVIRONMENTAL_CONTEXT_PROMPT" + for target in node.targets + ) + ) + prompt = ast.literal_eval(assignment.value).format( + user_location="UTC", + working_directory=".", + operating_system="Windows", + os_version="11", + os_platform="win32", + ) + self.assertIn('shell="powershell"', prompt) + + +@unittest.skipUnless(sys.platform == "win32", "Requires real Windows interpreters") +class WindowsShellTests(unittest.TestCase): + def setUp(self): + self.run_shell, self.metadata = load_action() + app = types.ModuleType("app") + app.node_runtime = types.SimpleNamespace(child_env=lambda: os.environ.copy()) + self.cancellation = types.ModuleType("agent_core.core.impl.action.cancellation") + self.cancellation.register_process = Mock() + self.cancellation.unregister_process = Mock() + modules = patch.dict( + sys.modules, + {"app": app, self.cancellation.__name__: self.cancellation}, + ) + modules.start() + self.addCleanup(modules.stop) + + def execute(self, command, shell="powershell", **kwargs): + if not shutil.which(shell + ".exe"): + self.skipTest(shell + " is unavailable") + return self.run_shell( + {"command": command, "shell": shell, "timeout": 15, **kwargs} + ) + + def assert_output(self, result, expected): + self.assertEqual(result["status"], "success", result) + self.assertEqual(result["return_code"], 0, result) + self.assertEqual(result["stdout"].replace("\r\n", "\n"), expected) + + def test_direct_scripts_preserve_semantics(self): + cases = [ + ("Write-Output 'hello world'", "hello world"), + ('Write-Output "hello world"', "hello world"), + ("$n=2; 1..$n | ForEach-Object { $_ * 2 }", "2\n4"), + ("$n=2\nWrite-Output $n", "2"), + ( + '[pscustomobject]@{"Free (GB)"=2} | ConvertTo-Json -Compress', + '{"Free (GB)":2}', + ), + ("Write-Output 'a\\\"b'", 'a\\"b'), + ( + 'Write-Output "a&b|c(d)%CRAFTBOT_QUOTE_REPRO%"', + "a&b|c(d)%CRAFTBOT_QUOTE_REPRO%", + ), + ] + for shell in ("powershell", "pwsh"): + if not shutil.which(shell + ".exe"): + continue + for command, expected in cases: + with self.subTest(shell=shell, command=command): + result = self.execute(command, shell) + self.assert_output(result, expected) + self.assertEqual(result["guidance"], "") + self.assertEqual(result["shell"], shell) + self.assertEqual( + result["shell_executable"], shutil.which(shell + ".exe") + ) + + def test_disk_query_with_parenthesized_calculated_property(self): + result = self.execute( + "@(Get-CimInstance Win32_LogicalDisk -Filter 'DriveType=3' | " + 'Select-Object @{Name="Free (GB)";Expression={[math]::Round($_.FreeSpace/1GB,2)}}) ' + "| ConvertTo-Json -Compress" + ) + self.assertEqual(result["status"], "success", result) + disks = json.loads(result["stdout"]) + self.assertGreater(len(disks), 0) + self.assertTrue( + all(isinstance(disk["Free (GB)"], (int, float)) for disk in disks) + ) + + def test_nested_variables_fail_with_actionable_guidance(self): + result = self.execute('powershell -NoProfile -Command "$n=2; Write-Output $n"') + self.assertEqual(result["status"], "error", result) + self.assertIn("=2", result["stderr"]) + self.assertIn("raw script", result["guidance"]) + self.assertEqual(self.cancellation.register_process.call_count, 1) + self.assertEqual(self.cancellation.unregister_process.call_count, 1) + self.assert_output(self.execute("$n=2; Write-Output $n"), "2") + + def test_single_quote_wrapper_silent_wrong_output_is_flagged(self): + result = self.execute( + "powershell -NoProfile -Command 'Write-Output \"hello world\"'", "cmd" + ) + self.assert_output(result, "Write-Output hello world") + self.assertIn("even with exit code 0", result["guidance"]) + self.assert_output(self.execute("Write-Output 'hello world'"), "hello world") + + def test_cmd_percent_expansion_is_flagged_and_raw_script_preserves_literal(self): + env = {"CRAFTBOT_QUOTE_REPRO": "expanded-by-cmd"} + result = self.execute( + "powershell -NoProfile -Command \"Write-Output '%CRAFTBOT_QUOTE_REPRO%'\"", + "cmd", + env=env, + ) + self.assert_output(result, "expanded-by-cmd") + self.assertIn("expand variables", result["guidance"]) + self.assert_output( + self.execute("Write-Output '%CRAFTBOT_QUOTE_REPRO%'", env=env), + "%CRAFTBOT_QUOTE_REPRO%", + ) + + def test_intentional_nested_wrapper_still_executes_once(self): + result = self.execute( + "cmd /c powershell -NoProfile -Command \"Write-Output 'ok'\"", "cmd" + ) + self.assert_output(result, "ok") + self.assertIn("intentional", result["guidance"]) + self.assertEqual(self.cancellation.register_process.call_count, 1) + + def test_parser_failure_has_shell_guidance(self): + result = self.execute("$x = (") + self.assertEqual(result["status"], "error", result) + self.assertIn("syntax/quoting error under powershell", result["guidance"]) + self.assertIn("has not retried", result["guidance"]) + self.assertEqual(self.cancellation.register_process.call_count, 1) + + def test_ordinary_nonzero_exit_is_preserved(self): + result = self.execute("[Console]::Error.WriteLine('sentinel'); exit 7") + self.assertEqual(result["return_code"], 7) + self.assertEqual(result["stderr"], "sentinel") + self.assertEqual(result["guidance"], "") + + def test_cmd_default_auto_and_explicit_remain_compatible(self): + for choice in (None, "auto", "cmd"): + with self.subTest(shell=choice): + payload = {"command": "echo cmd-compatible", "timeout": 15} + if choice: + payload["shell"] = choice + result = self.run_shell(payload) + self.assert_output(result, "cmd-compatible") + self.assertEqual(result["shell"], "cmd") + self.assertEqual(result["guidance"], "") + + def test_cmd_quoted_executable_path(self): + result = self.execute(f'"{sys.executable}" -c "print(123)"', "cmd") + self.assert_output(result, "123") + + def test_mentions_inside_script_do_not_trigger_wrapper_advice(self): + result = self.execute("Write-Output 'powershell -Command example'") + self.assert_output(result, "powershell -Command example") + self.assertEqual(result["guidance"], "") + + def test_timeout_unregisters_process(self): + result = self.execute("Start-Sleep -Seconds 10", timeout=1) + self.assertEqual(result["status"], "error") + self.assertIn("Timed out", result["message"]) + process = self.cancellation.register_process.call_args.args[1] + self.assertIsNotNone(process.poll()) + self.cancellation.unregister_process.assert_called_once_with("", process) + + def test_background_returns_pid_and_completes(self): + with tempfile.TemporaryDirectory() as directory: + marker = Path(directory) / "done.txt" + escaped = str(marker).replace("'", "''") + processes = [] + real_popen = subprocess.Popen + + def track_process(*args, **kwargs): + process = real_popen(*args, **kwargs) + processes.append(process) + return process + + with patch("subprocess.Popen", side_effect=track_process): + result = self.execute( + f"Start-Sleep -Milliseconds 300; [IO.File]::WriteAllText('{escaped}', 'done')", + background=True, + ) + self.assertEqual(result["status"], "background", result) + self.assertGreater(result["pid"], 0) + try: + deadline = time.monotonic() + 15 + while not marker.exists() and time.monotonic() < deadline: + time.sleep(0.05) + self.assertEqual(marker.read_text(), "done") + finally: + subprocess.run( + ["taskkill", "/F", "/T", "/PID", str(result["pid"])], + capture_output=True, + creationflags=subprocess.CREATE_NO_WINDOW, + ) + processes[0].wait(timeout=5) + + +if __name__ == "__main__": + unittest.main() From c27d14209c41306162ec30204f0349e3ef4c577a Mon Sep 17 00:00:00 2001 From: ahmad-ajmal Date: Thu, 8 Oct 2026 10:58:31 +0100 Subject: [PATCH 2/5] fix #475: scope shell guidance and preserve native errors Signed-off-by: ahmad-ajmal --- .github/workflows/windows-shell.yml | 2 - agent_core/core/prompts/context.py | 7 --- app/data/action/run_shell.py | 56 +++++-------------- app/data/agent_file_system_template/AGENT.md | 4 +- tests/windows_shell/test_run_shell_windows.py | 49 +++++++--------- 5 files changed, 34 insertions(+), 84 deletions(-) diff --git a/.github/workflows/windows-shell.yml b/.github/workflows/windows-shell.yml index b2690c52..2c0add72 100644 --- a/.github/workflows/windows-shell.yml +++ b/.github/workflows/windows-shell.yml @@ -4,13 +4,11 @@ on: push: paths: - 'app/data/action/run_shell.py' - - 'agent_core/core/prompts/context.py' - 'tests/windows_shell/**' - '.github/workflows/windows-shell.yml' pull_request: paths: - 'app/data/action/run_shell.py' - - 'agent_core/core/prompts/context.py' - 'tests/windows_shell/**' - '.github/workflows/windows-shell.yml' diff --git a/agent_core/core/prompts/context.py b/agent_core/core/prompts/context.py index 1da54626..deacc08d 100644 --- a/agent_core/core/prompts/context.py +++ b/agent_core/core/prompts/context.py @@ -171,13 +171,6 @@ - User Location: {user_location} - Current Working Directory: {working_directory} - Operating System: {operating_system} {os_version} ({os_platform}) -On Windows, run_shell defaults to cmd. For PowerShell source select shell="powershell" -or "pwsh" and submit the raw script; run_shell launches the interpreter. For example, -command="Get-CimInstance Win32_LogicalDisk -Filter 'DriveType=3'" with shell="powershell". -Omit powershell -Command and cmd /c wrappers unless nesting is intentional. Read -run_shell's guidance and verify stdout even on success; nested quoting can silently -change the result. After a quoting failure, check the selected shell and partial -side effects before correcting the source. Avoid repeated escape-character guesses. """ diff --git a/app/data/action/run_shell.py b/app/data/action/run_shell.py index f6eb671f..59a9448f 100644 --- a/app/data/action/run_shell.py +++ b/app/data/action/run_shell.py @@ -219,13 +219,8 @@ def shell_exec(input_data: dict) -> dict: @action( name="run_shell", description=( - "Executes a command using the selected Windows shell, capturing stdout, " - "stderr and exit code. For PowerShell scripts set shell='powershell' " - "or 'pwsh' and pass raw source; the tool launches the interpreter, so " - "omit powershell -Command and cmd /c wrappers. Default/auto remains cmd. " - "Check guidance and verify stdout: nested quoting can return exit code 0 " - "with the wrong result. Stdin is closed (EOF). Set background=true for " - "dev servers, watchers and other commands that do not terminate." + "Executes Windows shell source, capturing stdout, stderr and exit code. " + "Stdin is closed. Set background=true for servers/watchers." ), platforms=["windows"], default=True, @@ -234,12 +229,12 @@ def shell_exec(input_data: dict) -> dict: "command": { "type": "string", "example": "Get-CimInstance Win32_LogicalDisk -Filter 'DriveType=3'", - "description": "Source for the selected shell. For shell='powershell'/'pwsh', send a raw script without an interpreter wrapper. JSON escaping is decoded once; do not add another escaping layer.", + "description": "Raw source for the selected shell; the tool starts the interpreter. Omit powershell -Command/cmd /c wrappers unless intentional.", }, "shell": { "type": "string", "example": "powershell", - "description": "Windows: 'cmd' (default/auto), 'powershell', or 'pwsh'. Choose powershell/pwsh for PowerShell source; the tool supplies -Command. Choose cmd for cmd syntax and batch workflows. bash/zsh/sh are unavailable.", + "description": "cmd (default/auto), powershell, or pwsh. Use powershell/pwsh for PowerShell source.", }, "timeout": { "type": "integer", @@ -277,7 +272,7 @@ def shell_exec(input_data: dict) -> dict: "guidance": { "type": "string", "example": "", - "description": "Advisory for nested interpreters or recognized parser failures. Commands are never rewritten or retried automatically.", + "description": "Advisory for nested PowerShell wrappers. Commands are never rewritten or retried automatically.", }, "status": {"type": "string", "example": "success"}, "stdout": {"type": "string", "example": "Command output text"}, @@ -319,42 +314,19 @@ def shell_exec_windows(input_data: dict) -> dict: command, re.IGNORECASE | re.DOTALL, ) - script_guidance = ( - "For PowerShell source, set shell='powershell' (or 'pwsh') and pass " - "the raw script as command, without powershell -Command or cmd /c " - "wrappers. The tool launches the interpreter." - ) def result(output): output["shell"] = shell_choice output["shell_executable"] = shell_executable - guidance = [] - if nested_shell: - guidance.append( - "A nested PowerShell -Command wrapper was detected. Outer " - "shells can expand variables or interpret quotes before the " - "script reaches PowerShell, even with exit code 0. " - + script_guidance - + " If nesting is intentional, verify stdout matches the intended result." - ) - stderr = output.get("stderr", "").lower() - if output.get("return_code", 0) != 0 and any( - marker in stderr - for marker in ( - "parsererror", - "unexpectedtoken", - "terminatorexpectedatendofstring", - "missingclosing", - "was unexpected at this time", - ) - ): - guidance.append( - f"The command failed with a possible syntax/quoting error under {shell_choice}. " - + script_guidance - + " For cmd syntax, keep shell='cmd'. Check for partial side effects " - "before issuing a corrected command; this tool has not retried it." - ) - output["guidance"] = " ".join(guidance) + # Preserve native stderr and exit status; stderr text is not a reliable + # error-type protocol across shells, applications or localized output. + output["guidance"] = ( + "Nested PowerShell -Command detected: outer shells may expand variables " + "or quotes, even with exit code 0. Prefer shell='powershell'/'pwsh' with " + "raw script; verify stdout if nesting is intentional." + if nested_shell + else "" + ) return output simulated_mode = input_data.get("simulated_mode", False) diff --git a/app/data/agent_file_system_template/AGENT.md b/app/data/agent_file_system_template/AGENT.md index 67997209..11674af6 100644 --- a/app/data/agent_file_system_template/AGENT.md +++ b/app/data/agent_file_system_template/AGENT.md @@ -1,5 +1,5 @@ --- -version: 10 +version: 9 purpose: agent operations manual --- @@ -520,8 +520,6 @@ There are four failure types. Identify which one you are in, then follow the mat **File / shell / Python action returns `status=error`** - Read the `message` field. It often points at the fix (file not found, permission, syntax error, missing dep). -- On Windows, read `shell`, `shell_executable`, and `guidance` from `run_shell`. For PowerShell source select `shell="powershell"` (or `"pwsh"`) and submit raw source: `command="Get-CimInstance Win32_LogicalDisk -Filter 'DriveType=3'"`. The tool launches the interpreter; omit `powershell -Command` and `cmd /c` wrappers unless nesting is intentional. Keep `shell="cmd"` for cmd syntax and batch workflows. -- For quoting/parser failures, correct the shell choice and source instead of guessing escapes or adding wrappers. Check partial side effects before retrying. Verify stdout even with exit code 0: nested shells can expand variables or turn source into a literal string. `guidance` is advisory; the tool never rewrites or retries a command automatically. - If the message says a missing dependency while running a script via `run_shell` (e.g. a Python `ModuleNotFoundError`), install it with `pip install`/`npm install` in a follow-up `run_shell` call. - If it says path not found, `find_files` or `list_folder` to locate before retry. diff --git a/tests/windows_shell/test_run_shell_windows.py b/tests/windows_shell/test_run_shell_windows.py index b7546aed..c1d2109a 100644 --- a/tests/windows_shell/test_run_shell_windows.py +++ b/tests/windows_shell/test_run_shell_windows.py @@ -49,7 +49,9 @@ def setUp(self): self.run_shell, self.metadata = load_action() def test_schema_teaches_raw_powershell(self): - self.assertIn("raw source", self.metadata["description"]) + self.assertIn( + "Raw source", self.metadata["input_schema"]["command"]["description"] + ) self.assertEqual( self.metadata["input_schema"]["shell"]["example"], "powershell" ) @@ -72,29 +74,6 @@ def test_empty_command_reports_default_without_launch(self): self.assertIsNone(result["shell_executable"]) self.assertEqual(result["message"], "command is required.") - def test_environment_prompt_can_still_be_formatted(self): - tree = ast.parse( - (ROOT / "agent_core/core/prompts/context.py").read_text("utf-8") - ) - assignment = next( - node - for node in tree.body - if isinstance(node, ast.Assign) - and any( - isinstance(target, ast.Name) - and target.id == "ENVIRONMENTAL_CONTEXT_PROMPT" - for target in node.targets - ) - ) - prompt = ast.literal_eval(assignment.value).format( - user_location="UTC", - working_directory=".", - operating_system="Windows", - os_version="11", - os_platform="win32", - ) - self.assertIn('shell="powershell"', prompt) - @unittest.skipUnless(sys.platform == "win32", "Requires real Windows interpreters") class WindowsShellTests(unittest.TestCase): @@ -205,17 +184,27 @@ def test_intentional_nested_wrapper_still_executes_once(self): self.assertIn("intentional", result["guidance"]) self.assertEqual(self.cancellation.register_process.call_count, 1) - def test_parser_failure_has_shell_guidance(self): + def test_parser_failure_preserves_native_error(self): result = self.execute("$x = (") self.assertEqual(result["status"], "error", result) - self.assertIn("syntax/quoting error under powershell", result["guidance"]) - self.assertIn("has not retried", result["guidance"]) + self.assertNotEqual(result["return_code"], 0) + self.assertTrue(result["stderr"]) + self.assertEqual(result["guidance"], "") self.assertEqual(self.cancellation.register_process.call_count, 1) - def test_ordinary_nonzero_exit_is_preserved(self): - result = self.execute("[Console]::Error.WriteLine('sentinel'); exit 7") + def test_application_error_text_is_not_classified_as_a_parser_failure(self): + stderr = "UnexpectedToken: application-defined failure" + result = self.execute(f"[Console]::Error.WriteLine('{stderr}'); exit 7") self.assertEqual(result["return_code"], 7) - self.assertEqual(result["stderr"], "sentinel") + self.assertEqual(result["stderr"], stderr) + self.assertEqual(result["guidance"], "") + + def test_successful_stderr_does_not_imply_failure(self): + result = self.execute( + "[Console]::Error.WriteLine('warning'); Write-Output 'ok'" + ) + self.assert_output(result, "ok") + self.assertEqual(result["stderr"], "warning") self.assertEqual(result["guidance"], "") def test_cmd_default_auto_and_explicit_remain_compatible(self): From fb5df8574bb2f10226127d3fe3c024ffe9bc2d4b Mon Sep 17 00:00:00 2001 From: CraftBot Date: Tue, 13 Oct 2026 15:57:51 +0900 Subject: [PATCH 3/5] added run powershell action for windows only to prevent run shell encoding issue --- .github/workflows/windows-shell.yml | 4 +- agent_core/core/action_framework/registry.py | 9 +- agent_core/core/impl/action/cancellation.py | 17 + app/agent_app/construction_events.py | 3 + app/data/action/run_powershell.py | 237 ++++++++++++++ app/data/action/run_shell.py | 288 ++++++----------- tests/test_shell_actions.py | 116 +++++++ tests/windows_shell/test_run_shell_windows.py | 297 ++++++++++-------- 8 files changed, 654 insertions(+), 317 deletions(-) create mode 100644 app/data/action/run_powershell.py create mode 100644 tests/test_shell_actions.py diff --git a/.github/workflows/windows-shell.yml b/.github/workflows/windows-shell.yml index 2c0add72..86f8ee7c 100644 --- a/.github/workflows/windows-shell.yml +++ b/.github/workflows/windows-shell.yml @@ -4,11 +4,13 @@ on: push: paths: - 'app/data/action/run_shell.py' + - 'app/data/action/run_powershell.py' - 'tests/windows_shell/**' - '.github/workflows/windows-shell.yml' pull_request: paths: - 'app/data/action/run_shell.py' + - 'app/data/action/run_powershell.py' - 'tests/windows_shell/**' - '.github/workflows/windows-shell.yml' @@ -20,5 +22,5 @@ jobs: - uses: actions/setup-python@v5 with: python-version: '3.10' - - name: Verify real cmd, Windows PowerShell and pwsh execution + - name: Verify real cmd and Windows PowerShell execution run: python -m unittest discover -s tests/windows_shell -v diff --git a/agent_core/core/action_framework/registry.py b/agent_core/core/action_framework/registry.py index a8e95aa6..71dec805 100644 --- a/agent_core/core/action_framework/registry.py +++ b/agent_core/core/action_framework/registry.py @@ -220,8 +220,13 @@ def list_all_actions_as_json(self) -> List[Dict[str, Any]]: return json_actions_list def find_action_by_name(self, action_name: str) -> Optional[Dict[str, Any]]: - """Find an action by name and return its JSON representation.""" - if action_name not in self._registry: + """Find an action by name and return its JSON representation. + + An action with no implementation for this OS does not exist here. + Action sets already leave it out of the prompt; this keeps a name the + model produces anyway from resolving to another OS's code. + """ + if self.get_action_implementation(action_name) is None: return None platform_impls = self._registry[action_name] diff --git a/agent_core/core/impl/action/cancellation.py b/agent_core/core/impl/action/cancellation.py index 27b5918f..7461f6ba 100644 --- a/agent_core/core/impl/action/cancellation.py +++ b/agent_core/core/impl/action/cancellation.py @@ -27,6 +27,9 @@ trees included, and is safe to call at any time (missing/exited processes are ignored). It is blocking (taskkill / killpg) — call it from a worker thread, not the event loop. + +``was_killed(proc)`` lets an action that holds a registered handle tell a +stop apart from its own command failing: both end in a non-zero exit code. """ from __future__ import annotations @@ -36,6 +39,7 @@ import subprocess import tempfile import threading +import weakref from pathlib import Path from typing import Dict, Optional @@ -44,6 +48,9 @@ _lock = threading.Lock() # session_id -> {pid: Popen}. Popen handles registered by in-process actions. _procs: Dict[str, Dict[int, subprocess.Popen]] = {} +# Handles a stop killed. Weak: an entry lasts only as long as its action +# still holds the handle, so nothing here needs clearing. +_killed: "weakref.WeakSet[subprocess.Popen]" = weakref.WeakSet() def _marker_dir(session_id: str) -> Path: @@ -73,6 +80,12 @@ def unregister_process(session_id: str, proc: subprocess.Popen) -> None: _procs.pop(session_id, None) +def was_killed(proc: subprocess.Popen) -> bool: + """Whether a user stop killed this process, so its exit is not a failure.""" + with _lock: + return proc in _killed + + # ─────────────────────── Cross-process pid markers ─────────────────────── # psutil's create_time can wobble by a fraction of a second between reads. @@ -178,6 +191,10 @@ def kill_session_processes(session_id: str) -> int: killed = 0 for proc in handles: if proc.poll() is None: + # Marked BEFORE the kill: the action's communicate() returns the + # instant the process dies, and must already see the mark. + with _lock: + _killed.add(proc) _kill_tree(proc.pid) killed += 1 try: diff --git a/app/agent_app/construction_events.py b/app/agent_app/construction_events.py index 55dbf6ee..a1345c04 100644 --- a/app/agent_app/construction_events.py +++ b/app/agent_app/construction_events.py @@ -47,6 +47,7 @@ "agent_app_notify_ready", "agent_app_ops_verify", "run_shell", + "run_powershell", "spawn_subagent", } ) @@ -543,6 +544,8 @@ def _classify_activity( ) elif name == "run_shell": kind, area, label = "run", "other", f"Ran {_clip(inputs.get('command'), 40)}" + elif name == "run_powershell": + kind, area, label = "run", "other", f"Ran {_clip(inputs.get('script'), 40)}" elif name == "spawn_subagent": agent_type = str(inputs.get("agent_type", "")) if agent_type == "walk_verify": diff --git a/app/data/action/run_powershell.py b/app/data/action/run_powershell.py new file mode 100644 index 00000000..d0167116 --- /dev/null +++ b/app/data/action/run_powershell.py @@ -0,0 +1,237 @@ +from agent_core import action + + +@action( + name="run_powershell", + description="Runs a PowerShell script with Windows PowerShell, capturing stdout, stderr, and exit code. Stdin is closed (EOF) by default. IMPORTANT: For long-running scripts that don't terminate (e.g., dev servers, watch processes), you MUST set background=true. Otherwise, the script will block the entire task until timeout and may not capture any output.", + platforms=["windows"], + default=True, + action_sets=["core"], + input_schema={ + "script": { + "type": "string", + "example": "Get-CimInstance Win32_LogicalDisk -Filter 'DriveType=3' | Select-Object DeviceID, FreeSpace, Size", + "description": "The PowerShell script to run, exactly as typed at a PowerShell prompt. Can span multiple lines.", + }, + "timeout": { + "type": "integer", + "example": 600, + "description": "Optional timeout (seconds). If exceeded, the process is terminated.", + }, + "cwd": { + "type": "string", + "example": "C:\\\\Users\\\\me", + "description": "Optional working directory for the script.", + }, + "env": { + "type": "object", + "additionalProperties": {"type": "string"}, + "example": {"MY_VAR": "123"}, + "description": "Optional environment variable overrides.", + }, + "background": { + "type": "boolean", + "example": False, + "description": "Set to true for long-running processes (dev servers, watchers, etc.). The script will start in the background and return immediately with the process ID. Required for commands like 'npm run dev', 'npm start', 'python -m http.server'.", + }, + }, + output_schema={ + "status": {"type": "string", "example": "success"}, + "stdout": {"type": "string", "example": "Script output text"}, + "stderr": {"type": "string", "example": ""}, + "return_code": {"type": "integer", "example": 0}, + "message": {"type": "string", "example": "Timed out after 30s."}, + "pid": { + "type": "integer", + "example": 12345, + "description": "Process ID when running in background mode.", + }, + }, + test_payload={ + "script": "Get-ChildItem C:\\\\Windows\\\\System32", + "timeout": 600, + "cwd": "C:\\\\Users\\\\me", + "env": {"MY_VAR": "123"}, + "background": False, + "simulated_mode": True, + }, +) +def powershell_exec_windows(input_data: dict) -> dict: + import os + import subprocess + + simulated_mode = input_data.get("simulated_mode", False) + + if simulated_mode: + # Return mock result for testing + return { + "status": "success", + "stdout": "Simulated script output", + "stderr": "", + "return_code": 0, + "message": "", + "pid": None, + } + + script = str(input_data.get("script", "")).strip() + timeout_val = input_data.get("timeout") + cwd = input_data.get("cwd") + env_input = input_data.get("env") or {} + background = input_data.get("background", False) + + timeout_seconds = float(timeout_val) if timeout_val is not None else 600.0 + + if not script: + return { + "status": "error", + "stdout": "", + "stderr": "", + "return_code": -1, + "message": "script is required.", + "pid": None, + } + + if cwd and not os.path.isdir(cwd): + return { + "status": "error", + "stdout": "", + "stderr": "", + "return_code": -1, + "message": "Working directory does not exist.", + "pid": None, + } + + # Same environment as run_shell: the resolved Node runtime leads PATH, so + # bare `node` is the major the agent-app CLI needs (app/node_runtime.py). + from app import node_runtime + + env = node_runtime.child_env() + for k, v in env_input.items(): + env[str(k)] = str(v) + + # Windows PowerShell ships with every Windows install, so the dialect a + # script is written for is the same on every machine. + args = [ + "powershell.exe", + "-NoLogo", + "-NonInteractive", + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-Command", + script, + ] + + creation_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0) + + # Background mode: start process and return immediately + if background: + try: + # Use CREATE_NEW_PROCESS_GROUP to detach from parent + bg_flags = creation_flags | subprocess.CREATE_NEW_PROCESS_GROUP + process = subprocess.Popen( + args, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + stdin=subprocess.DEVNULL, + cwd=cwd if cwd else None, + env=env, + creationflags=bg_flags, + ) + return { + "status": "background", + "stdout": "", + "stderr": "", + "return_code": 0, + "message": f"Process started in background with PID {process.pid}", + "pid": process.pid, + } + except Exception as e: + return { + "status": "error", + "stdout": "", + "stderr": str(e), + "return_code": -1, + "message": str(e), + "pid": None, + } + + # Foreground mode with proper timeout handling + try: + # Register with the run-cancel registry so a user force-stop can + # kill this process tree while communicate() blocks the pool thread. + from agent_core.core.impl.action.cancellation import ( + register_process, + unregister_process, + was_killed, + ) + + run_session_id = input_data.get("_session_id") or "" + + # Use CREATE_NEW_PROCESS_GROUP so we can kill the entire process tree + fg_flags = creation_flags | subprocess.CREATE_NEW_PROCESS_GROUP + process = subprocess.Popen( + args, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + stdin=subprocess.DEVNULL, + cwd=cwd if cwd else None, + env=env, + text=True, + errors="replace", + creationflags=fg_flags, + ) + register_process(run_session_id, process) + + try: + stdout, stderr = process.communicate(timeout=timeout_seconds) + if was_killed(process): + # A user stop killed it; the exit code says nothing about + # the script itself. + return { + "status": "error", + "stdout": stdout.strip() if stdout else "", + "stderr": stderr.strip() if stderr else "", + "return_code": -1, + "message": "Stopped by the user before the script finished.", + "pid": None, + } + return { + "status": "success" if process.returncode == 0 else "error", + "stdout": stdout.strip() if stdout else "", + "stderr": stderr.strip() if stderr else "", + "return_code": process.returncode, + "message": "", + "pid": None, + } + except subprocess.TimeoutExpired: + # Kill the entire process tree on Windows using taskkill + try: + subprocess.run( + ["taskkill", "/F", "/T", "/PID", str(process.pid)], + capture_output=True, + creationflags=creation_flags, + ) + except Exception: + pass + process.kill() + stdout, stderr = process.communicate() + return { + "status": "error", + "stdout": (stdout or "").strip(), + "stderr": (stderr or "").strip(), + "return_code": -1, + "message": f"Timed out after {timeout_seconds}s.", + "pid": None, + } + finally: + unregister_process(run_session_id, process) + except Exception as e: + return { + "status": "error", + "stdout": "", + "stderr": str(e), + "return_code": -1, + "message": str(e), + "pid": None, + } diff --git a/app/data/action/run_shell.py b/app/data/action/run_shell.py index 59a9448f..6dfe1218 100644 --- a/app/data/action/run_shell.py +++ b/app/data/action/run_shell.py @@ -156,6 +156,7 @@ def shell_exec(input_data: dict) -> dict: from agent_core.core.impl.action.cancellation import ( register_process, unregister_process, + was_killed, ) run_session_id = input_data.get("_session_id") or "" @@ -176,6 +177,17 @@ def shell_exec(input_data: dict) -> dict: try: stdout, stderr = process.communicate(timeout=timeout_seconds) + if was_killed(process): + # A user stop killed it; the exit code says nothing about + # the command itself. + return { + "status": "error", + "stdout": stdout.strip() if stdout else "", + "stderr": stderr.strip() if stderr else "", + "return_code": -1, + "message": "Stopped by the user before the command finished.", + "pid": None, + } return { "status": "success" if process.returncode == 0 else "error", "stdout": stdout.strip() if stdout else "", @@ -218,23 +230,15 @@ def shell_exec(input_data: dict) -> dict: @action( name="run_shell", - description=( - "Executes Windows shell source, capturing stdout, stderr and exit code. " - "Stdin is closed. Set background=true for servers/watchers." - ), + description="Executes a command line with cmd.exe, capturing stdout, stderr, and exit code. Stdin is closed (EOF) by default. For PowerShell, use run_powershell. IMPORTANT: For long-running commands that don't terminate (e.g., 'npm run dev', 'npm start', 'python -m http.server', 'flask run', watch processes, dev servers), you MUST set background=true. Otherwise, the command will block the entire task until timeout and may not capture any output.", platforms=["windows"], default=True, action_sets=["core"], input_schema={ "command": { "type": "string", - "example": "Get-CimInstance Win32_LogicalDisk -Filter 'DriveType=3'", - "description": "Raw source for the selected shell; the tool starts the interpreter. Omit powershell -Command/cmd /c wrappers unless intentional.", - }, - "shell": { - "type": "string", - "example": "powershell", - "description": "cmd (default/auto), powershell, or pwsh. Use powershell/pwsh for PowerShell source.", + "example": "dir C:\\\\Windows\\\\System32", + "description": "The cmd.exe command line to execute.", }, "timeout": { "type": "integer", @@ -259,21 +263,6 @@ def shell_exec(input_data: dict) -> dict: }, }, output_schema={ - "shell": { - "type": "string", - "example": "powershell", - "description": "Normalized shell selected for this execution.", - }, - "shell_executable": { - "type": "string", - "example": "powershell.exe", - "description": "Resolved interpreter path, or executable name if unavailable. Null when no launch was prepared.", - }, - "guidance": { - "type": "string", - "example": "", - "description": "Advisory for nested PowerShell wrappers. Commands are never rewritten or retried automatically.", - }, "status": {"type": "string", "example": "success"}, "stdout": {"type": "string", "example": "Command output text"}, "stderr": {"type": "string", "example": ""}, @@ -287,7 +276,6 @@ def shell_exec(input_data: dict) -> dict: }, test_payload={ "command": "dir C:\\\\Windows\\\\System32", - "shell": "auto", "timeout": 600, "cwd": "/home/user", "env": {"MY_VAR": "123"}, @@ -297,75 +285,22 @@ def shell_exec(input_data: dict) -> dict: ) def shell_exec_windows(input_data: dict) -> dict: import os - import re - import shutil import subprocess - shell_choice = str(input_data.get("shell", "cmd")).strip().lower() - if shell_choice in ("", "auto"): - shell_choice = "cmd" - shell_executable = None - command = str(input_data.get("command", "")).strip() - # Advisory only: intentional nested shells remain supported. Never rewrite - # or re-execute a command that might already have had side effects. - nested_shell = re.match( - r"^(?:cmd(?:\.exe)?\s+(?:/[dsk]\s+)*/c\s+)?" - r"(?:powershell|pwsh)(?:\.exe)?\s+.*?-(?:command|c)\b", - command, - re.IGNORECASE | re.DOTALL, - ) - - def result(output): - output["shell"] = shell_choice - output["shell_executable"] = shell_executable - # Preserve native stderr and exit status; stderr text is not a reliable - # error-type protocol across shells, applications or localized output. - output["guidance"] = ( - "Nested PowerShell -Command detected: outer shells may expand variables " - "or quotes, even with exit code 0. Prefer shell='powershell'/'pwsh' with " - "raw script; verify stdout if nesting is intentional." - if nested_shell - else "" - ) - return output - simulated_mode = input_data.get("simulated_mode", False) if simulated_mode: # Return mock result for testing - return result( - { - "status": "success", - "stdout": "Simulated command output", - "stderr": "", - "return_code": 0, - "message": "", - "pid": None, - } - ) + return { + "status": "success", + "stdout": "Simulated command output", + "stderr": "", + "return_code": 0, + "message": "", + "pid": None, + } - if shell_choice not in ("cmd", "powershell", "pwsh"): - # Previously any unsupported value (e.g. "bash", "sh", "zsh") was - # silently coerced to cmd, so a bash heredoc would run under cmd and - # fail with a cryptic "<< was unexpected at this time." Return an - # explicit error instead so the caller knows its shell choice was - # rejected and why. - return result( - { - "status": "error", - "stdout": "", - "stderr": "", - "return_code": -1, - "message": ( - f"Shell '{shell_choice}' is not available on Windows. " - "Supported shells: cmd, powershell, pwsh. " - "bash/zsh/sh syntax (e.g. heredocs) will NOT run here — " - "use PowerShell for scripting, or write files via a file action " - "rather than shell redirection." - ), - "pid": None, - } - ) + command = str(input_data.get("command", "")).strip() timeout_val = input_data.get("timeout") cwd = input_data.get("cwd") env_input = input_data.get("env") or {} @@ -374,28 +309,24 @@ def result(output): timeout_seconds = float(timeout_val) if timeout_val is not None else 600.0 if not command: - return result( - { - "status": "error", - "stdout": "", - "stderr": "", - "return_code": -1, - "message": "command is required.", - "pid": None, - } - ) + return { + "status": "error", + "stdout": "", + "stderr": "", + "return_code": -1, + "message": "command is required.", + "pid": None, + } if cwd and not os.path.isdir(cwd): - return result( - { - "status": "error", - "stdout": "", - "stderr": "", - "return_code": -1, - "message": "Working directory does not exist.", - "pid": None, - } - ) + return { + "status": "error", + "stdout": "", + "stderr": "", + "return_code": -1, + "message": "Working directory does not exist.", + "pid": None, + } # Resolved Node runtime leads PATH: the agent is instructed to run the # agent-app CLI (TypeScript, needs node >= 24) via bare `node` through this @@ -406,36 +337,11 @@ def result(output): for k, v in env_input.items(): env[str(k)] = str(v) - shell_executable = ( - shutil.which(shell_choice + ".exe", path=env.get("PATH")) - or shell_choice + ".exe" - ) - if shell_choice == "powershell": - args = [ - shell_executable, - "-NoLogo", - "-NonInteractive", - "-NoProfile", - "-ExecutionPolicy", - "Bypass", - "-Command", - command, - ] - elif shell_choice == "pwsh": - args = [ - shell_executable, - "-NoLogo", - "-NonInteractive", - "-NoProfile", - "-Command", - command, - ] - else: - # Build the command line as a raw string: passing a list makes Popen - # escape embedded quotes as \" (MSVCRT rules), which cmd.exe does not - # understand, mangling any command containing a quoted path. With - # /s /c, cmd strips the outer quotes and runs the command verbatim. - args = '"' + shell_executable + '" /d /s /c "' + command + '"' + # Build the command line as a raw string: passing a list makes Popen + # escape embedded quotes as \" (MSVCRT rules), which cmd.exe does not + # understand, mangling any command containing a quoted path. With + # /s /c, cmd strips the outer quotes and runs the command verbatim. + args = 'cmd.exe /d /s /c "' + command + '"' creation_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0) @@ -453,27 +359,23 @@ def result(output): env=env, creationflags=bg_flags, ) - return result( - { - "status": "background", - "stdout": "", - "stderr": "", - "return_code": 0, - "message": f"Process started in background with PID {process.pid}", - "pid": process.pid, - } - ) + return { + "status": "background", + "stdout": "", + "stderr": "", + "return_code": 0, + "message": f"Process started in background with PID {process.pid}", + "pid": process.pid, + } except Exception as e: - return result( - { - "status": "error", - "stdout": "", - "stderr": str(e), - "return_code": -1, - "message": str(e), - "pid": None, - } - ) + return { + "status": "error", + "stdout": "", + "stderr": str(e), + "return_code": -1, + "message": str(e), + "pid": None, + } # Foreground mode with proper timeout handling try: @@ -482,6 +384,7 @@ def result(output): from agent_core.core.impl.action.cancellation import ( register_process, unregister_process, + was_killed, ) run_session_id = input_data.get("_session_id") or "" @@ -503,16 +406,25 @@ def result(output): try: stdout, stderr = process.communicate(timeout=timeout_seconds) - return result( - { - "status": "success" if process.returncode == 0 else "error", + if was_killed(process): + # A user stop killed it; the exit code says nothing about + # the command itself. + return { + "status": "error", "stdout": stdout.strip() if stdout else "", "stderr": stderr.strip() if stderr else "", - "return_code": process.returncode, - "message": "", + "return_code": -1, + "message": "Stopped by the user before the command finished.", "pid": None, } - ) + return { + "status": "success" if process.returncode == 0 else "error", + "stdout": stdout.strip() if stdout else "", + "stderr": stderr.strip() if stderr else "", + "return_code": process.returncode, + "message": "", + "pid": None, + } except subprocess.TimeoutExpired: # Kill the entire process tree on Windows using taskkill try: @@ -525,29 +437,25 @@ def result(output): pass process.kill() stdout, stderr = process.communicate() - return result( - { - "status": "error", - "stdout": (stdout or "").strip(), - "stderr": (stderr or "").strip(), - "return_code": -1, - "message": f"Timed out after {timeout_seconds}s.", - "pid": None, - } - ) - finally: - unregister_process(run_session_id, process) - except Exception as e: - return result( - { + return { "status": "error", - "stdout": "", - "stderr": str(e), + "stdout": (stdout or "").strip(), + "stderr": (stderr or "").strip(), "return_code": -1, - "message": str(e), + "message": f"Timed out after {timeout_seconds}s.", "pid": None, } - ) + finally: + unregister_process(run_session_id, process) + except Exception as e: + return { + "status": "error", + "stdout": "", + "stderr": str(e), + "return_code": -1, + "message": str(e), + "pid": None, + } @action( @@ -711,6 +619,7 @@ def shell_exec_darwin(input_data: dict) -> dict: from agent_core.core.impl.action.cancellation import ( register_process, unregister_process, + was_killed, ) run_session_id = input_data.get("_session_id") or "" @@ -730,6 +639,17 @@ def shell_exec_darwin(input_data: dict) -> dict: try: stdout, stderr = process.communicate(timeout=timeout_seconds) + if was_killed(process): + # A user stop killed it; the exit code says nothing about + # the command itself. + return { + "status": "error", + "stdout": stdout.strip() if stdout else "", + "stderr": stderr.strip() if stderr else "", + "return_code": -1, + "message": "Stopped by the user before the command finished.", + "pid": None, + } return { "status": "success" if process.returncode == 0 else "error", "stdout": stdout.strip() if stdout else "", diff --git a/tests/test_shell_actions.py b/tests/test_shell_actions.py new file mode 100644 index 00000000..65c3d0a6 --- /dev/null +++ b/tests/test_shell_actions.py @@ -0,0 +1,116 @@ +"""Shell actions: which OS is offered which one, and how a stop is told apart +from a failure. + +run_powershell has an implementation for Windows alone. Every other action +has one for each OS (or for "all"), so it is the one name in the registry +with nothing to run on Linux or macOS. +""" + +import platform +import subprocess +import sys + +import pytest + +from agent_core.core.action_framework.loader import load_actions_from_directories +from agent_core.core.action_framework.registry import registry_instance +from agent_core.core.impl.action import cancellation +from app.action.action_set import action_set_manager +from app.config import PROJECT_ROOT + + +@pytest.fixture(scope="module", autouse=True) +def actions_loaded(): + load_actions_from_directories( + base_dir=str(PROJECT_ROOT), paths_to_scan=["app/data/action"] + ) + + +@pytest.fixture +def on_os(monkeypatch): + def switch(system): + monkeypatch.setattr(platform, "system", lambda: system) + + return switch + + +# ── run_powershell is Windows-only ──────────────────────────────────────── + + +@pytest.mark.parametrize("system", ["Linux", "Darwin"]) +def test_other_os_neither_sees_nor_resolves_run_powershell(on_os, system): + on_os(system) + # Not offered: the compiled list is what the model is shown. + assert "run_powershell" not in action_set_manager.compile_action_list([]) + assert "run_powershell" not in action_set_manager.get_actions_in_set("core") + # Not callable either, should a model produce the name unprompted. + assert registry_instance.find_action_by_name("run_powershell") is None + + +@pytest.mark.parametrize("system", ["Linux", "Darwin", "Windows"]) +def test_every_os_keeps_its_own_run_shell(on_os, system): + on_os(system) + assert "run_shell" in action_set_manager.compile_action_list([]) + assert registry_instance.find_action_by_name("run_shell") is not None + + +def test_windows_is_offered_both_shell_actions(on_os): + on_os("Windows") + compiled = action_set_manager.compile_action_list([]) + assert {"run_shell", "run_powershell"} <= set(compiled) + + powershell = registry_instance.find_action_by_name("run_powershell") + assert powershell["platforms"] == ["windows"] + assert next(iter(powershell["input_schema"])) == "script" + # cmd.exe only: PowerShell has its own action, so there is no shell to pick. + shell = registry_instance.find_action_by_name("run_shell") + assert "shell" not in shell["input_schema"] + + +def test_unix_run_shell_schemas_are_untouched(on_os): + for system in ("Linux", "Darwin"): + on_os(system) + shell = registry_instance.find_action_by_name("run_shell") + assert list(shell["input_schema"]) == [ + "command", + "shell", + "timeout", + "cwd", + "env", + "background", + ] + assert "run_powershell" not in shell["description"] + + +# ── a stop is not a failure ─────────────────────────────────────────────── + + +@pytest.fixture +def session(monkeypatch, tmp_path): + monkeypatch.setattr(cancellation, "_marker_dir", lambda sid: tmp_path / sid) + return "sess-shell" + + +@pytest.fixture +def sleeper(): + proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(120)"]) + yield proc + if proc.poll() is None: + proc.kill() + proc.wait(timeout=10) + + +def test_stop_marks_the_registered_process_it_killed(session, sleeper): + cancellation.register_process(session, sleeper) + assert not cancellation.was_killed(sleeper) + assert cancellation.kill_session_processes(session) == 1 + assert sleeper.poll() is not None + assert cancellation.was_killed(sleeper) + + +def test_process_that_already_exited_is_not_marked(session, sleeper): + cancellation.register_process(session, sleeper) + sleeper.kill() + sleeper.wait(timeout=10) + assert cancellation.kill_session_processes(session) == 0 + assert not cancellation.was_killed(sleeper) diff --git a/tests/windows_shell/test_run_shell_windows.py b/tests/windows_shell/test_run_shell_windows.py index c1d2109a..80a70ac0 100644 --- a/tests/windows_shell/test_run_shell_windows.py +++ b/tests/windows_shell/test_run_shell_windows.py @@ -1,19 +1,19 @@ """Real Windows shell regressions, runnable with only the Python standard library. Run: python -m unittest discover -s tests/windows_shell -v -Like the action registry, load the function's source without its decorator. +Like the action registry, load each function's source without its decorator. Only unrelated Node PATH setup and cancellation registration are stubbed; -commands execute through the production launcher using real OS processes. +commands execute through the production launchers using real OS processes. """ import ast import json import os from pathlib import Path -import shutil import subprocess import sys import tempfile +import threading import time import types import unittest @@ -21,15 +21,17 @@ ROOT = Path(__file__).resolve().parents[2] -SOURCE = ROOT / "app/data/action/run_shell.py" +ACTIONS = ROOT / "app/data/action" +HERE = Path(__file__).resolve().parent -def load_action(): - tree = ast.parse(SOURCE.read_text(encoding="utf-8")) +def load_action(file_name, function_name): + source = ACTIONS / file_name + tree = ast.parse(source.read_text(encoding="utf-8")) function = next( node for node in tree.body - if isinstance(node, ast.FunctionDef) and node.name == "shell_exec_windows" + if isinstance(node, ast.FunctionDef) and node.name == function_name ) metadata = { keyword.arg: ast.literal_eval(keyword.value) @@ -38,52 +40,71 @@ def load_action(): function.decorator_list = [] namespace = {} exec( - compile(ast.Module(body=[function], type_ignores=[]), str(SOURCE), "exec"), + compile(ast.Module(body=[function], type_ignores=[]), str(source), "exec"), namespace, ) - return namespace["shell_exec_windows"], metadata + return namespace[function_name], metadata -class ShellGuidanceTests(unittest.TestCase): - def setUp(self): - self.run_shell, self.metadata = load_action() - - def test_schema_teaches_raw_powershell(self): - self.assertIn( - "Raw source", self.metadata["input_schema"]["command"]["description"] - ) - self.assertEqual( - self.metadata["input_schema"]["shell"]["example"], "powershell" - ) - self.assertIn( - "Get-CimInstance", self.metadata["input_schema"]["command"]["example"] - ) - self.assertIn("guidance", self.metadata["output_schema"]) +def load_actions(test): + test.run_shell, test.shell_metadata = load_action( + "run_shell.py", "shell_exec_windows" + ) + test.run_powershell, test.powershell_metadata = load_action( + "run_powershell.py", "powershell_exec_windows" + ) - def test_invalid_shell_does_not_launch(self): - with patch("subprocess.Popen") as spawn: - result = self.run_shell({"command": "echo x", "shell": "bash"}) - spawn.assert_not_called() - self.assertEqual(result["status"], "error") - self.assertEqual(result["shell"], "bash") - self.assertIsNone(result["shell_executable"]) - def test_empty_command_reports_default_without_launch(self): - result = self.run_shell({"command": "", "shell": "auto"}) - self.assertEqual(result["shell"], "cmd") - self.assertIsNone(result["shell_executable"]) - self.assertEqual(result["message"], "command is required.") +class ShellSchemaTests(unittest.TestCase): + def setUp(self): + load_actions(self) + + def test_powershell_is_a_windows_only_action_that_takes_a_script(self): + self.assertEqual(self.powershell_metadata["name"], "run_powershell") + self.assertEqual(self.powershell_metadata["platforms"], ["windows"]) + params = list(self.powershell_metadata["input_schema"]) + # The interpreter is the action itself, so the model commits to it + # before it writes any source, and there is no shell left to choose. + self.assertEqual(params[0], "script") + self.assertNotIn("command", params) + self.assertNotIn("shell", params) + + def test_windows_run_shell_is_cmd_only(self): + self.assertEqual(self.shell_metadata["platforms"], ["windows"]) + self.assertNotIn("shell", self.shell_metadata["input_schema"]) + self.assertIn("cmd.exe", self.shell_metadata["description"]) + self.assertIn("run_powershell", self.shell_metadata["description"]) + + def test_results_carry_no_advice_fields(self): + for metadata in (self.shell_metadata, self.powershell_metadata): + self.assertEqual( + sorted(metadata["output_schema"]), + ["message", "pid", "return_code", "status", "stderr", "stdout"], + ) + + def test_empty_source_does_not_launch(self): + cases = ( + (self.run_shell, {"command": ""}, "command is required."), + (self.run_powershell, {"script": ""}, "script is required."), + ) + for action, payload, message in cases: + with self.subTest(message=message), patch("subprocess.Popen") as spawn: + result = action(payload) + spawn.assert_not_called() + self.assertEqual(result["status"], "error") + self.assertEqual(result["message"], message) @unittest.skipUnless(sys.platform == "win32", "Requires real Windows interpreters") class WindowsShellTests(unittest.TestCase): def setUp(self): - self.run_shell, self.metadata = load_action() + load_actions(self) app = types.ModuleType("app") app.node_runtime = types.SimpleNamespace(child_env=lambda: os.environ.copy()) self.cancellation = types.ModuleType("agent_core.core.impl.action.cancellation") self.cancellation.register_process = Mock() self.cancellation.unregister_process = Mock() + self.cancellation.was_killed = Mock(return_value=False) modules = patch.dict( sys.modules, {"app": app, self.cancellation.__name__: self.cancellation}, @@ -91,19 +112,20 @@ def setUp(self): modules.start() self.addCleanup(modules.stop) - def execute(self, command, shell="powershell", **kwargs): - if not shutil.which(shell + ".exe"): - self.skipTest(shell + " is unavailable") - return self.run_shell( - {"command": command, "shell": shell, "timeout": 15, **kwargs} - ) + def powershell(self, script, **kwargs): + return self.run_powershell({"script": script, "timeout": 15, **kwargs}) + + def cmd(self, command, **kwargs): + return self.run_shell({"command": command, "timeout": 15, **kwargs}) def assert_output(self, result, expected): self.assertEqual(result["status"], "success", result) self.assertEqual(result["return_code"], 0, result) self.assertEqual(result["stdout"].replace("\r\n", "\n"), expected) - def test_direct_scripts_preserve_semantics(self): + # ── run_powershell ─────────────────────────────────────────────────── + + def test_scripts_run_as_written(self): cases = [ ("Write-Output 'hello world'", "hello world"), ('Write-Output "hello world"', "hello world"), @@ -119,21 +141,12 @@ def test_direct_scripts_preserve_semantics(self): "a&b|c(d)%CRAFTBOT_QUOTE_REPRO%", ), ] - for shell in ("powershell", "pwsh"): - if not shutil.which(shell + ".exe"): - continue - for command, expected in cases: - with self.subTest(shell=shell, command=command): - result = self.execute(command, shell) - self.assert_output(result, expected) - self.assertEqual(result["guidance"], "") - self.assertEqual(result["shell"], shell) - self.assertEqual( - result["shell_executable"], shutil.which(shell + ".exe") - ) + for script, expected in cases: + with self.subTest(script=script): + self.assert_output(self.powershell(script), expected) def test_disk_query_with_parenthesized_calculated_property(self): - result = self.execute( + result = self.powershell( "@(Get-CimInstance Win32_LogicalDisk -Filter 'DriveType=3' | " 'Select-Object @{Name="Free (GB)";Expression={[math]::Round($_.FreeSpace/1GB,2)}}) ' "| ConvertTo-Json -Compress" @@ -145,95 +158,48 @@ def test_disk_query_with_parenthesized_calculated_property(self): all(isinstance(disk["Free (GB)"], (int, float)) for disk in disks) ) - def test_nested_variables_fail_with_actionable_guidance(self): - result = self.execute('powershell -NoProfile -Command "$n=2; Write-Output $n"') - self.assertEqual(result["status"], "error", result) - self.assertIn("=2", result["stderr"]) - self.assertIn("raw script", result["guidance"]) - self.assertEqual(self.cancellation.register_process.call_count, 1) - self.assertEqual(self.cancellation.unregister_process.call_count, 1) - self.assert_output(self.execute("$n=2; Write-Output $n"), "2") - - def test_single_quote_wrapper_silent_wrong_output_is_flagged(self): - result = self.execute( - "powershell -NoProfile -Command 'Write-Output \"hello world\"'", "cmd" - ) - self.assert_output(result, "Write-Output hello world") - self.assertIn("even with exit code 0", result["guidance"]) - self.assert_output(self.execute("Write-Output 'hello world'"), "hello world") - - def test_cmd_percent_expansion_is_flagged_and_raw_script_preserves_literal(self): - env = {"CRAFTBOT_QUOTE_REPRO": "expanded-by-cmd"} - result = self.execute( - "powershell -NoProfile -Command \"Write-Output '%CRAFTBOT_QUOTE_REPRO%'\"", - "cmd", - env=env, - ) - self.assert_output(result, "expanded-by-cmd") - self.assertIn("expand variables", result["guidance"]) - self.assert_output( - self.execute("Write-Output '%CRAFTBOT_QUOTE_REPRO%'", env=env), - "%CRAFTBOT_QUOTE_REPRO%", + def test_largest_files_query_keeps_its_pipeline_variable(self): + # The script a model wrote for "what is taking up my space?". Sent + # inside a `powershell -Command "..."` wrapper it lost every `$_` to + # the outer interpreter; as a script it runs as written. + folder = str(HERE).replace("'", "''") + result = self.powershell( + f"Get-ChildItem '{folder}' -Recurse -File -ErrorAction SilentlyContinue | " + "Sort Length -Descending | Select -First 1 Name, " + "@{Name='SizeKB';Expression={[math]::round($_.Length/1KB,2)}} " + "| ConvertTo-Json -Compress" ) - - def test_intentional_nested_wrapper_still_executes_once(self): - result = self.execute( - "cmd /c powershell -NoProfile -Command \"Write-Output 'ok'\"", "cmd" + self.assertEqual(result["status"], "success", result) + largest = json.loads(result["stdout"]) + self.assertTrue(largest["Name"]) + self.assertIsInstance(largest["SizeKB"], (int, float)) + + def test_literal_percent_expression_is_not_expanded(self): + result = self.powershell( + "Write-Output '%CRAFTBOT_QUOTE_REPRO%'", + env={"CRAFTBOT_QUOTE_REPRO": "expanded-by-cmd"}, ) - self.assert_output(result, "ok") - self.assertIn("intentional", result["guidance"]) - self.assertEqual(self.cancellation.register_process.call_count, 1) + self.assert_output(result, "%CRAFTBOT_QUOTE_REPRO%") def test_parser_failure_preserves_native_error(self): - result = self.execute("$x = (") + result = self.powershell("$x = (") self.assertEqual(result["status"], "error", result) self.assertNotEqual(result["return_code"], 0) self.assertTrue(result["stderr"]) - self.assertEqual(result["guidance"], "") - self.assertEqual(self.cancellation.register_process.call_count, 1) + self.assertEqual(result["message"], "") - def test_application_error_text_is_not_classified_as_a_parser_failure(self): + def test_application_stderr_and_exit_code_pass_through(self): stderr = "UnexpectedToken: application-defined failure" - result = self.execute(f"[Console]::Error.WriteLine('{stderr}'); exit 7") + result = self.powershell(f"[Console]::Error.WriteLine('{stderr}'); exit 7") self.assertEqual(result["return_code"], 7) self.assertEqual(result["stderr"], stderr) - self.assertEqual(result["guidance"], "") def test_successful_stderr_does_not_imply_failure(self): - result = self.execute( + result = self.powershell( "[Console]::Error.WriteLine('warning'); Write-Output 'ok'" ) self.assert_output(result, "ok") self.assertEqual(result["stderr"], "warning") - self.assertEqual(result["guidance"], "") - - def test_cmd_default_auto_and_explicit_remain_compatible(self): - for choice in (None, "auto", "cmd"): - with self.subTest(shell=choice): - payload = {"command": "echo cmd-compatible", "timeout": 15} - if choice: - payload["shell"] = choice - result = self.run_shell(payload) - self.assert_output(result, "cmd-compatible") - self.assertEqual(result["shell"], "cmd") - self.assertEqual(result["guidance"], "") - - def test_cmd_quoted_executable_path(self): - result = self.execute(f'"{sys.executable}" -c "print(123)"', "cmd") - self.assert_output(result, "123") - - def test_mentions_inside_script_do_not_trigger_wrapper_advice(self): - result = self.execute("Write-Output 'powershell -Command example'") - self.assert_output(result, "powershell -Command example") - self.assertEqual(result["guidance"], "") - - def test_timeout_unregisters_process(self): - result = self.execute("Start-Sleep -Seconds 10", timeout=1) - self.assertEqual(result["status"], "error") - self.assertIn("Timed out", result["message"]) - process = self.cancellation.register_process.call_args.args[1] - self.assertIsNotNone(process.poll()) - self.cancellation.unregister_process.assert_called_once_with("", process) def test_background_returns_pid_and_completes(self): with tempfile.TemporaryDirectory() as directory: @@ -248,7 +214,7 @@ def track_process(*args, **kwargs): return process with patch("subprocess.Popen", side_effect=track_process): - result = self.execute( + result = self.powershell( f"Start-Sleep -Milliseconds 300; [IO.File]::WriteAllText('{escaped}', 'done')", background=True, ) @@ -267,6 +233,77 @@ def track_process(*args, **kwargs): ) processes[0].wait(timeout=5) + # ── run_shell (cmd.exe) ────────────────────────────────────────────── + + def test_cmd_runs_command_lines(self): + self.assert_output(self.cmd("echo cmd-compatible"), "cmd-compatible") + self.assert_output( + self.cmd("echo %CRAFTBOT_SHELL_TEST%", env={"CRAFTBOT_SHELL_TEST": "set"}), + "set", + ) + + def test_cmd_quoted_executable_path(self): + result = self.cmd(f'"{sys.executable}" -c "print(123)"') + self.assert_output(result, "123") + + def test_cmd_can_still_start_powershell_as_a_command_line(self): + result = self.cmd("powershell -NoProfile -Command \"Write-Output 'ok'\"") + self.assert_output(result, "ok") + self.assertEqual(self.cancellation.register_process.call_count, 1) + + # ── both actions ───────────────────────────────────────────────────── + + def test_timeout_unregisters_process(self): + cases = ( + (self.powershell, "Start-Sleep -Seconds 10"), + (self.cmd, "ping -n 11 127.0.0.1 >nul"), + ) + for run, source in cases: + with self.subTest(source=source): + self.cancellation.register_process.reset_mock() + self.cancellation.unregister_process.reset_mock() + result = run(source, timeout=1) + self.assertEqual(result["status"], "error") + self.assertIn("Timed out", result["message"]) + process = self.cancellation.register_process.call_args.args[1] + self.assertIsNotNone(process.poll()) + self.cancellation.unregister_process.assert_called_once_with( + "", process + ) + + def test_user_stop_is_reported_as_a_stop_not_a_failure(self): + # Stand in for the run-cancel registry: half a second after the + # process registers, mark it and kill its tree, as a user stop does. + killed = [] + + def stop_soon(_session_id, process): + def stop(): + killed.append(process) + subprocess.run( + ["taskkill", "/F", "/T", "/PID", str(process.pid)], + capture_output=True, + creationflags=subprocess.CREATE_NO_WINDOW, + ) + + threading.Timer(0.5, stop).start() + + self.cancellation.register_process.side_effect = stop_soon + self.cancellation.was_killed.side_effect = lambda process: process in killed + + cases = ( + (self.powershell, "Start-Sleep -Seconds 30", "script"), + (self.cmd, "ping -n 31 127.0.0.1 >nul", "command"), + ) + for run, source, noun in cases: + with self.subTest(source=source): + result = run(source) + self.assertEqual(result["status"], "error", result) + self.assertEqual(result["return_code"], -1, result) + self.assertEqual( + result["message"], + f"Stopped by the user before the {noun} finished.", + ) + if __name__ == "__main__": unittest.main() From 77ac1ec5096bd0ac368cc1e09716964b85253e18 Mon Sep 17 00:00:00 2001 From: ahmad-ajmal Date: Tue, 13 Oct 2026 09:50:28 +0100 Subject: [PATCH 4/5] fix #475: reject non-cmd Windows shell selections Signed-off-by: ahmad-ajmal --- app/data/action/run_shell.py | 17 ++++++ tests/windows_shell/test_run_shell_windows.py | 53 +++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/app/data/action/run_shell.py b/app/data/action/run_shell.py index 6dfe1218..5cbd81ef 100644 --- a/app/data/action/run_shell.py +++ b/app/data/action/run_shell.py @@ -287,6 +287,23 @@ def shell_exec_windows(input_data: dict) -> dict: import os import subprocess + # Older callers may still send the removed shell parameter. Refuse a + # different interpreter instead of executing its source under cmd.exe. + shell_choice = str(input_data.get("shell", "cmd")).strip().lower() + if shell_choice not in ("cmd", "auto", ""): + return { + "status": "error", + "stdout": "", + "stderr": "", + "return_code": -1, + "message": ( + f"Windows run_shell only supports cmd.exe; requested shell '{shell_choice}'. " + "Use run_powershell with the script parameter for Windows PowerShell, " + "or explicitly launch the desired executable from a cmd command line." + ), + "pid": None, + } + simulated_mode = input_data.get("simulated_mode", False) if simulated_mode: diff --git a/tests/windows_shell/test_run_shell_windows.py b/tests/windows_shell/test_run_shell_windows.py index 80a70ac0..d5b6731d 100644 --- a/tests/windows_shell/test_run_shell_windows.py +++ b/tests/windows_shell/test_run_shell_windows.py @@ -94,6 +94,33 @@ def test_empty_source_does_not_launch(self): self.assertEqual(result["status"], "error") self.assertEqual(result["message"], message) + def test_non_cmd_shell_is_rejected_before_launch(self): + app = types.ModuleType("app") + child_env = Mock(return_value=os.environ.copy()) + app.node_runtime = types.SimpleNamespace(child_env=child_env) + shells = ("powershell", "pwsh", "bash", "sh", "zsh", " PowerShell ", None) + for shell in shells: + for mode in ({}, {"background": True}, {"simulated_mode": True}): + with ( + self.subTest(shell=shell, mode=mode), + patch.dict(sys.modules, {"app": app}), + patch("subprocess.Popen") as spawn, + patch("subprocess.run") as run, + ): + result = self.run_shell( + {"command": "echo should-not-run", "shell": shell, **mode} + ) + self.assertEqual(result["status"], "error", result) + self.assertEqual(result["return_code"], -1, result) + self.assertEqual(result["stdout"], "") + self.assertEqual(result["stderr"], "") + self.assertIsNone(result["pid"]) + self.assertIn("cmd.exe", result["message"]) + self.assertIn("run_powershell", result["message"]) + spawn.assert_not_called() + run.assert_not_called() + child_env.assert_not_called() + @unittest.skipUnless(sys.platform == "win32", "Requires real Windows interpreters") class WindowsShellTests(unittest.TestCase): @@ -242,6 +269,32 @@ def test_cmd_runs_command_lines(self): "set", ) + def test_legacy_cmd_selections_still_run(self): + for shell in ("cmd", " CMD ", "auto", " AUTO ", "", " "): + with self.subTest(shell=shell): + self.assert_output( + self.cmd("echo cmd-compatible", shell=shell), "cmd-compatible" + ) + + def test_legacy_powershell_selection_cannot_expand_cmd_variables(self): + result = self.cmd( + "echo %CRAFTBOT_SHELL_TEST%", + shell="powershell", + env={"CRAFTBOT_SHELL_TEST": "expanded-by-cmd"}, + ) + self.assertEqual(result["status"], "error", result) + self.assertEqual(result["stdout"], "") + self.assertIn("run_powershell", result["message"]) + + def test_rejected_shell_has_no_filesystem_side_effect(self): + with tempfile.TemporaryDirectory() as directory: + marker = Path(directory) / "should-not-exist.txt" + for shell in ("powershell", "pwsh", "bash"): + with self.subTest(shell=shell): + result = self.cmd(f'echo executed>"{marker}"', shell=shell) + self.assertEqual(result["status"], "error", result) + self.assertFalse(marker.exists()) + def test_cmd_quoted_executable_path(self): result = self.cmd(f'"{sys.executable}" -c "print(123)"') self.assert_output(result, "123") From d232fcc736079adebd13d9e655d5efbef8b5e1be Mon Sep 17 00:00:00 2001 From: ahmad-ajmal Date: Wed, 14 Oct 2026 17:49:02 +0100 Subject: [PATCH 5/5] fix #475: allow PowerShell validation checks Signed-off-by: ahmad-ajmal --- app/subagent/definitions/validation_agent.py | 1 + 1 file changed, 1 insertion(+) diff --git a/app/subagent/definitions/validation_agent.py b/app/subagent/definitions/validation_agent.py index fde57373..5afee118 100644 --- a/app/subagent/definitions/validation_agent.py +++ b/app/subagent/definitions/validation_agent.py @@ -160,6 +160,7 @@ "list_folder", # Execute checks "run_shell", + "run_powershell", # External standards & API verification "web_search", "web_fetch",