File size: 1,429 Bytes
96e6518 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 | from __future__ import annotations
import subprocess
import sys
from pathlib import Path
class WorkspaceExecutor:
"""Executes model-requested actions inside the task workspace."""
def __init__(self, workspace: Path, timeout_s: int = 120) -> None:
self.workspace = workspace.resolve()
self.timeout_s = timeout_s
self.workspace.mkdir(parents=True, exist_ok=True)
def run_python(self, code: str) -> str:
return self._run([sys.executable, "-c", code])
def run_shell(self, command: str) -> str:
if any(token in command for token in ["../", "rm -rf /", "sudo "]):
return "Blocked potentially unsafe shell command."
return self._run(["bash", "-lc", command])
def _run(self, argv: list[str]) -> str:
try:
proc = subprocess.run(
argv,
cwd=self.workspace,
text=True,
capture_output=True,
timeout=self.timeout_s,
check=False,
)
except subprocess.TimeoutExpired:
return f"Command timed out after {self.timeout_s}s."
stdout = proc.stdout.strip()
stderr = proc.stderr.strip()
parts = [f"exit_code={proc.returncode}"]
if stdout:
parts.append(f"stdout:\n{stdout}")
if stderr:
parts.append(f"stderr:\n{stderr}")
return "\n\n".join(parts)
|