#!/usr/bin/env python3 """ Python Script Engine — Write, Validate, Run Code Safely ======================================================== Provides: 1. Code generation with GLM-guided structure 2. AST parsing for structural validation 3. Sandboxed execution with timeout/memory limits 4. NRCI scoring for code coherence 5. CRG grounding for concept verification 6. Three Column Thinking: Code + Logic + Verification This replaces the screenplay engine with real script writing. """ import ast, sys, os, re, time, hashlib, traceback, signal, io from typing import List, Dict, Tuple, Optional, Any, Set from dataclasses import dataclass, field from contextlib import contextmanager from fractions import Fraction # ══════════════════════════════════════════════════════════════════════════════ # CODE PARSER — AST Analysis # ══════════════════════════════════════════════════════════════════════════════ @dataclass class CodeAnalysis: """Structural analysis of Python code.""" valid_syntax: bool error_message: str = "" # Structure n_functions: int = 0 n_classes: int = 0 n_imports: int = 0 n_variables: int = 0 n_loops: int = 0 n_conditionals: int = 0 n_try_except: int = 0 n_list_comps: int = 0 n_decorators: int = 0 # Complexity cyclomatic_complexity: int = 0 max_depth: int = 0 total_lines: int = 0 blank_lines: int = 0 comment_lines: int = 0 # Names function_names: List[str] = field(default_factory=list) class_names: List[str] = field(default_factory=list) import_names: List[str] = field(default_factory=list) variable_names: List[str] = field(default_factory=list) # Quality signals has_docstrings: bool = False has_type_hints: bool = False has_error_handling: bool = False has_main_guard: bool = False has_tests: bool = False # Coherence nrci_score: float = 0.0 quality_verdict: str = "" class CodeParser: """Parses Python code into structural analysis.""" def analyze(self, code: str) -> CodeAnalysis: """Full structural analysis of Python code.""" result = CodeAnalysis(valid_syntax=True) result.total_lines = len(code.split('\n')) result.blank_lines = sum(1 for l in code.split('\n') if not l.strip()) result.comment_lines = sum(1 for l in code.split('\n') if l.strip().startswith('#')) # Check for main guard result.has_main_guard = 'if __name__' in code or "if __name__=='__main__'" in code # Check for tests result.has_tests = bool(re.search(r'def test_|class Test|import unittest|import pytest', code)) try: tree = ast.parse(code) except SyntaxError as e: result.valid_syntax = False result.error_message = f"SyntaxError at line {e.lineno}: {e.msg}" return result # Walk AST self._walk(tree, result, depth=0) # Compute cyclomatic complexity result.cyclomatic_complexity = 1 + result.n_conditionals + result.n_loops + result.n_try_except # Compute NRCI score result.nrci_score = self._compute_nrci(result) # Quality verdict result.quality_verdict = self._verdict(result) return result def _walk(self, node, result: CodeAnalysis, depth: int): """Recursively walk AST.""" result.max_depth = max(result.max_depth, depth) # Functions if isinstance(node, ast.FunctionDef) or isinstance(node, ast.AsyncFunctionDef): result.n_functions += 1 result.function_names.append(node.name) # Check for docstring if (node.body and isinstance(node.body[0], ast.Expr) and isinstance(node.body[0].value, (ast.Constant, ast.Str))): result.has_docstrings = True # Check for type hints if node.returns: result.has_type_hints = True # Check for test functions if node.name.startswith('test_'): result.has_tests = True # Classes elif isinstance(node, ast.ClassDef): result.n_classes += 1 result.class_names.append(node.name) # Imports elif isinstance(node, (ast.Import, ast.ImportFrom)): result.n_imports += 1 if isinstance(node, ast.Import): for alias in node.names: result.import_names.append(alias.name) else: result.import_names.append(node.module or "") # Assignments elif isinstance(node, ast.Assign): result.n_variables += len(node.targets) for target in node.targets: if isinstance(target, ast.Name): result.variable_names.append(target.id) # Loops elif isinstance(node, (ast.For, ast.While, ast.AsyncFor)): result.n_loops += 1 # Conditionals elif isinstance(node, (ast.If, ast.IfExp)): result.n_conditionals += 1 # Try/except elif isinstance(node, ast.Try): result.n_try_except += 1 result.has_error_handling = True # List comprehensions elif isinstance(node, (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)): result.n_list_comps += 1 # Decorators if hasattr(node, 'decorator_list'): result.n_decorators += len(node.decorator_list) # Recurse for child in ast.iter_child_nodes(node): self._walk(child, result, depth + 1) def _compute_nrci(self, analysis: CodeAnalysis) -> float: """Compute NRCI-like coherence score for code. Based on structural balance: - Functions should have docstrings - Error handling should be present - Complexity should be moderate (not too simple, not too complex) - Code should have tests """ score = 0.5 # base # Docstrings (+0.1) if analysis.has_docstrings: score += 0.1 # Type hints (+0.05) if analysis.has_type_hints: score += 0.05 # Error handling (+0.1) if analysis.has_error_handling: score += 0.1 # Main guard (+0.05) if analysis.has_main_guard: score += 0.05 # Tests (+0.1) if analysis.has_tests: score += 0.1 # Complexity penalty (too simple or too complex) if analysis.cyclomatic_complexity <= 1: score -= 0.05 # too simple elif analysis.cyclomatic_complexity > 20: score -= 0.15 # too complex elif analysis.cyclomatic_complexity > 10: score -= 0.05 # getting complex # Comment ratio if analysis.total_lines > 0: comment_ratio = analysis.comment_lines / analysis.total_lines if 0.05 <= comment_ratio <= 0.3: score += 0.05 # good comment ratio elif comment_ratio > 0.5: score -= 0.05 # over-commented return max(0.0, min(1.0, score)) def _verdict(self, analysis: CodeAnalysis) -> str: """Quality verdict based on analysis.""" score = analysis.nrci_score issues = [] if not analysis.has_docstrings: issues.append("no docstrings") if not analysis.has_error_handling and analysis.n_functions > 0: issues.append("no error handling") if analysis.cyclomatic_complexity > 15: issues.append(f"high complexity ({analysis.cyclomatic_complexity})") if not analysis.has_tests and analysis.n_functions > 2: issues.append("no tests") if analysis.max_depth > 6: issues.append(f"deep nesting ({analysis.max_depth})") if score >= 0.75: return "EXCELLENT" + (f" ({', '.join(issues)})" if issues else "") elif score >= 0.60: return "GOOD" + (f" — {', '.join(issues)}" if issues else "") elif score >= 0.45: return "FAIR" + (f" — {', '.join(issues)}" if issues else "") else: return "NEEDS_WORK" + (f" — {', '.join(issues)}" if issues else "") # ══════════════════════════════════════════════════════════════════════════════ # SANDBOXED EXECUTOR # ══════════════════════════════════════════════════════════════════════════════ @dataclass class ExecutionResult: """Result of sandboxed code execution.""" success: bool stdout: str = "" stderr: str = "" return_value: Any = None exception: str = "" execution_time_ms: float = 0.0 timeout: bool = False memory_exceeded: bool = False class SandboxExecutor: """Executes Python code in a restricted sandbox. Restrictions: - No file system access (except /tmp) - No network access - No subprocess/os.system - No import of dangerous modules - Timeout enforcement - Output capture """ # Safe modules that are always allowed SAFE_MODULES = { 'math', 'json', 're', 'hashlib', 'random', 'itertools', 'collections', 'functools', 'fractions', 'decimal', 'string', 'textwrap', 'typing', 'dataclasses', 'copy', 'operator', 'bisect', 'heapq', 'array', 'datetime', 'time', 'calendar', 'statistics', 'numbers', 'cmath', } # Blocked modules BLOCKED_MODULES = { 'os', 'sys', 'subprocess', 'shutil', 'socket', 'http', 'urllib', 'requests', 'ftplib', 'smtplib', 'telnetlib', 'xmlrpc', 'ctypes', 'importlib', 'compileall', 'py_compile', 'multiprocessing', 'threading', 'signal', 'shelve', 'pickle', 'marshal', 'dbm', 'webbrowser', 'cgi', 'wsgiref', } # Blocked builtins BLOCKED_BUILTINS = { 'exec', 'eval', 'compile', 'breakpoint', 'exit', 'quit', } def __init__(self, timeout_seconds: float = 5.0, allow_io: bool = False, allow_imports: Optional[Set[str]] = None): self.timeout = timeout_seconds self.allow_io = allow_io self.allow_imports = allow_imports or set() def execute(self, code: str, context: Optional[Dict] = None) -> ExecutionResult: """Execute code in sandbox. Args: code: Python code to execute context: Optional variables to inject into scope Returns: ExecutionResult with stdout, stderr, return value, etc. """ t0 = time.perf_counter() # Pre-check: block dangerous patterns block_reason = self._check_safety(code) if block_reason: return ExecutionResult( success=False, exception=f"Blocked: {block_reason}", execution_time_ms=(time.perf_counter() - t0) * 1000, ) # Capture stdout/stderr old_stdout = sys.stdout old_stderr = sys.stderr captured_stdout = io.StringIO() captured_stderr = io.StringIO() # Build restricted globals safe_globals = self._build_safe_globals(context) timeout_hit = False try: sys.stdout = captured_stdout sys.stderr = captured_stderr # Compile and execute with timeout compiled = compile(code, '', 'exec') # Execute with alarm-based timeout def timeout_handler(signum, frame): nonlocal timeout_hit timeout_hit = True raise TimeoutError(f"Execution exceeded {self.timeout}s") # Set alarm (Unix only) if hasattr(signal, 'SIGALRM'): old_handler = signal.signal(signal.SIGALRM, timeout_handler) signal.setitimer(signal.ITIMER_REAL, self.timeout) try: exec(compiled, safe_globals) finally: if hasattr(signal, 'SIGALRM'): signal.setitimer(signal.ITIMER_REAL, 0) signal.signal(signal.SIGALRM, old_handler) # Extract return value (if last expression) return_val = safe_globals.get('_result', None) return ExecutionResult( success=True, stdout=captured_stdout.getvalue(), stderr=captured_stderr.getvalue(), return_value=return_val, execution_time_ms=(time.perf_counter() - t0) * 1000, ) except TimeoutError: return ExecutionResult( success=False, stdout=captured_stdout.getvalue(), stderr=captured_stderr.getvalue(), exception="TimeoutError: execution exceeded time limit", timeout=True, execution_time_ms=(time.perf_counter() - t0) * 1000, ) except Exception as e: return ExecutionResult( success=False, stdout=captured_stdout.getvalue(), stderr=captured_stderr.getvalue(), exception=f"{type(e).__name__}: {e}", execution_time_ms=(time.perf_counter() - t0) * 1000, ) finally: sys.stdout = old_stdout sys.stderr = old_stderr def _check_safety(self, code: str) -> Optional[str]: """Pre-execution safety check. Returns reason if blocked, None if safe.""" code_lower = code.lower() # Block dangerous imports (allow safe modules) for mod in self.BLOCKED_MODULES: if mod in self.allow_imports: continue # Check for import statements patterns = [ f'import {mod}', f'from {mod}', f'import{mod}', # no space variant ] for pattern in patterns: if pattern in code_lower: return f"blocked module: {mod}" # Allow safe module imports (don't block them) # The safe modules are already injected into sandbox globals # So `from fractions import Fraction` will work naturally # Block dangerous builtins for builtin in self.BLOCKED_BUILTINS: if re.search(rf'\b{builtin}\b', code) and builtin not in ('open',): return f"blocked builtin: {builtin}" # Block file system operations (unless allow_io) if not self.allow_io: fs_patterns = [ r'open\s*\(', r'\.read\(', r'\.write\(', r'\.delete\(', r'os\.', r'pathlib', r'shutil', r'glob', ] for pattern in fs_patterns: if re.search(pattern, code): return f"blocked: file system access ({pattern})" # Block network net_patterns = [ r'requests\.', r'urllib', r'http\.', r'socket\.', r'aiohttp', r'httpx', ] for pattern in net_patterns: if re.search(pattern, code): return f"blocked: network access ({pattern})" # Block exec/eval if re.search(r'\bexec\s*\(', code) or re.search(r'\beval\s*\(', code): return "blocked: exec/eval" return None def _build_safe_globals(self, context: Optional[Dict] = None) -> Dict: """Build restricted global namespace.""" import math, json, re, hashlib, random, itertools, collections, functools import fractions, decimal, string, textwrap, typing, dataclasses from fractions import Fraction from decimal import Decimal from collections import Counter, defaultdict, OrderedDict from typing import List, Dict, Tuple, Optional, Set, Any safe_builtins = { 'abs': abs, 'all': all, 'any': any, 'bool': bool, 'chr': chr, 'dict': dict, 'dir': dir, 'divmod': divmod, 'enumerate': enumerate, 'filter': filter, 'float': float, 'format': format, 'frozenset': frozenset, 'getattr': getattr, 'hasattr': hasattr, 'hash': hash, 'hex': hex, 'id': id, 'int': int, 'isinstance': isinstance, 'issubclass': issubclass, 'iter': iter, 'len': len, 'list': list, 'map': map, 'max': max, 'min': min, 'next': next, 'object': object, 'oct': oct, 'ord': ord, 'pow': pow, 'print': print, 'property': property, 'range': range, 'repr': repr, 'reversed': reversed, 'round': round, 'set': set, 'setattr': setattr, 'slice': slice, 'sorted': sorted, 'str': str, 'sum': sum, 'super': super, 'tuple': tuple, 'type': type, 'vars': vars, 'zip': zip, 'True': True, 'False': False, 'None': None, # Exception classes 'Exception': Exception, 'ValueError': ValueError, 'TypeError': TypeError, 'KeyError': KeyError, 'IndexError': IndexError, 'AttributeError': AttributeError, 'ZeroDivisionError': ZeroDivisionError, 'StopIteration': StopIteration, 'RuntimeError': RuntimeError, 'NotImplementedError': NotImplementedError, 'OverflowError': OverflowError, 'ImportError': ImportError, 'NameError': NameError, 'SyntaxError': SyntaxError, 'AssertionError': AssertionError, 'OSError': OSError, 'FileNotFoundError': FileNotFoundError, 'PermissionError': PermissionError, 'TimeoutError': TimeoutError, 'MemoryError': MemoryError, 'ArithmeticError': ArithmeticError, 'LookupError': LookupError, 'UnicodeError': UnicodeError, 'DeprecationWarning': DeprecationWarning, 'Warning': Warning, 'UserWarning': UserWarning, # Other useful builtins 'map': map, 'filter': filter, 'zip': zip, 'enumerate': enumerate, 'reversed': reversed, 'sorted': sorted, 'any': any, 'all': all, 'min': min, 'max': max, 'sum': sum, 'abs': abs, 'round': round, 'pow': pow, 'divmod': divmod, 'isinstance': isinstance, 'issubclass': issubclass, 'getattr': getattr, 'setattr': setattr, 'hasattr': hasattr, 'dir': dir, 'vars': vars, 'type': type, 'id': id, 'hash': hash, 'repr': repr, 'format': format, 'chr': chr, 'ord': ord, 'hex': hex, 'oct': oct, 'bool': bool, 'int': int, 'float': float, 'str': str, 'list': list, 'dict': dict, 'set': set, 'tuple': tuple, 'frozenset': frozenset, 'slice': slice, 'property': property, 'object': object, 'super': super, } safe_modules = { 'math': math, 'json': json, 're': re, 'hashlib': hashlib, 'random': random, 'itertools': itertools, 'collections': collections, 'functools': functools, 'fractions': fractions, 'decimal': decimal, 'string': string, 'textwrap': textwrap, 'typing': typing, 'dataclasses': dataclasses, 'Fraction': Fraction, 'Decimal': Decimal, 'Counter': Counter, 'defaultdict': defaultdict, 'OrderedDict': OrderedDict, 'List': List, 'Dict': Dict, 'Tuple': Tuple, 'Optional': Optional, 'Set': Set, 'Any': Any, } # Safe import function that only allows safe modules safe_imports = set(self.SAFE_MODULES) | set(self.allow_imports) def safe_import(name, *args, **kwargs): base_name = name.split('.')[0] if base_name in safe_imports: return __import__(name, *args, **kwargs) raise ImportError(f"Import of '{name}' is not allowed in sandbox") safe_builtins['__import__'] = safe_import safe_globals = { '__builtins__': safe_builtins, '__name__': '__main__', '__doc__': None, '__file__': '', **safe_modules, } # Inject context if context: safe_globals.update(context) return safe_globals # ══════════════════════════════════════════════════════════════════════════════ # THREE COLUMN THINKING FOR CODE # ══════════════════════════════════════════════════════════════════════════════ class CodeComposer: """Generates code analysis using Three Column Thinking. Column 1: CODE — The actual Python implementation Column 2: LOGIC — Formal description of what the code does Column 3: VERIFICATION — Tests and assertions to prove correctness """ def __init__(self): self.parser = CodeParser() self.sandbox = SandboxExecutor(timeout_seconds=5.0) def analyze_and_verify(self, code: str, run_tests: bool = True) -> Dict[str, Any]: """Full analysis: parse → verify structure → run → test. Returns comprehensive report with Three Column Thinking. """ result = { "code": code, "analysis": None, "execution": None, "test_results": None, "three_column": None, "overall_verdict": "", } # Column 1: Parse analysis = self.parser.analyze(code) result["analysis"] = analysis if not analysis.valid_syntax: result["overall_verdict"] = f"SYNTAX_ERROR: {analysis.error_message}" return result # Column 2: Execute exec_result = self.sandbox.execute(code) result["execution"] = exec_result # Column 3: Extract and run tests if run_tests and analysis.has_tests: test_code = self._extract_tests(code) if test_code: test_result = self.sandbox.execute(test_code) result["test_results"] = test_result # Three Column Thinking result["three_column"] = self._three_column(code, analysis, exec_result) # Overall verdict result["overall_verdict"] = self._overall_verdict(analysis, exec_result, result.get("test_results")) return result def _extract_tests(self, code: str) -> Optional[str]: """Extract test functions from code.""" try: tree = ast.parse(code) except SyntaxError: return None test_lines = [] for node in ast.walk(tree): if isinstance(node, ast.FunctionDef) and node.name.startswith('test_'): # Get the function source start = node.lineno - 1 end = node.end_lineno if hasattr(node, 'end_lineno') else start + 10 lines = code.split('\n')[start:end] test_lines.extend(lines) test_lines.append(f" {node.name}()") test_lines.append("") if not test_lines: return None # Build test runner test_code = code + "\n\n# Auto-extracted tests\n" test_code += "\n".join(test_lines) return test_code def _three_column(self, code: str, analysis: CodeAnalysis, exec_result: ExecutionResult) -> Dict[str, str]: """Generate Three Column Thinking analysis.""" # Column 1: CODE code_col = f"Python code: {analysis.n_functions} functions, {analysis.n_classes} classes, " code_col += f"{analysis.n_imports} imports. " code_col += f"Complexity: {analysis.cyclomatic_complexity}. " if analysis.has_main_guard: code_col += "Has __main__ guard. " if analysis.has_tests: code_col += "Has tests. " # Column 2: LOGIC logic_col = "" if analysis.function_names: logic_col += f"Functions: {', '.join(analysis.function_names[:5])}. " if analysis.class_names: logic_col += f"Classes: {', '.join(analysis.class_names[:3])}. " if analysis.has_error_handling: logic_col += "Includes error handling. " if analysis.has_docstrings: logic_col += "Documented with docstrings. " logic_col += f"NRCI coherence: {analysis.nrci_score:.2f}. " # Column 3: VERIFICATION ver_col = "" if exec_result.success: ver_col += "Execution: PASS. " if exec_result.stdout: ver_col += f"Output: {exec_result.stdout[:100]}. " else: ver_col += f"Execution: FAIL ({exec_result.exception}). " if analysis.has_tests: ver_col += "Tests present. " ver_col += f"Verdict: {analysis.quality_verdict}. " return { "code": code_col, "logic": logic_col, "verification": ver_col, } def _overall_verdict(self, analysis: CodeAnalysis, exec_result: ExecutionResult, test_result: Optional[ExecutionResult]) -> str: """Determine overall verdict.""" if not analysis.valid_syntax: return "SYNTAX_ERROR" if not exec_result.success: return f"RUNTIME_ERROR: {exec_result.exception}" if test_result and not test_result.success: return f"TEST_FAILURE: {test_result.exception}" score = analysis.nrci_score if score >= 0.75: return "EXCELLENT" elif score >= 0.60: return "GOOD" elif score >= 0.45: return "FAIR" else: return "NEEDS_WORK" # ══════════════════════════════════════════════════════════════════════════════ # CODE VALIDATOR (for user-provided code) # ══════════════════════════════════════════════════════════════════════════════ class CodeValidator: """Validates Python code for correctness and safety. Use this to check code before execution or to verify generated code meets quality standards. """ def __init__(self): self.parser = CodeParser() self.sandbox = SandboxExecutor(timeout_seconds=2.0) def validate(self, code: str) -> Dict[str, Any]: """Validate code. Returns detailed report.""" analysis = self.parser.analyze(code) # Safety check sandbox = SandboxExecutor() safety_issue = sandbox._check_safety(code) # Try to compile try: compile(code, '', 'exec') compiles = True compile_error = "" except SyntaxError as e: compiles = False compile_error = f"Line {e.lineno}: {e.msg}" return { "valid_syntax": analysis.valid_syntax, "compiles": compiles, "compile_error": compile_error, "safe": safety_issue is None, "safety_issue": safety_issue, "analysis": analysis, "nrci_score": analysis.nrci_score, "verdict": analysis.quality_verdict, "recommendations": self._recommendations(analysis, safety_issue), } def _recommendations(self, analysis: CodeAnalysis, safety_issue: Optional[str]) -> List[str]: """Generate improvement recommendations.""" recs = [] if not analysis.has_docstrings: recs.append("Add docstrings to functions and classes") if not analysis.has_type_hints: recs.append("Add type hints for better code clarity") if not analysis.has_error_handling and analysis.n_functions > 0: recs.append("Add try/except for error handling") if not analysis.has_tests and analysis.n_functions > 2: recs.append("Add test functions (def test_*)") if analysis.cyclomatic_complexity > 15: recs.append(f"Reduce complexity (currently {analysis.cyclomatic_complexity})") if analysis.max_depth > 5: recs.append(f"Reduce nesting depth (currently {analysis.max_depth})") if safety_issue: recs.append(f"Safety issue: {safety_issue}") return recs # ══════════════════════════════════════════════════════════════════════════════ # CONVENIENCE FUNCTIONS # ══════════════════════════════════════════════════════════════════════════════ def analyze_code(code: str) -> CodeAnalysis: """Quick structural analysis.""" return CodeParser().analyze(code) def run_code(code: str, timeout: float = 5.0) -> ExecutionResult: """Quick sandboxed execution.""" return SandboxExecutor(timeout_seconds=timeout).execute(code) def validate_code(code: str) -> Dict[str, Any]: """Quick validation.""" return CodeValidator().validate(code) def verify_code(code: str) -> Dict[str, Any]: """Full Three Column Thinking verification.""" return CodeComposer().analyze_and_verify(code)