46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
"""Local sandbox: NO isolation, dev convenience only.
|
|
|
|
Runs code in a subprocess on the host with a timeout. Fine for quick
|
|
iteration on harnesses; never use for untrusted model output in shared
|
|
environments — switch the recipe to sandbox='docker' for that.
|
|
"""
|
|
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Dict, Optional
|
|
|
|
from .base import ExecResult, Sandbox, register_sandbox
|
|
|
|
|
|
@register_sandbox('local')
|
|
class LocalSandbox(Sandbox):
|
|
name = 'local'
|
|
|
|
def exec(
|
|
self,
|
|
files: Dict[str, str],
|
|
entry: str = 'main.py',
|
|
mounts: Optional[Dict[str, str]] = None,
|
|
timeout_s: int = 60,
|
|
image: str = '',
|
|
) -> ExecResult:
|
|
with tempfile.TemporaryDirectory(prefix='eh-local-') as td:
|
|
work = Path(td)
|
|
for fname, content in (files or {}).items():
|
|
dest = work / fname
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_text(content, encoding='utf-8')
|
|
t0 = time.time()
|
|
try:
|
|
proc = subprocess.run(
|
|
['python', str(work / entry)], capture_output=True, text=True,
|
|
timeout=timeout_s, cwd=work,
|
|
)
|
|
return ExecResult(exit_code=proc.returncode, stdout=proc.stdout,
|
|
stderr=proc.stderr, duration_s=round(time.time() - t0, 2))
|
|
except subprocess.TimeoutExpired:
|
|
return ExecResult(exit_code=-1, timed_out=True, duration_s=timeout_s,
|
|
error=f'timeout after {timeout_s}s')
|