mirror of
https://github.com/MervinPraison/PraisonAI.git
synced 2026-11-07 22:52:58 -10:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
eb73d528a8 |
No files matched your search
@@ -81,17 +81,21 @@ class PraisonAICodeAgent(BaseInstalledAgent):
|
||||
model = self.model_name or "openai/gpt-4o-mini"
|
||||
|
||||
command = (
|
||||
# Private per-run log (mktemp, 0600) instead of a predictable
|
||||
# /tmp/praisonai_code.log path another process could pre-create as a
|
||||
# symlink to redirect/clobber a file writable by the agent user.
|
||||
"log=$(mktemp /tmp/praisonai_code.XXXXXX.log); "
|
||||
f"praisonai code -p --output json {shlex.quote(instruction)} "
|
||||
f"--dangerously-skip-approval "
|
||||
f"--model {shlex.quote(model)} "
|
||||
"> /tmp/praisonai_code.log 2>&1; "
|
||||
'> "$log" 2>&1; '
|
||||
# Capture the real exit status so install/auth/startup failures still
|
||||
# surface instead of being silently masked (Harbor otherwise records a
|
||||
# completed attempt for a crashed agent).
|
||||
"status=$?; "
|
||||
# Echo the envelope so Harbor captures it for post-run accounting.
|
||||
"cp /tmp/praisonai_code.log /tmp/praisonai_code_run.log 2>/dev/null || true; "
|
||||
"cat /tmp/praisonai_code.log; "
|
||||
# Echo the envelope so Harbor captures it for post-run accounting,
|
||||
# then remove the private log.
|
||||
'cat "$log"; rm -f "$log"; '
|
||||
# `praisonai code` normally exits 0 (Harbor grades by task
|
||||
# verification), so a genuine benchmark miss won't fail here — but a
|
||||
# nonzero status means the assistant itself failed to run.
|
||||
@@ -128,7 +132,6 @@ class PraisonAICodeAgent(BaseInstalledAgent):
|
||||
"agent_type": "code-cli",
|
||||
"agent_name": self.name(),
|
||||
"model": self.model_name,
|
||||
"log_path": "/tmp/praisonai_code.log",
|
||||
"session_id": envelope.get("session_id") if envelope else None,
|
||||
"status": envelope.get("status") if envelope else "unknown",
|
||||
}
|
||||
|
||||
@@ -491,47 +491,51 @@ def _run_print_code(
|
||||
except Exception:
|
||||
resolved_session = None
|
||||
|
||||
workspace = os.environ.get("PRAISONAI_WORKSPACE") or os.getcwd()
|
||||
tool_groups = ["basic", "acp", "edit", "search", "lsp"]
|
||||
code_tools = _get_headless_code_tools(
|
||||
groups=tool_groups,
|
||||
workspace=workspace,
|
||||
)
|
||||
|
||||
agent_config = {
|
||||
"name": "CodeAgent",
|
||||
"role": "Code Assistant",
|
||||
"goal": "Help with coding tasks",
|
||||
"instructions": (
|
||||
"You are a terminal coding assistant. Inspect the workspace and "
|
||||
"use the provided tools for coding tasks. Keep all file operations "
|
||||
"inside the configured workspace."
|
||||
),
|
||||
# A minimal output preset keeps the agent from printing its own
|
||||
# decorations so stdout carries only our envelope.
|
||||
"output": "minimal",
|
||||
"tools": code_tools,
|
||||
}
|
||||
if model:
|
||||
agent_config["llm"] = model
|
||||
|
||||
memory_cfg = None
|
||||
if resolved_session:
|
||||
try:
|
||||
from ..state.project_sessions import build_cli_memory_config
|
||||
|
||||
memory_cfg = build_cli_memory_config(
|
||||
session_id=resolved_session, auto_save=resolved_session
|
||||
)
|
||||
except Exception:
|
||||
memory_cfg = None
|
||||
if memory_cfg is not None:
|
||||
agent_config["memory"] = memory_cfg
|
||||
|
||||
status = "ok"
|
||||
result = None
|
||||
error_message = None
|
||||
try:
|
||||
# Resolve the workspace and load the coding tools *inside* the protected
|
||||
# lifecycle: tool loading spins up the ACP/LSP runtime, which can raise.
|
||||
# Keeping it here guarantees a failure still emits the JSON error
|
||||
# envelope and always runs cleanup_runtime() via the finally block,
|
||||
# instead of aborting before either.
|
||||
workspace = os.environ.get("PRAISONAI_WORKSPACE") or os.getcwd()
|
||||
tool_groups = ["basic", "acp", "edit", "search", "lsp"]
|
||||
code_tools = _get_headless_code_tools(
|
||||
groups=tool_groups,
|
||||
workspace=workspace,
|
||||
)
|
||||
|
||||
agent_config = {
|
||||
"name": "CodeAgent",
|
||||
"role": "Code Assistant",
|
||||
"goal": "Help with coding tasks",
|
||||
"instructions": (
|
||||
"You are a terminal coding assistant. Inspect the workspace and "
|
||||
"use the provided tools for coding tasks. Keep all file "
|
||||
"operations inside the configured workspace."
|
||||
),
|
||||
# A minimal output preset keeps the agent from printing its own
|
||||
# decorations so stdout carries only our envelope.
|
||||
"output": "minimal",
|
||||
"tools": code_tools,
|
||||
}
|
||||
if model:
|
||||
agent_config["llm"] = model
|
||||
|
||||
if resolved_session:
|
||||
try:
|
||||
from ..state.project_sessions import build_cli_memory_config
|
||||
|
||||
memory_cfg = build_cli_memory_config(
|
||||
session_id=resolved_session, auto_save=resolved_session
|
||||
)
|
||||
except Exception:
|
||||
memory_cfg = None
|
||||
if memory_cfg is not None:
|
||||
agent_config["memory"] = memory_cfg
|
||||
|
||||
agent = Agent(**agent_config)
|
||||
if thinking_budget is not None:
|
||||
agent.thinking_budget = thinking_budget
|
||||
|
||||
@@ -189,34 +189,99 @@ def resolve_tool_groups(
|
||||
return tool_names
|
||||
|
||||
|
||||
def _load_basic_tools() -> Dict[str, Callable]:
|
||||
"""Lazy load basic tools from praisonaiagents."""
|
||||
def _bind_basic_file_shell_tools(config: ToolConfig) -> Optional[Dict[str, Callable]]:
|
||||
"""Bind the basic file/shell tools to the configured workspace.
|
||||
|
||||
The module-level ``read_file``/``write_file``/``list_files`` helpers and the
|
||||
raw ``execute_command`` are unbound: their fallback path only rejects ``..``
|
||||
traversal and still accepts absolute paths (e.g. ``/etc/hosts``) resolved
|
||||
against the *process* working directory, and ``execute_command`` honours any
|
||||
caller-supplied ``cwd``. Under auto-approval (``--dangerously-skip-approval``)
|
||||
that lets the agent read/write/execute outside ``--workspace``.
|
||||
|
||||
Mirror the fail-closed containment already used by the ``edit`` group: bind
|
||||
``FileTools`` to a ``Workspace`` so relative and absolute paths resolve
|
||||
*inside* the workspace, and force ``execute_command`` to run with its ``cwd``
|
||||
pinned to the workspace root. Returns ``None`` when containment is
|
||||
unavailable so the caller can fall back to the unbound tools (manual
|
||||
approval still guards them).
|
||||
"""
|
||||
try:
|
||||
from pathlib import Path
|
||||
from praisonaiagents.workspace import Workspace # type: ignore
|
||||
from praisonaiagents.tools.file_tools import FileTools
|
||||
from praisonaiagents.tools.shell_tools import ShellTools
|
||||
except ImportError as e:
|
||||
logger.debug(f"Workspace-bound basic tools unavailable: {e}")
|
||||
return None
|
||||
|
||||
try:
|
||||
workspace = Workspace(root=Path(config.workspace))
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Basic tools not workspace-bound: containment unavailable: {e}"
|
||||
)
|
||||
return None
|
||||
|
||||
file_tools = FileTools(workspace=workspace)
|
||||
shell_tools = ShellTools()
|
||||
workspace_root = str(Path(config.workspace).resolve())
|
||||
|
||||
bound: Dict[str, Callable] = {
|
||||
"read_file": file_tools.read_file,
|
||||
"write_file": file_tools.write_file,
|
||||
"list_files": file_tools.list_files,
|
||||
}
|
||||
|
||||
raw_execute = shell_tools.execute_command
|
||||
|
||||
def execute_command(command: str, timeout: int = 30, **kwargs):
|
||||
"""Execute a shell command with its working directory pinned to the
|
||||
configured workspace so it cannot escape ``--workspace``."""
|
||||
kwargs.pop("cwd", None)
|
||||
return raw_execute(command, cwd=workspace_root, timeout=timeout, **kwargs)
|
||||
|
||||
execute_command.__name__ = "execute_command"
|
||||
bound["execute_command"] = execute_command
|
||||
return bound
|
||||
|
||||
|
||||
def _load_basic_tools(config: Optional[ToolConfig] = None) -> Dict[str, Callable]:
|
||||
"""Lazy load basic tools from praisonaiagents.
|
||||
|
||||
When ``config`` carries a workspace, the file and shell tools are bound to
|
||||
it (fail-closed) so an auto-approved headless run stays inside ``--workspace``.
|
||||
"""
|
||||
tools = {}
|
||||
|
||||
try:
|
||||
from praisonaiagents.tools import read_file
|
||||
tools["read_file"] = read_file
|
||||
except ImportError:
|
||||
logger.debug("read_file not available")
|
||||
|
||||
try:
|
||||
from praisonaiagents.tools import write_file
|
||||
tools["write_file"] = write_file
|
||||
except ImportError:
|
||||
logger.debug("write_file not available")
|
||||
|
||||
try:
|
||||
from praisonaiagents.tools import list_files
|
||||
tools["list_files"] = list_files
|
||||
except ImportError:
|
||||
logger.debug("list_files not available")
|
||||
|
||||
try:
|
||||
from praisonaiagents.tools import execute_command
|
||||
tools["execute_command"] = execute_command
|
||||
except ImportError:
|
||||
logger.debug("execute_command not available")
|
||||
|
||||
|
||||
bound = _bind_basic_file_shell_tools(config) if config is not None else None
|
||||
if bound is not None:
|
||||
tools.update(bound)
|
||||
else:
|
||||
try:
|
||||
from praisonaiagents.tools import read_file
|
||||
tools["read_file"] = read_file
|
||||
except ImportError:
|
||||
logger.debug("read_file not available")
|
||||
|
||||
try:
|
||||
from praisonaiagents.tools import write_file
|
||||
tools["write_file"] = write_file
|
||||
except ImportError:
|
||||
logger.debug("write_file not available")
|
||||
|
||||
try:
|
||||
from praisonaiagents.tools import list_files
|
||||
tools["list_files"] = list_files
|
||||
except ImportError:
|
||||
logger.debug("list_files not available")
|
||||
|
||||
try:
|
||||
from praisonaiagents.tools import execute_command
|
||||
tools["execute_command"] = execute_command
|
||||
except ImportError:
|
||||
logger.debug("execute_command not available")
|
||||
|
||||
try:
|
||||
from praisonaiagents.tools import internet_search
|
||||
tools["internet_search"] = internet_search
|
||||
@@ -550,9 +615,10 @@ def get_interactive_tools(
|
||||
# Load tools by group
|
||||
all_tools: Dict[str, Callable] = {}
|
||||
|
||||
# Load basic tools (always available, no runtime needed)
|
||||
# Load basic tools (always available, no runtime needed). Pass the config so
|
||||
# the file/shell tools bind to the configured workspace (fail-closed).
|
||||
if config.enable_basic and not (disable and "basic" in disable):
|
||||
basic_tools = _load_basic_tools()
|
||||
basic_tools = _load_basic_tools(config)
|
||||
all_tools.update(basic_tools)
|
||||
|
||||
# Load fast search tools (always available, no runtime needed)
|
||||
|
||||
@@ -154,6 +154,9 @@ def test_code_print_real_agent_writes_workspace(tmp_path):
|
||||
"gpt-4o-mini",
|
||||
"--workspace",
|
||||
str(tmp_path),
|
||||
# CliRunner has no interactive approval path, so skip the prompt or
|
||||
# the write tool would hang/deny and never create the probe file.
|
||||
"--dangerously-skip-approval",
|
||||
f"Create {target.name} containing exactly HEADLESS_OK.",
|
||||
],
|
||||
)
|
||||
@@ -164,6 +167,40 @@ def test_code_print_real_agent_writes_workspace(tmp_path):
|
||||
assert target.read_text(encoding="utf-8").strip() == "HEADLESS_OK"
|
||||
|
||||
|
||||
def test_headless_basic_tools_are_workspace_bound(tmp_path):
|
||||
"""The headless `basic` group must stay inside `--workspace`.
|
||||
|
||||
Under `--dangerously-skip-approval` these tools run unguarded, so an
|
||||
absolute path outside the workspace must be rejected rather than written —
|
||||
matching the fail-closed containment used by the `edit` group.
|
||||
"""
|
||||
import os
|
||||
|
||||
os.environ["PRAISON_APPROVAL_MODE"] = "auto"
|
||||
os.environ["PRAISONAI_TOOL_SAFETY"] = "off"
|
||||
from praisonaiagents.approval import add_yaml_approved_tools
|
||||
|
||||
add_yaml_approved_tools(["write_file", "read_file", "list_files"])
|
||||
|
||||
from praisonai_code.cli.features.interactive_tools import (
|
||||
ToolConfig,
|
||||
_load_basic_tools,
|
||||
)
|
||||
|
||||
tools = _load_basic_tools(ToolConfig(workspace=str(tmp_path)))
|
||||
write_file = tools["write_file"]
|
||||
|
||||
assert write_file("inside.txt", "hello") is True
|
||||
assert (tmp_path / "inside.txt").read_text() == "hello"
|
||||
|
||||
escape = tmp_path.parent / "escape_probe.txt"
|
||||
if escape.exists():
|
||||
escape.unlink()
|
||||
assert write_file(str(escape), "escaped") is False
|
||||
assert not escape.exists()
|
||||
assert tools["execute_command"].__name__ == "execute_command"
|
||||
|
||||
|
||||
def test_code_print_text_mode_clean_stdout(monkeypatch):
|
||||
"""`-p --output text` prints just the result text, no envelope or decorations."""
|
||||
import praisonaiagents
|
||||
|
||||
Reference in new issue
Block a user