# domain/math_validator.py — V1.1 (STABLE REGEX POLYGRAPH) import re import logging import multiprocessing import time import asyncio from typing import Tuple, List, Optional import sympy from sympy.parsing.sympy_parser import parse_expr logger = logging.getLogger(__name__) # Expressions that are structurally unparseable but pedagogically harmless _PIPE_SEPARATED_RESULT = re.compile(r'\|') _HEBREW_ONLY = re.compile(r'^[\u0590-\u05FF\s]+$') # LaTeX commands that SymPy cannot parse — strip layout but KEEP math functions _LATEX_STRIP = re.compile( r'\\(?:left|right|cdot|times|div|pm|mp|leq|geq|neq' r'|approx|infty|text|mathrm|mathbf|boxed|underbrace|overbrace|hat|bar|vec|dot|overline|underline)\b' ) # V307.0: Functions that often lack parentheses in LLM output (e.g. lnx) _MATH_FUNC_PARENS = ['ln', 'sin', 'cos', 'tan', 'sqrt', 'log', 'exp', 'Abs'] def _latex_to_sympy_str(latex_str: str) -> str: """ Best-effort LaTeX → SymPy-parseable string. V310.0: Aggressive Hebrew stripping and malformed notation cleanup. """ if latex_str is None: return "" s = str(latex_str).strip() # 0. V310.0: Strip Hebrew characters and BOM/Zero-width chars immediately s = re.sub(r'[\u0590-\u05FF\u200B-\u200D\uFEFF]', ' ', s) # 1. Handle \frac{a}{b} → (a)/(b) loop_counter = 0 max_loops = 15 while r'\frac' in s and loop_counter < max_loops: old_s = s s = re.sub(r'\\frac\s*\{([^{}]*)\}\s*\{([^{}]*)\}', r'(\1)/(\2)', s) if old_s == s: s = s.replace(r'\frac', '(frac_err)') break loop_counter += 1 # 2. Convert LaTeX functions to plain words (e.g. \ln -> ln) s = s.replace(r'\ln', ' ln ').replace(r'\sin', ' sin ').replace(r'\cos', ' cos ') s = s.replace(r'\tan', ' tan ').replace(r'\sqrt', ' sqrt ').replace(r'\log', ' log ') s = s.replace(r'\exp', ' exp ').replace(r'\pi', ' pi ').replace(r'\theta', ' theta ') # 3. Remove remaining purely structural LaTeX commands s = _LATEX_STRIP.sub(' ', s) # 4. Remove LaTeX delimiters/wrappers s = s.replace('{', '(').replace('}', ')').replace('$', '') s = s.replace(r'\left', '').replace(r'\right', '') # 5. V307.0: Fix implicit function arguments (e.g. lnx -> ln(x)) for func in _MATH_FUNC_PARENS: pattern = r'\b' + func + r'\b\s*([^( \t\n\r\f\v,]+)' s = re.sub(pattern, func + r'(\1)', s) # 6. Handle absolute value pipes |x| -> Abs(x) loop_counter = 0 while '|' in s and s.count('|') >= 2 and loop_counter < 10: s = re.sub(r'\|([^|]+)\|', r'Abs(\1)', s) loop_counter += 1 # 7. Implicit multiplication: 2x → 2*x (only if not inside a word) s = re.sub(r'(\d)([a-zA-Z(])', r'\1*\2', s) # 8. V280.0: Equals sign handling is now moved to _check_segment # for more robust parsing of equations. # 9. Final cleanup: Remove illegal SymPy chars like ', ", ?, ! s = re.sub(r'[?!\'"]', '', s) s = re.sub(r'\s+', ' ', s) return s.strip() def _is_plaintext(expr_str: str) -> bool: if _HEBREW_ONLY.match(expr_str): return True if _PIPE_SEPARATED_RESULT.search(expr_str) and not any(c in expr_str for c in ['+', '-', '*', '/', '^', '=']): return True return False class MathPolygraph: TIMEOUT_SECONDS = 3 @staticmethod def _sympify_worker(expr_str: str, queue: multiprocessing.Queue): """ V280.0: Security Hardened Worker. 1. Character Whitelist: Only allow safe mathematical characters. 2. parse_expr(evaluate=False): Prevent RCE and immediate evaluation. """ try: # V317.8: Suppress SymPy Deprecation Warnings (e.g. non-Expr in Pow) import warnings from sympy.utilities.exceptions import SymPyDeprecationWarning warnings.filterwarnings("ignore", category=SymPyDeprecationWarning) # RCE Prevention: Extreme character whitelist before parsing # V280.0 FIX: Added ! for factorials and ensured strict match. safe_pattern = r'^[a-zA-Z0-9\s\+\-\*\/\^\(\)\.\,\!\%\=]+$' if not re.match(safe_pattern, expr_str): queue.put(False) return # Security: evaluate=False stops automatic eval() of passed strings. res = parse_expr(expr_str, evaluate=False) # V280.0 FIX: Catch arithmetic errors like 1/0. # In SymPy, 1/0 evaluates to 'zoo' (ComplexInfinity). if res is not None: # evaluate the expression evaluated = res.doit() # If the result is infinite (zoo, oo, -oo) or NaN, treat as error # We check is_finite directly. if hasattr(evaluated, 'is_finite') and evaluated.is_finite is False: raise ZeroDivisionError("Infinite or undefined result") if hasattr(evaluated, 'is_nan') and evaluated.is_nan: raise ValueError("NaN result") queue.put(True) except (ZeroDivisionError, TypeError, ValueError, Exception) as e: queue.put(False) @staticmethod def _sympify_with_timeout(expr_str: str) -> bool: """Helper to run parsing in a separate process to enforce timeout.""" if not expr_str or not expr_str.strip(): return True # Strip characters that might survive _latex_to_sympy_str but fail whitelist s = expr_str.replace('\\', '').replace('_', '').replace('{', '(').replace('}', ')') queue = multiprocessing.Queue() process = multiprocessing.Process(target=MathPolygraph._sympify_worker, args=(s, queue)) try: process.start() # Windows needs a generous timeout for cold process start + SymPy import. # 10 seconds is safe for verification/testing. process.join(timeout=10) if process.is_alive(): process.terminate() process.join() with open('debug_math.val', 'a', encoding='utf-8') as f: f.write(f"[{time.time()}] TIMEOUT on '{s}'\n") return None # TIMEOUT if not queue.empty(): return queue.get() return False except Exception: if process.is_alive(): process.terminate() return False @staticmethod async def _validate_single(text: str, step_id) -> Tuple[bool, str]: """ V280.0 REDESIGN: 1. No Blind Stripping: Extracts $...$ or $$...$$ using re.finditer with DOTALL. 2. Security: Uses parse_expr(evaluate=False). 3. Equations: Splits by '=' and validates parts to bypass SymPy's '=' limitation. 4. Multi-Equal: Handles x=y=5 without crashing. 5. Empty Guard: Skips $$$$. """ if not text or not text.strip(): return True, "" # regex: find both $$display$$ and $inline$ blocks. DOTALL allows multi-line display math. # Group 1 = display math, Group 2 = inline math math_pattern = re.compile(r'\$\$(.*?)\$\$|\$(.*?)\$', re.DOTALL) matches = list(re.finditer(math_pattern, text)) if not matches: # V280.0 Rule: If no delimiters are found, treat the whole string as plain text # or try to parse if it looks like math (existing behavior for backward compatibility) if _is_plaintext(text): return True, "" return await MathPolygraph._check_segment(text, step_id) for match in matches: # Group 1 (Display) or Group 2 (Inline) content = (match.group(1) or match.group(2) or "").strip() # 5. Empty String Guard if not content: continue # V280.0 Fix: Multi-line display math might contain multiple equations. # Split by newline before validating segments. sub_segments = [s.strip() for s in content.split('\n') if s.strip()] for sub in sub_segments: ok, reason = await MathPolygraph._check_segment(sub, step_id) if not ok: return False, reason return True, "" @staticmethod async def _check_segment(raw_segment: str, step_id) -> Tuple[bool, str]: """Internal helper to validate a single extracted math segment.""" # 4. Multi-Equal Sign Handling & Unpacking Crash Prevention eq_count = raw_segment.count('=') parts_to_check = [] if eq_count >= 1: # Split by all equalities and check each segment (e.g. x=y=5 -> check x, y, 5) # This bypasses SymPy's inability to parse "=" and prevents split() unpacking errors. parts_to_check = [p.strip() for p in raw_segment.split('=') if p.strip()] else: parts_to_check = [raw_segment] for part in parts_to_check: sympy_str = _latex_to_sympy_str(part) if not sympy_str or sympy_str in ('', '-', '()', '( )'): continue try: # Run with timeout to prevent ReDoS or complex simplification hangs status = await asyncio.to_thread(MathPolygraph._sympify_with_timeout, sympy_str) if status is False: return False, f"SYMPY_PARSE_ERROR:step_{step_id}" elif status is None: # Timeout is treated as a soft warning for now logger.warning(f"[V280.0] SymPy timeout on segment: {part}") except Exception as e: logger.error(f"[V280.0] Unexpected validation crash: {e}") return False, f"SYMPY_CRASH:step_{step_id}" return True, "" @staticmethod async def validate_step_sequence(steps: List[dict], topic: str = "GENERAL") -> Tuple[bool, str]: if not steps: return True, "" # V8.9.4: Skip deep SymPy parsing for discrete sequence steps to avoid false-positive SyntaxErrors is_sequence = topic and "SEQUENCE" in topic.upper() for step in steps: step_id = step.get('step_id', step.get('step_number', '?')) math_fields = [] for field in ('math_latex', 'block_math', 'math'): val = step.get(field) if val and isinstance(val, str) and val.strip(): math_fields.append(val.strip()) if not math_fields: continue # If sequence, we only check if it's "valid-ish" LaTeX vs deep SymPy check if is_sequence: # Basic sanity check for LaTeX balance if math_fields[0].count('{') != math_fields[0].count('}'): return False, f"LATEX_BRACKET_MISMATCH:step_{step_id}" continue ok, reason = await MathPolygraph._validate_single(math_fields[0], step_id) if not ok: return False, reason return True, "" @staticmethod def are_equivalent(latex1: str, latex2: str) -> bool: """ V1.2: Checks if two LaTeX expressions are mathematically equivalent. Supports expressions and equations (by converting to 'expr = 0'). """ try: # V280.0: Handle Equations in Equivalence Check # If both contain '=', split and compare parts. # Only recurse once! if '=' in latex1 and '=' in latex2 and latex1.count('=') == 1 and latex2.count('=') == 1: parts1 = [p.strip() for p in latex1.split('=') if p.strip()] parts2 = [p.strip() for p in latex2.split('=') if p.strip()] if len(parts1) == 2 and len(parts2) == 2: return MathPolygraph.are_equivalent(parts1[0], parts2[0]) and \ MathPolygraph.are_equivalent(parts1[1], parts2[1]) s1_raw = _latex_to_sympy_str(latex1) s2_raw = _latex_to_sympy_str(latex2) # Check for inequalities in raw LaTeX to be safe inequalities = ['<', '>', r'\leq', r'\geq', r'\neq', r'\leq', r'\geq'] if any(iq in latex1 for iq in inequalities) or any(iq in latex2 for iq in inequalities): return latex1.strip() == latex2.strip() # Security: Strict Whitelist for Equivalence Check safe_pattern = r'^[a-zA-Z0-9\s\+\-\*\/\^\(\)\.\,\!\=]+$' def is_safe(s): clean = s.replace('\\', '').replace('_', '').replace('{', '(').replace('}', ')') return bool(re.match(safe_pattern, clean)) if not (is_safe(s1_raw) and is_safe(s2_raw)): return latex1.strip() == latex2.strip() expr1 = parse_expr(s1_raw, evaluate=False) expr2 = parse_expr(s2_raw, evaluate=False) # "Variable Trap": Basic structural equivalence if variables are involved if len(expr1.free_symbols) > 0 or len(expr2.free_symbols) > 0: return sympy.simplify(expr1 - expr2) == 0 # Numerical Identity check: simplify(LHS - RHS) == 0 diff = sympy.simplify(expr1 - expr2) return diff == 0 except Exception as e: logger.warning(f"[POLYGRAPH] Equivalence check failed: {e}") return False @staticmethod async def verify_algebraic_consistency(steps: List[dict], topic: str = "GENERAL") -> Tuple[bool, str]: """ V1.3: Checks if a sequence of steps is algebraically consistent. Currently checks if subsequent steps are equivalent (for simplifications). """ # V8.9.4: Skip deep SymPy parsing for discrete sequence steps if topic and "SEQUENCE" in topic.upper(): return True, "" math_steps = [] for step in steps: math = step.get('math_latex') or step.get('block_math') or step.get('math') if math and isinstance(math, str) and math.strip(): # Avoid validating plaintext logic blocks if not _is_plaintext(math): math_steps.append({'id': step.get('step_id', '?'), 'math': math}) if len(math_steps) < 2: return True, "" for i in range(len(math_steps) - 1): s1 = math_steps[i]['math'] s2 = math_steps[i+1]['math'] # Simple heuristic: Only verify if they look like comparable equations/expressions if not MathPolygraph.are_equivalent(s1, s2): logger.info(f"[POLYGRAPH] Consistency warning between {s1} and {s2}") # We return False only if we are VERY sure. # For now, we'll return False to trigger self-correction as requested. return False, f"ALGEBRAIC_INCONSISTENCY:step_{math_steps[i+1]['id']}" return True, ""