domain/math_validator.py

# 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, ""