diff --git a/src/plyngent/tools/process/run_command.py b/src/plyngent/tools/process/run_command.py index 8b34cd8..814dfbb 100644 --- a/src/plyngent/tools/process/run_command.py +++ b/src/plyngent/tools/process/run_command.py @@ -1,7 +1,7 @@ from __future__ import annotations from plyngent.agent import ToolTag, tool -from plyngent.tools.workspace import WorkspaceError +from plyngent.tools.workspace import WorkspaceError, argv_shape_error from .command_exec import ( DEFAULT_TIMEOUT_SECONDS, @@ -20,6 +20,9 @@ async def run_command( env: dict[str, str] | None = None, ) -> str: """Run ``command`` (argv, no shell) under the workspace; capture stdout/stderr.""" + shape_error = argv_shape_error(command) + if shape_error is not None: + return f"error: `run_command` {shape_error}" try: result = await execute_argv( command, diff --git a/src/plyngent/tools/workspace.py b/src/plyngent/tools/workspace.py index 6dfa469..d4fb0b0 100644 --- a/src/plyngent/tools/workspace.py +++ b/src/plyngent/tools/workspace.py @@ -2,7 +2,7 @@ from __future__ import annotations from dataclasses import dataclass, field from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast if TYPE_CHECKING: from collections.abc import Callable, Sequence @@ -268,6 +268,23 @@ def resolve_path(path: str | Path) -> Path: return resolved +def argv_shape_error(argv: object) -> str | None: + """Return a model-facing error when ``argv`` is not a non-empty argv list. + + Models occasionally pass a shell-style string or a JSON-encoded array as a + string; duck-typing would otherwise let it reach ``create_subprocess_exec`` + and splat into single characters. Declared as ``object`` so the isinstance + checks are not flagged as unnecessary. + """ + if isinstance(argv, str): + return 'only accepts an array of args, not a string; pass e.g. ["ls", "-la"]' + if not isinstance(argv, list): + return f"only accepts an array of args, not a {type(argv).__name__}" + if not all(isinstance(part, str) for part in cast("list[object]", argv)): + return "only accepts an array of string args" + return None + + def check_command_allowed(argv: list[str]) -> None: """Raise if argv is empty or the executable basename is denylisted. @@ -275,6 +292,10 @@ def check_command_allowed(argv: list[str]) -> None: installed: the human is asked (with a timeout; default deny). Session grants skip re-prompting for the same basename. Independent of YOLO soft-confirm. """ + shape_error = argv_shape_error(argv) + if shape_error is not None: + msg = f"command {shape_error}" + raise WorkspaceError(msg) if not argv: msg = "command argv must not be empty" raise WorkspaceError(msg) diff --git a/tests/test_tools/test_process.py b/tests/test_tools/test_process.py index 9dcda69..4154622 100644 --- a/tests/test_tools/test_process.py +++ b/tests/test_tools/test_process.py @@ -48,6 +48,36 @@ async def test_run_command_echo(workspace: object) -> None: assert "hi" in out +async def test_run_command_string_command_error(workspace: object) -> None: + del workspace + out = await call_async(run_command, "ls -la") + assert "error:" in out + assert "only accepts an array" in out + assert "executable not found" not in out + + +async def test_run_command_stringified_json_array_error(workspace: object) -> None: + del workspace + out = await call_async(run_command, '["ls", "-la"]') + assert "error:" in out + assert "only accepts an array" in out + assert "executable not found" not in out + + +async def test_run_command_non_string_args_error(workspace: object) -> None: + del workspace + out = await call_async(run_command, ["ls", 1]) + assert "error:" in out + assert "array of string args" in out + + +async def test_open_pty_string_command_error(workspace: object) -> None: + del workspace + out = await call_async(open_pty, "echo hi") + assert "error:" in out + assert "only accepts an array" in out + + async def test_run_command_batch_serial(workspace: object) -> None: del workspace out = await call_async(