From 587e9ec229a5a206c8db64b6ef3d694fd0eeb521 Mon Sep 17 00:00:00 2001 From: ahmad-ajmal Date: Wed, 7 Oct 2026 11:54:13 +0100 Subject: [PATCH] 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()