diff --git a/.github/workflows/windows-shell.yml b/.github/workflows/windows-shell.yml new file mode 100644 index 00000000..86f8ee7c --- /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' + - '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' + +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 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 689b095f..5cbd81ef 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,7 +230,7 @@ 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 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"], @@ -226,12 +238,7 @@ def shell_exec(input_data: dict) -> dict: "command": { "type": "string", "example": "dir C:\\\\Windows\\\\System32", - "description": "The shell command to execute.", - }, - "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).", + "description": "The cmd.exe command line to execute.", }, "timeout": { "type": "integer", @@ -269,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"}, @@ -281,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: @@ -295,29 +318,6 @@ def shell_exec_windows(input_data: dict) -> dict: } 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, - } timeout_val = input_data.get("timeout") cwd = input_data.get("cwd") env_input = input_data.get("env") or {} @@ -354,32 +354,11 @@ def shell_exec_windows(input_data: dict) -> dict: for k, v in env_input.items(): env[str(k)] = str(v) - if shell_choice == "powershell": - args = [ - "powershell.exe", - "-NoLogo", - "-NonInteractive", - "-NoProfile", - "-ExecutionPolicy", - "Bypass", - "-Command", - command, - ] - elif shell_choice == "pwsh": - args = [ - "pwsh.exe", - "-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 = 'cmd.exe /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) @@ -422,6 +401,7 @@ def shell_exec_windows(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 "" @@ -443,6 +423,17 @@ def shell_exec_windows(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 "", @@ -645,6 +636,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 "" @@ -664,6 +656,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/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", 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 new file mode 100644 index 00000000..d5b6731d --- /dev/null +++ b/tests/windows_shell/test_run_shell_windows.py @@ -0,0 +1,362 @@ +"""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 each function's source without its decorator. +Only unrelated Node PATH setup and cancellation registration are stubbed; +commands execute through the production launchers using real OS processes. +""" + +import ast +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import threading +import time +import types +import unittest +from unittest.mock import Mock, patch + + +ROOT = Path(__file__).resolve().parents[2] +ACTIONS = ROOT / "app/data/action" +HERE = Path(__file__).resolve().parent + + +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 == function_name + ) + 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[function_name], metadata + + +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" + ) + + +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) + + 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): + def setUp(self): + 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}, + ) + modules.start() + self.addCleanup(modules.stop) + + 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) + + # ── run_powershell ─────────────────────────────────────────────────── + + def test_scripts_run_as_written(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 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.powershell( + "@(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_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" + ) + 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, "%CRAFTBOT_QUOTE_REPRO%") + + def test_parser_failure_preserves_native_error(self): + result = self.powershell("$x = (") + self.assertEqual(result["status"], "error", result) + self.assertNotEqual(result["return_code"], 0) + self.assertTrue(result["stderr"]) + self.assertEqual(result["message"], "") + + def test_application_stderr_and_exit_code_pass_through(self): + stderr = "UnexpectedToken: application-defined failure" + result = self.powershell(f"[Console]::Error.WriteLine('{stderr}'); exit 7") + self.assertEqual(result["return_code"], 7) + self.assertEqual(result["stderr"], stderr) + + def test_successful_stderr_does_not_imply_failure(self): + result = self.powershell( + "[Console]::Error.WriteLine('warning'); Write-Output 'ok'" + ) + self.assert_output(result, "ok") + self.assertEqual(result["stderr"], "warning") + + 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.powershell( + 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) + + # ── 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_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") + + 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()