Upload fsi_sandbox.py with huggingface_hub

fsi_sandbox.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+import os, sys, subprocess, tempfile, json, traceback, time, resource
+
+class ExecutionSandbox:
+    def __init__(self, engine=None, timeout=5, max_memory_mb=256):
+        self.engine = engine
+        self.timeout = timeout
+        self.max_memory = max_memory_mb * 1024 * 1024
+        self.execution_history = []
+
+    def _run_in_subprocess(self, code, test_input=None):
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
+            f.write(code)
+            tmp = f.name
+        try:
+            script = f"""
+import sys, json, traceback, builtins
+import resource
+resource.setrlimit(resource.RLIMIT_CPU, ({self.timeout}, {self.timeout}))
+resource.setrlimit(resource.RLIMIT_AS, ({self.max_memory}, {self.max_memory}))
+_original = builtins.__import__
+def _restricted(name, *args, **kwargs):
+    blocked = ['os.system', 'subprocess', 'socket', 'urllib', 'http.client', 'ftplib']
+    if any(b in name for b in blocked):
+        raise ImportError(f"Blocked: {{name}}")
+    return _original(name, *args, **kwargs)
+builtins.__import__ = _restricted
+result = {{"stdout": "", "stderr": "", "output": None, "error": None, "success": False}}
+old_out, old_err = sys.stdout, sys.stderr
+try:
+    from io import StringIO
+    sys.stdout, sys.stderr = StringIO(), StringIO()
+    ns = {{"__builtins__": builtins}}
+    with open("{tmp}") as f:
+        exec(f.read(), ns)
+    for name, obj in ns.items():
+        if callable(obj) and name in ["main", "run", "solve", "test"]:
+            try:
+                result["output"] = obj("{test_input or ''}") if "{test_input or ''}" else obj()
+            except Exception as e:
+                result["error"] = str(e)
+    result["stdout"] = sys.stdout.getvalue()
+    result["stderr"] = sys.stderr.getvalue()
+    result["success"] = result["error"] is None
+except Exception as e:
+    result["error"] = traceback.format_exc()
+    result["success"] = False
+finally:
+    sys.stdout, sys.stderr = old_out, old_err
+print(json.dumps(result))
+"""
+            proc = subprocess.run([sys.executable, '-c', script], capture_output=True, text=True, timeout=self.timeout+2, cwd='/tmp')
+            out = proc.stdout.strip()
+            if out:
+                try:
+                    return json.loads(out.split('\n')[-1])
+                except:
+                    return {"stdout": proc.stdout, "stderr": proc.stderr, "error": proc.stderr or "Failed", "success": False}
+            return {"stdout": "", "stderr": proc.stderr, "error": proc.stderr or "No output", "success": False}
+        except subprocess.TimeoutExpired:
+            return {"error": f"Timeout after {self.timeout}s", "success": False}
+        except Exception as e:
+            return {"error": str(e), "success": False}
+        finally:
+            try: os.unlink(tmp)
+            except: pass
+
+    def execute_and_verify(self, code, expected_output=None, test_input=None):
+        result = self._run_in_subprocess(code, test_input)
+        if expected_output is not None:
+            actual = result.get("output")
+            result["verified"] = True
+            result["match"] = actual == expected_output
+            result["expected"] = expected_output
+            result["actual"] = actual
+        else:
+            result["verified"] = False
+        if self.engine and hasattr(self.engine, '_update_pheromone'):
+            success = result.get("success") and result.get("match", True)
+            for idx in range(1000):
+                self.engine._update_pheromone(idx, success)
+        self.execution_history.append({"success": result.get("success"), "match": result.get("match")})
+        return result
+
+    def auto_fix(self, code, error, max_attempts=3):
+        if not self.engine:
+            return code, {"fixed": False}
+        current = code
+        for attempt in range(max_attempts):
+            prompt = f"Fix this Python code. Error: {error}\n\n```python\n{current}\n```\n\nProvide ONLY the corrected code."
+            try:
+                fixed = self.engine.generate(prompt, max_tokens=800)
+                if "```python" in fixed:
+                    fixed = fixed.split("```python")[1].split("```")[0]
+                elif "```" in fixed:
+                    fixed = fixed.split("```")[1].split("```")[0]
+                result = self._run_in_subprocess(fixed)
+                if result.get("success"):
+                    return fixed, {"fixed": True, "attempts": attempt + 1}
+                current = fixed
+                error = result.get("error", "Unknown")
+            except Exception as e:
+                return current, {"fixed": False, "reason": str(e)}
+        return current, {"fixed": False, "reason": f"Failed after {max_attempts}"}
+
+    def get_stats(self):
+        total = len(self.execution_history)
+        if total == 0:
+            return {"total": 0, "success_rate": 0}
+        successes = sum(1 for h in self.execution_history if h.get("success"))
+        matches = sum(1 for h in self.execution_history if h.get("match"))
+        return {"total": total, "successes": successes, "success_rate": successes/total*100, "matches": matches}
+
+if __name__ == "__main__":
+    s = ExecutionSandbox()
+    r1 = s.execute_and_verify("def add(a,b): return a+b\nprint(add(2,3))")
+    print(f"Test 1: success={r1['success']}, stdout={r1.get('stdout','')[:50]}")
+    r2 = s.execute_and_verify("def div(a,b): return a/b\ndiv(10,0)")
+    print(f"Test 2: success={r2['success']}, error={r2.get('error','')[:50]}")
+    print(f"Stats: {s.get_stats()}")