gemma_glm / llm_glm /script_engine.py
DigitalEuan's picture
Upload 14 files
9870a25 verified
Raw
History Blame Contribute Delete
31 kB
#!/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, '<sandbox>', '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__': '<sandbox>',
**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, '<validate>', '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)