core/tools: clear argv error when run_command gets a string

This commit is contained in:
2026-08-04 20:35:22 +08:00
parent c1dfdc8478
commit 23f7b4e299
3 changed files with 56 additions and 2 deletions
+4 -1
View File
@@ -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,
+22 -1
View File
@@ -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)
+30
View File
@@ -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(