core/tools: rename run_command to run_argv (argv param)

This commit is contained in:
2026-08-04 21:38:20 +08:00
parent 23f7b4e299
commit d698fbf06f
8 changed files with 49 additions and 49 deletions
+1 -1
View File
@@ -41,7 +41,7 @@ from .process import ask_into_pty as ask_into_pty
from .process import close_pty as close_pty
from .process import open_pty as open_pty
from .process import read_pty as read_pty
from .process import run_command as run_command
from .process import run_argv as run_argv
from .process import run_command_batch as run_command_batch
from .process import write_pty as write_pty
from .process import write_pty_keys as write_pty_keys
+8 -8
View File
@@ -59,8 +59,8 @@ def _basename(argv0: str) -> str:
return name.lower()
def _as_argv(args: Mapping[str, object]) -> list[str] | None:
command = args.get("command")
def _as_argv(args: Mapping[str, object], *, key: str = "command") -> list[str] | None:
command = args.get(key)
if not isinstance(command, list) or not command:
return None
out: list[str] = []
@@ -110,7 +110,7 @@ def _shell_or_dash_c_reason(argv: Sequence[str], *, via: str) -> str | None:
"""Confirm interactive shells and ``*-c`` one-liners so the user can inspect argv.
Multi-line reason (shown inside the CLI confirm box). ``via`` is a short
label such as ``run_command`` or ``open_pty``.
label such as ``run_argv`` or ``open_pty``.
For ``-c`` scripts, argv shows ``$(command)`` instead of inlining the body;
the full script is printed below untruncated, with every line indented.
@@ -181,11 +181,11 @@ def _delete_path_reason(args: Mapping[str, object]) -> str | None:
return f"delete path {path!r}{extra}"
def _run_command_reason(args: Mapping[str, object]) -> str | None:
argv = _as_argv(args)
def _run_argv_reason(args: Mapping[str, object]) -> str | None:
argv = _as_argv(args, key="argv")
if argv is None:
return None
return _shell_or_dash_c_reason(argv, via="run_command")
return _shell_or_dash_c_reason(argv, via="run_argv")
def _batch_step_argv(item: object) -> list[str] | None:
@@ -262,8 +262,8 @@ def classify_danger(name: str, args: Mapping[str, object]) -> str | None: # noq
return _copy_path_reason(args)
if name == "write_file":
return _write_file_reason(args)
if name == "run_command":
return _run_command_reason(args)
if name == "run_argv":
return _run_argv_reason(args)
if name == "run_command_batch":
return _run_command_batch_reason(args)
if name == "open_pty":
+2 -2
View File
@@ -4,13 +4,13 @@ from .open_pty import open_pty as open_pty
from .pty_terminal import decode_write_data as decode_write_data
from .pty_terminal import sanitize_pty_output_for_tool as sanitize_pty_output_for_tool
from .read_pty import read_pty as read_pty
from .run_command import run_command as run_command
from .run_argv import run_argv as run_argv
from .run_command_batch import run_command_batch as run_command_batch
from .write_pty import write_pty as write_pty
from .write_pty_keys import write_pty_keys as write_pty_keys
PROCESS_TOOLS = [
run_command,
run_argv,
run_command_batch,
open_pty,
read_pty,
+1 -1
View File
@@ -83,7 +83,7 @@ async def execute_argv(
timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
mix_stderr: bool = False,
) -> CommandStepResult:
"""Run one argv command; no shell. Used by run_command and run_command_batch."""
"""Run one argv command; no shell. Used by run_argv and run_command_batch."""
if not command:
msg = "command argv must not be empty"
raise WorkspaceError(msg)
@@ -11,21 +11,21 @@ from .command_exec import (
@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO)
async def run_command(
command: list[str],
async def run_argv(
argv: list[str],
*,
cwd: str = ".",
timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
stdin: str | None = None,
env: dict[str, str] | None = None,
) -> str:
"""Run ``command`` (argv, no shell) under the workspace; capture stdout/stderr."""
shape_error = argv_shape_error(command)
"""Run ``argv`` (argv, no shell) under the workspace; capture stdout/stderr."""
shape_error = argv_shape_error(argv)
if shape_error is not None:
return f"error: `run_command` {shape_error}"
return f"error: `run_argv` {shape_error}"
try:
result = await execute_argv(
command,
argv,
cwd=cwd,
env=env,
stdin=stdin,
+3 -3
View File
@@ -30,10 +30,10 @@ def test_prompt_confirm_tool_default_deny() -> None:
def test_format_tool_confirm_box_multiline() -> None:
reason = "run_command: python3 -c (review code before allow)\nargv:\n python3 -c print(1)\n-c code:\n print(1)"
box = format_tool_confirm_box("run_command", reason)
reason = "run_argv: python3 -c (review code before allow)\nargv:\n python3 -c print(1)\n-c code:\n print(1)"
box = format_tool_confirm_box("run_argv", reason)
assert "┌" in box and "└" in box
assert "confirm · tool 'run_command'" in box
assert "confirm · tool 'run_argv'" in box
assert "python3 -c" in box
assert "print(1)" in box
# Distinct lines, not one jammed row
+7 -7
View File
@@ -43,9 +43,9 @@ def test_classify_write_overwrite_only(tmp_path: Path) -> None:
def test_classify_safe_tools() -> None:
assert classify_danger("read_file", {"path": "a"}) is None
assert classify_danger("listdir", {"path": "."}) is None
assert classify_danger("run_command", {"command": ["echo", "hi"]}) is None
assert classify_danger("run_argv", {"argv": ["echo", "hi"]}) is None
assert classify_danger("open_pty", {"command": ["true"]}) is None
assert classify_danger("run_command", {"command": ["ls", "-la"]}) is None
assert classify_danger("run_argv", {"argv": ["ls", "-la"]}) is None
def test_classify_run_command_batch_risky() -> None:
@@ -71,18 +71,18 @@ def test_classify_run_command_batch_risky() -> None:
def test_classify_shell_and_dash_c() -> None:
r = classify_danger("run_command", {"command": ["bash", "-c", "rm -rf /"]})
r = classify_danger("run_argv", {"argv": ["bash", "-c", "rm -rf /"]})
assert r is not None
assert "bash -c" in r
assert "rm -rf" in r
r2 = classify_danger("run_command", {"command": ["python3", "-c", "print(1)"]})
r2 = classify_danger("run_argv", {"argv": ["python3", "-c", "print(1)"]})
assert r2 is not None
assert "python" in r2
assert "-c" in r2
assert "print(1)" in r2
r_py = classify_danger("run_command", {"command": ["python", "-c", "import os"]})
r_py = classify_danger("run_argv", {"argv": ["python", "-c", "import os"]})
assert r_py is not None and "python -c" in r_py
r3 = classify_danger("open_pty", {"command": ["bash"]})
@@ -118,8 +118,8 @@ async def test_confirm_deny_with_comment() -> None:
def test_shell_confirm_formats_command_placeholder() -> None:
script = chr(10).join(["line1", "line2", "line3"])
reason = classify_danger(
"run_command",
{"command": ["bash", "-c", script]},
"run_argv",
{"argv": ["bash", "-c", script]},
)
assert reason is not None
assert "$(command)" in reason
+21 -21
View File
@@ -9,7 +9,7 @@ from plyngent.tools.process import (
close_pty,
open_pty,
read_pty,
run_command,
run_argv,
run_command_batch,
write_pty,
write_pty_keys,
@@ -41,32 +41,32 @@ def _py(code: str) -> list[str]:
return [sys.executable, "-c", code]
async def test_run_command_echo(workspace: object) -> None:
async def test_run_argv_echo(workspace: object) -> None:
del workspace
out = await call_async(run_command, _py("print('hi')"))
out = await call_async(run_argv, _py("print('hi')"))
assert "exit_code=0" in out
assert "hi" in out
async def test_run_command_string_command_error(workspace: object) -> None:
async def test_run_argv_string_command_error(workspace: object) -> None:
del workspace
out = await call_async(run_command, "ls -la")
out = await call_async(run_argv, "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:
async def test_run_argv_stringified_json_array_error(workspace: object) -> None:
del workspace
out = await call_async(run_command, '["ls", "-la"]')
out = await call_async(run_argv, '["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:
async def test_run_argv_non_string_args_error(workspace: object) -> None:
del workspace
out = await call_async(run_command, ["ls", 1])
out = await call_async(run_argv, ["ls", 1])
assert "error:" in out
assert "array of string args" in out
@@ -157,32 +157,32 @@ async def test_run_command_batch_mix_stderr(workspace: object) -> None:
assert "out" in out and "err" in out
async def test_run_command_denied(workspace: object) -> None:
async def test_run_argv_denied(workspace: object) -> None:
del workspace
out = await call_async(run_command, ["rm", "-rf", "."])
out = await call_async(run_argv, ["rm", "-rf", "."])
assert "denied" in out
async def test_run_command_cwd(workspace: object) -> None:
async def test_run_argv_cwd(workspace: object) -> None:
assert isinstance(workspace, Path)
sub = workspace / "sub"
sub.mkdir()
_ = (sub / "f.txt").write_text("z", encoding="utf-8")
out = await call_async(run_command, _py("import os; print('\\n'.join(os.listdir()))"), cwd="sub")
out = await call_async(run_argv, _py("import os; print('\\n'.join(os.listdir()))"), cwd="sub")
assert "f.txt" in out
async def test_run_command_timeout(workspace: object) -> None:
async def test_run_argv_timeout(workspace: object) -> None:
del workspace
out = await call_async(run_command, _py("import time; time.sleep(5)"), timeout_seconds=0.2)
out = await call_async(run_argv, _py("import time; time.sleep(5)"), timeout_seconds=0.2)
assert "timed_out=true" in out
async def test_run_command_timeout_keeps_partial_output(workspace: object) -> None:
async def test_run_argv_timeout_keeps_partial_output(workspace: object) -> None:
del workspace
# Print then sleep past the timeout so communicate has partial stdout after kill.
out = await call_async(
run_command,
run_argv,
_py("import sys, time; sys.stdout.write('partial-out'); sys.stdout.flush(); time.sleep(5)"),
timeout_seconds=0.3,
)
@@ -190,10 +190,10 @@ async def test_run_command_timeout_keeps_partial_output(workspace: object) -> No
assert "partial-out" in out
async def test_run_command_stdin(workspace: object) -> None:
async def test_run_argv_stdin(workspace: object) -> None:
del workspace
out = await call_async(
run_command,
run_argv,
_py("import sys; print(sys.stdin.read(), end='')"),
stdin="hello-stdin\n",
)
@@ -201,10 +201,10 @@ async def test_run_command_stdin(workspace: object) -> None:
assert "hello-stdin" in out
async def test_run_command_env(workspace: object) -> None:
async def test_run_argv_env(workspace: object) -> None:
del workspace
out = await call_async(
run_command,
run_argv,
_py("import os; print(os.environ.get('PLYNGENT_TEST_VAR', ''), end='')"),
env={"PLYNGENT_TEST_VAR": "from-env"},
)