| 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) |
|
|
