Spaces:
Sleeping
Sleeping
| """ | |
| 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" | |