mirror of
https://github.com/tiennm99/serena.git
synced 2026-10-13 20:15:20 +00:00
Add shell facade (shell) for shell command execution
Introduce ShellApi with the single editing operation execute_shell_command, which returns a ShellCommandOutput exposing stdout, stderr, return code and working directory to code and rendering as JSON as before. Keeping shell command execution in its own facade makes it a natural unit for exclusion. The shell command tool delegates to the API via ShellApiMixin; the shell facade is always part of the REPL.
This commit is contained in:
1 parent
5f129628d9
commit
a46a465773
4 files changed
+150
-18
No files matched your search
@@ -52,6 +52,7 @@ from serena.repl.api.fs_api import FsApi
|
||||
from serena.repl.api.jb_api import JetBrainsApi
|
||||
from serena.repl.api.lsp_api import LspApi
|
||||
from serena.repl.api.mem_api import MemoryApi
|
||||
from serena.repl.api.shell_api import ShellApi
|
||||
from serena.repl.facade import ApiScope, Facade
|
||||
from serena.repl.repl import SerenaRepl
|
||||
from serena.task_executor import TaskExecutor
|
||||
@@ -1190,6 +1191,7 @@ class SerenaAgent:
|
||||
Facade.from_api(FsApi(self), api_scope),
|
||||
Facade.from_api(EditApi(self), api_scope),
|
||||
Facade.from_api(MemoryApi(self), api_scope),
|
||||
Facade.from_api(ShellApi(self), api_scope),
|
||||
]
|
||||
if self._language_backend.is_lsp():
|
||||
facades.append(Facade.from_api(LspApi(self), api_scope))
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""
|
||||
The implementation of shell command execution.
|
||||
"""
|
||||
|
||||
import os.path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from serena.util.shell import ShellCommandResult, execute_shell_command
|
||||
|
||||
from ..facade import FacadeApi, facade_method
|
||||
from ..representable import Renderer, RepresentableViaRenderer
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from serena.agent import SerenaAgent
|
||||
|
||||
|
||||
class ShellCommandOutput(RepresentableViaRenderer):
|
||||
"""
|
||||
The outcome of a shell command: `stdout`, `stderr` (None if not captured), `return_code` and `cwd`.
|
||||
"""
|
||||
|
||||
def __init__(self, result: ShellCommandResult, renderer: "ShellCommandOutputRenderer"):
|
||||
"""
|
||||
:param result: the result of the command execution
|
||||
:param renderer: the renderer to use for representing the output
|
||||
"""
|
||||
super().__init__(renderer)
|
||||
self.result_ = result
|
||||
|
||||
@property
|
||||
def stdout(self) -> str:
|
||||
return self.result_.stdout
|
||||
|
||||
@property
|
||||
def stderr(self) -> str | None:
|
||||
return self.result_.stderr
|
||||
|
||||
@property
|
||||
def return_code(self) -> int:
|
||||
return self.result_.return_code
|
||||
|
||||
@property
|
||||
def cwd(self) -> str:
|
||||
return self.result_.cwd
|
||||
|
||||
|
||||
class ShellCommandOutputRenderer(Renderer[ShellCommandOutput]):
|
||||
def render(self, obj: ShellCommandOutput) -> str:
|
||||
return self._limit_length(obj.result_.model_dump_json())
|
||||
|
||||
|
||||
class ShellApi(FacadeApi):
|
||||
def __init__(self, agent: "SerenaAgent") -> None:
|
||||
super().__init__(agent, name="shell", description="execution of shell commands")
|
||||
|
||||
@facade_method(can_edit=True)
|
||||
def execute_shell_command(
|
||||
self, command: str, cwd: str | None = None, capture_stderr: bool = True, max_answer_chars: int = -1
|
||||
) -> ShellCommandOutput:
|
||||
"""
|
||||
Executes a shell command and returns its output. If there is a memory about suggested commands, read that first.
|
||||
Never execute unsafe shell commands!
|
||||
IMPORTANT: Do not use this to start
|
||||
* long-running processes (e.g. servers) that are not intended to terminate quickly,
|
||||
* processes that require user interaction.
|
||||
|
||||
:param command: the shell command to execute
|
||||
:param cwd: the working directory to execute the command in (absolute, or relative to the project root).
|
||||
If None, the project root will be used.
|
||||
:param capture_stderr: whether to capture and return stderr output
|
||||
:param max_answer_chars: if the rendered output is longer than this number of characters, no content is rendered;
|
||||
-1 for the configured default. Don't adjust unless there is no other way to get the content required for the task.
|
||||
:return: the output (object with properties stdout, stderr, return_code and cwd)
|
||||
"""
|
||||
project_root = self._get_project().project_root
|
||||
if cwd is None:
|
||||
_cwd = project_root
|
||||
elif os.path.isabs(cwd):
|
||||
_cwd = cwd
|
||||
else:
|
||||
_cwd = os.path.join(project_root, cwd)
|
||||
if not os.path.isdir(_cwd):
|
||||
raise FileNotFoundError(
|
||||
f"Specified a relative working directory ({cwd}), but the resulting path is not a directory: {_cwd}"
|
||||
)
|
||||
|
||||
result = execute_shell_command(command, cwd=_cwd, capture_stderr=capture_stderr)
|
||||
return ShellCommandOutput(result, ShellCommandOutputRenderer(self._agent, max_answer_chars))
|
||||
@@ -3,13 +3,23 @@ Tools supporting the execution of (external) commands
|
||||
"""
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
import os.path
|
||||
from typing import cast
|
||||
|
||||
from serena.repl.api.shell_api import ShellApi
|
||||
from serena.tools import Tool, ToolMarkerCanEdit
|
||||
from serena.util.shell import execute_shell_command
|
||||
|
||||
|
||||
class ExecuteShellCommandTool(Tool, ToolMarkerCanEdit):
|
||||
class ShellApiMixin:
|
||||
"""
|
||||
Mixin for tools which delegate to the shell API
|
||||
"""
|
||||
|
||||
def _api(self) -> ShellApi:
|
||||
tool = cast(Tool, cast(object, self))
|
||||
return ShellApi(tool.agent)
|
||||
|
||||
|
||||
class ExecuteShellCommandTool(Tool, ToolMarkerCanEdit, ShellApiMixin):
|
||||
"""
|
||||
Executes a shell command.
|
||||
"""
|
||||
@@ -36,18 +46,4 @@ class ExecuteShellCommandTool(Tool, ToolMarkerCanEdit):
|
||||
required for the task.
|
||||
:return: a JSON object containing the command's stdout and optionally stderr output
|
||||
"""
|
||||
if cwd is None:
|
||||
_cwd = self.get_project_root()
|
||||
else:
|
||||
if os.path.isabs(cwd):
|
||||
_cwd = cwd
|
||||
else:
|
||||
_cwd = os.path.join(self.get_project_root(), cwd)
|
||||
if not os.path.isdir(_cwd):
|
||||
raise FileNotFoundError(
|
||||
f"Specified a relative working directory ({cwd}), but the resulting path is not a directory: {_cwd}"
|
||||
)
|
||||
|
||||
result = execute_shell_command(command, cwd=_cwd, capture_stderr=capture_stderr)
|
||||
result = result.model_dump_json()
|
||||
return self._limit_length(result, max_answer_chars)
|
||||
return self._api().execute_shell_command(command, cwd, capture_stderr, max_answer_chars).represent()
|
||||
@@ -0,0 +1,45 @@
|
||||
"""
|
||||
Tests for the shell facade API.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from serena.config.serena_config import SerenaConfig
|
||||
from serena.project import Project
|
||||
from serena.repl.api.shell_api import ShellApi
|
||||
from serena.repl.facade import ApiScope, Facade
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api(tmp_path: Path) -> ShellApi:
|
||||
(tmp_path / "sub").mkdir()
|
||||
project = Project.load(str(tmp_path), serena_config=SerenaConfig(gui_log_window=False, web_dashboard=False))
|
||||
agent = MagicMock()
|
||||
agent.get_active_project_or_raise.return_value = project
|
||||
agent.serena_config.default_max_tool_answer_chars = 10000
|
||||
return ShellApi(agent)
|
||||
|
||||
|
||||
def test_facade_exposes_shell_command_as_editing_operation(api: ShellApi) -> None:
|
||||
facade = Facade.from_api(api, ApiScope())
|
||||
assert facade.name == "shell"
|
||||
assert facade.enabled_method_names == ["execute_shell_command"]
|
||||
assert facade.get_method("execute_shell_command").info.can_edit
|
||||
|
||||
|
||||
def test_execute_shell_command(api: ShellApi, tmp_path: Path) -> None:
|
||||
print_cwd = "cd" if sys.platform == "win32" else "pwd"
|
||||
|
||||
output = api.execute_shell_command(f"{print_cwd}")
|
||||
assert output.return_code == 0
|
||||
assert Path(output.stdout.strip()).resolve() == tmp_path.resolve()
|
||||
assert '"stdout"' in output.represent() and '"return_code"' in output.represent()
|
||||
|
||||
# a relative working directory is resolved against the project root and must exist
|
||||
assert Path(api.execute_shell_command(print_cwd, cwd="sub").stdout.strip()).resolve() == (tmp_path / "sub").resolve()
|
||||
with pytest.raises(FileNotFoundError):
|
||||
api.execute_shell_command(print_cwd, cwd="missing")
|
||||
Reference in new issue
Block a user