Spaces:
Sleeping
Sleeping
File size: 4,096 Bytes
c14504c | 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 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """
Safe code execution engine.
Runs submitted code in a subprocess with timeout.
Writes code to a temp directory, generates a test harness,
and parses JSON results from stdout.
"""
import json
import subprocess
import tempfile
import textwrap
from pathlib import Path
from typing import Tuple
def run_code_safely(
submitted_code: str,
test_code: str,
timeout: int = 10,
) -> Tuple[list[dict], str, str]:
"""
Execute submitted code against test cases in an isolated subprocess.
Args:
submitted_code: The Python code the agent submitted as a fix.
test_code: Python snippet that populates a `results` list with test dicts.
timeout: Max seconds before killing the subprocess.
Returns:
(test_results, stdout_extra, stderr) where test_results is a list of
{"test_name", "passed", "expected", "actual", "error"} dicts.
"""
with tempfile.TemporaryDirectory() as tmpdir:
solution_path = Path(tmpdir) / "solution.py"
harness_path = Path(tmpdir) / "harness.py"
# Write the submitted code as a module
solution_path.write_text(submitted_code, encoding="utf-8")
# Build the test harness
harness = textwrap.dedent(f"""\
import sys, json, traceback
sys.path.insert(0, r"{tmpdir}")
results = []
# Execute the submitted solution in this namespace
try:
exec(open(r"{solution_path}", encoding="utf-8").read())
except Exception as e:
# If the solution itself fails to load, all tests fail
print(json.dumps([{{"test_name": "load", "passed": False,
"expected": "code loads", "actual": "",
"error": traceback.format_exc()}}]))
sys.exit(0)
# Run the test code (populates `results`)
{textwrap.indent(test_code, " ").strip()}
print(json.dumps(results))
""")
harness_path.write_text(harness, encoding="utf-8")
try:
proc = subprocess.run(
["python", str(harness_path)],
capture_output=True,
text=True,
timeout=timeout,
cwd=tmpdir,
)
stdout = proc.stdout.strip()
stderr = proc.stderr.strip()
# Parse test results from last line of stdout (the JSON array)
if stdout:
# The JSON array should be the last line
lines = stdout.split("\n")
json_line = lines[-1]
extra_output = "\n".join(lines[:-1]) if len(lines) > 1 else ""
try:
test_results = json.loads(json_line)
return test_results, extra_output, stderr
except json.JSONDecodeError:
return [
{
"test_name": "parse_error",
"passed": False,
"expected": "valid JSON output",
"actual": stdout[:200],
"error": "Could not parse test results from subprocess output",
}
], "", stderr
# No stdout at all — likely a crash
return [
{
"test_name": "execution_error",
"passed": False,
"expected": "code runs",
"actual": "",
"error": stderr[:500] if stderr else "No output produced",
}
], "", stderr
except subprocess.TimeoutExpired:
return [
{
"test_name": "timeout",
"passed": False,
"expected": f"completes within {timeout}s",
"actual": "timed out",
"error": f"Code execution exceeded {timeout} second timeout",
}
], "", "Execution timed out"
|