runtime/evaluation/math_verifier.py

from __future__ import annotations
import json
import os
import signal
import sys
from scoring.verification import symbolic_matrix_values, symbolic_structured_values
import re
from typing import Any
from scoring.verification import normalize_answer_text as normalize_answer, compare_answers, numeric_value, symbolic_solution_context, symbolic_math_text as latex_to_text, symbolic_matrix_values as matrix_values, symbolic_reference_is_supported, symbolic_structured_values as structured_values

def parse_expression(sp: Any, parser: Any, transformations: Any, value: Any) -> Any:
    text = latex_to_text(value)
    if not symbolic_reference_is_supported(value):
        raise ValueError('expression is outside the declared symbolic subset')
    if text.count('=') == 1:
        left, right = text.split('=', 1)
        text = f'({left})-({right})'
    elif '=' in text:
        raise ValueError('expression has multiple equality signs')
    names = set(re.findall('[^\\W\\d]\\w*', text, flags=re.UNICODE))
    functions = {'sqrt': sp.sqrt, 'sin': sp.sin, 'cos': sp.cos, 'tan': sp.tan, 'csc': sp.csc, 'arccos': sp.acos, 'arctan': sp.atan, 'log': sp.log, 'ln': sp.log, 'exp': sp.exp, 'abs': sp.Abs, 'factorial': sp.factorial, 'floor': sp.floor, 'root': sp.root, 'pi': sp.pi, 'e': sp.E, 'i': sp.I, 'oo': sp.oo}
    local_dict = {name: functions.get(name, sp.Symbol(name)) for name in sorted(names)}
    global_dict = {'__builtins__': {}, 'Integer': sp.Integer, 'Float': sp.Float, 'Rational': sp.Rational, 'Symbol': sp.Symbol}
    expression = parser.parse_expr(text, local_dict=local_dict, global_dict=global_dict, transformations=transformations, evaluate=True)
    if int(sp.count_ops(expression)) > 200:
        raise ValueError('expression is too complex')
    return expression

def simple_assignment_rhs(value: Any) -> str | None:
    text = normalize_answer(value)
    if text.count('=') != 1:
        return None
    left, right = text.split('=', 1)
    left = re.sub('_\\{?[^{}=]+\\}?$', '', left)
    if not re.fullmatch('(?:[a-zA-Z]|\\\\[a-zA-Z]+)(?:\\([a-zA-Z0-9,+-]+\\))?', left) or not right:
        return None
    return right

def proportional_polynomial_equations(sp: Any, left: Any, right: Any) -> bool:
    symbols = sorted(left.free_symbols | right.free_symbols, key=str)
    if not symbols:
        return False
    try:
        if left.as_poly(*symbols) is None or right.as_poly(*symbols) is None:
            return False
        ratio = sp.cancel(left / right)
        return bool(not ratio.free_symbols and ratio != 0 and (ratio.is_finite is not False))
    except Exception:
        return False

def scalar_equivalence(candidate: Any, reference: Any) -> dict[str, Any]:
    if not symbolic_reference_is_supported(reference):
        return {'verdict': 'UNCERTAIN', 'method': 'symbolic_reference_outside_contract'}
    if not symbolic_reference_is_supported(candidate):
        return {'verdict': 'UNCERTAIN', 'method': 'symbolic_candidate_outside_contract'}
    left_text = normalize_answer(candidate)
    right_text = normalize_answer(reference)
    if left_text == right_text:
        return {'verdict': 'PASS', 'method': 'normalized_exact'}
    numeric = compare_answers(candidate, [reference], answer_type='numeric')
    if numeric['verdict'] in {'PASS', 'FAIL'}:
        return numeric
    left_assignment = simple_assignment_rhs(candidate)
    right_assignment = simple_assignment_rhs(reference)
    if left_assignment is not None and right_assignment is None:
        result = scalar_equivalence(left_assignment, reference)
        if result['verdict'] == 'PASS':
            return {'verdict': 'PASS', 'method': 'simple_assignment_rhs', 'component': result}
    if right_assignment is not None and left_assignment is None:
        result = scalar_equivalence(candidate, right_assignment)
        if result['verdict'] == 'PASS':
            return {'verdict': 'PASS', 'method': 'simple_assignment_rhs', 'component': result}
    try:
        import sympy as sp
        from sympy.parsing import sympy_parser as parser
        transformations = parser.standard_transformations + (parser.convert_xor, parser.implicit_multiplication_application)
        left = parse_expression(sp, parser, transformations, candidate)
        right = parse_expression(sp, parser, transformations, reference)
        difference = sp.simplify((left - right).rewrite(sp.acos, sp.asin))
        equality = True if difference == 0 else difference.equals(0)
        equivalent = equality is True
        proportional = bool(not equivalent and left_text.count('=') == 1 and (right_text.count('=') == 1) and proportional_polynomial_equations(sp, left, right))
        return {'verdict': 'PASS' if equivalent or proportional else 'FAIL' if equality is False else 'UNCERTAIN', 'method': 'proportional_polynomial_equation' if proportional else 'sympy_simplify', 'candidate': str(left), 'reference': str(right)}
    except Exception as exc:
        return {'verdict': 'UNCERTAIN', 'method': 'symbolic_parse_failed', 'error': f'{type(exc).__name__}: {exc}'}

def symbolic_equivalence(candidate: Any, reference: Any, *, problem: str='') -> dict[str, Any]:
    right_matrix = matrix_values(reference)
    if right_matrix is not None:
        left_matrix = matrix_values(candidate)
        vector_sequence = structured_values(candidate, problem=problem)
        reference_is_vector = len(right_matrix) == 1 or all((len(row) == 1 for row in right_matrix))
        if left_matrix is None and reference_is_vector and (vector_sequence is not None) and (vector_sequence[0] == 'sequence') and re.search('\\bvectors?\\b|\\u5411\\u91cf|\\u77e2\\u91cf', problem, re.I):
            reference_values = [cell for row in right_matrix for cell in row]
            if len(vector_sequence[1]) != len(reference_values):
                return {'verdict': 'FAIL', 'method': 'vector_cardinality_mismatch', 'candidate_count': len(vector_sequence[1]), 'reference_count': len(reference_values)}
            components = [scalar_equivalence(left, right) for left, right in zip(vector_sequence[1], reference_values)]
            verdicts = [item['verdict'] for item in components]
            return {'verdict': 'PASS' if all((value == 'PASS' for value in verdicts)) else 'FAIL' if 'FAIL' in verdicts else 'UNCERTAIN', 'method': 'vector_sequence_to_matrix_elementwise', 'components': components}
        if left_matrix is None:
            return {'verdict': 'UNCERTAIN', 'method': 'matrix_candidate_unparseable'}
        left_shape = (len(left_matrix), len(left_matrix[0]))
        right_shape = (len(right_matrix), len(right_matrix[0]))
        if left_shape != right_shape:
            return {'verdict': 'FAIL', 'method': 'matrix_shape_mismatch', 'candidate_shape': left_shape, 'reference_shape': right_shape}
        components = [scalar_equivalence(left, right) for left_row, right_row in zip(left_matrix, right_matrix) for left, right in zip(left_row, right_row)]
        verdicts = [item['verdict'] for item in components]
        verdict = 'PASS' if all((value == 'PASS' for value in verdicts)) else 'FAIL' if 'FAIL' in verdicts else 'UNCERTAIN'
        return {'verdict': verdict, 'method': 'matrix_elementwise', 'shape': right_shape, 'components': components}
    if not symbolic_reference_is_supported(reference, problem=problem):
        return {'verdict': 'UNCERTAIN', 'method': 'symbolic_reference_outside_contract'}
    if not symbolic_reference_is_supported(candidate, problem=problem):
        return {'verdict': 'UNCERTAIN', 'method': 'symbolic_candidate_outside_contract'}
    left = structured_values(candidate, problem=problem)
    right = structured_values(reference, problem=problem)
    if any((value is not None and value[0].startswith('interval:') for value in (left, right))):
        left = structured_values(candidate, problem=problem, interval=True)
        right = structured_values(reference, problem=problem, interval=True)
    if symbolic_solution_context(problem):
        if left is None and right is not None and (right[0] == 'set'):
            left = ('set', [str(candidate)])
        if right is None and left is not None and (left[0] == 'set'):
            right = ('set', [str(reference)])
    if left is not None or right is not None:
        if left is None or right is None:
            structure = left or right
            scalar = candidate if left is None else reference
            if structure[0] != 'set' and numeric_value(scalar) is not None:
                return {'verdict': 'FAIL', 'method': 'structured_numeric_scalar_mismatch'}
            return {'verdict': 'UNCERTAIN', 'method': 'structured_scalar_semantics_unresolved'}
        if left[0] != right[0]:
            verdict = 'FAIL' if left[0].startswith('interval:') and right[0].startswith('interval:') else 'UNCERTAIN'
            return {'verdict': verdict, 'method': 'structured_kind_mismatch'}
        if len(left[1]) != len(right[1]):
            return {'verdict': 'FAIL', 'method': 'structured_cardinality_mismatch', 'candidate_count': len(left[1]), 'reference_count': len(right[1])}
        if left[0] == 'sequence' or left[0].startswith('interval:'):
            components = [scalar_equivalence(a, b) for a, b in zip(left[1], right[1])]
            verdicts = [item['verdict'] for item in components]
            return {'verdict': 'PASS' if all((v == 'PASS' for v in verdicts)) else 'FAIL' if 'FAIL' in verdicts else 'UNCERTAIN', 'method': 'interval_endpoints' if left[0].startswith('interval:') else 'ordered_sequence', 'components': components}
        unmatched = list(right[1])
        components = []
        for candidate_item in left[1]:
            match = None
            uncertain = False
            for index, reference_item in enumerate(unmatched):
                result = scalar_equivalence(candidate_item, reference_item)
                if result['verdict'] == 'PASS':
                    match = index
                    components.append(result)
                    break
                uncertain = uncertain or result['verdict'] == 'UNCERTAIN'
            if match is None:
                return {'verdict': 'UNCERTAIN' if uncertain else 'FAIL', 'method': 'unordered_set', 'components': components}
            unmatched.pop(match)
        return {'verdict': 'PASS', 'method': 'unordered_set', 'components': components}
    return scalar_equivalence(candidate, reference)

def fallback_verify(request: dict[str, Any]) -> dict[str, Any]:
    candidate = request.get('candidate')
    references = request.get('reference_answers')
    if references is None:
        references = [request.get('reference_answer')]
    if not isinstance(references, list) or not references or candidate is None:
        return {'verdict': 'UNCERTAIN', 'details': 'missing expression'}
    problem = str(request.get('problem') or '')
    results = []
    for reference in references:
        result = symbolic_equivalence(candidate, reference, problem=problem)
        results.append(result)
        if result['verdict'] == 'PASS':
            return {'verdict': 'PASS', 'details': results}
    verdict = 'FAIL' if results and all((item['verdict'] == 'FAIL' for item in results)) else 'UNCERTAIN'
    candidate_text = str(candidate)
    if verdict == 'FAIL' and re.search('\\bangle\\b|\\u89d2', problem, re.I) and re.search('\\\\pi|(?<![A-Za-z])pi(?![A-Za-z])', candidate_text, re.I) and any((re.search('(?<!\\d)180(?!\\d)', str(reference)) for reference in references)):
        return {'verdict': 'UNCERTAIN', 'details': [*results, {'method': 'angle_unit_convention_requires_model_review'}]}
    return {'verdict': verdict, 'details': results}
_MATH_VERIFY: tuple[Any, Any] | None | bool = None
SYMBOLIC_FALLBACK_TIMEOUT_SECONDS = 20.0

def _symbolic_fallback_timeout(_signum: int, _frame: Any) -> None:
    raise TimeoutError('symbolic fallback exceeded its verifier deadline')

def parsed_values(parse: Any, value: Any) -> list[Any]:
    text = str(value).strip()
    if not text or len(text) > 8192:
        return []
    values = parse(text, parsing_timeout=3, raise_on_error=False)
    if values:
        return values
    return parse(f'\\boxed{{{text}}}', parsing_timeout=3, raise_on_error=False)

def math_verify_functions() -> tuple[Any, Any] | None:
    global _MATH_VERIFY
    if _MATH_VERIFY is False:
        return None
    if _MATH_VERIFY is None:
        try:
            from math_verify import parse, verify
        except Exception:
            _MATH_VERIFY = False
            return None
        _MATH_VERIFY = (parse, verify)
    return _MATH_VERIFY

def run_fallback(request: dict[str, Any]) -> dict[str, Any]:
    previous = signal.signal(signal.SIGALRM, _symbolic_fallback_timeout)
    signal.setitimer(signal.ITIMER_REAL, SYMBOLIC_FALLBACK_TIMEOUT_SECONDS)
    try:
        return fallback_verify(request)
    except Exception as exc:
        return {'verdict': 'UNCERTAIN', 'details': [{'method': 'math_equivalence_fail_closed', 'error': f'{type(exc).__name__}: {str(exc)[:1000]}'}]}
    finally:
        signal.setitimer(signal.ITIMER_REAL, 0.0)
        signal.signal(signal.SIGALRM, previous)

def verify_request(request: dict[str, Any]) -> dict[str, Any]:
    candidate = request.get('candidate')
    references = request.get('reference_answers')
    if references is None:
        references = [request.get('reference_answer')]
    problem = str(request.get('problem') or '')
    structured_reference = isinstance(references, list) and any((symbolic_matrix_values(reference) is not None or symbolic_structured_values(reference, problem=problem) is not None for reference in references))
    structured_candidate = symbolic_matrix_values(candidate) is not None or symbolic_structured_values(candidate, problem=problem) is not None
    if candidate is not None and isinstance(references, list) and (not (structured_reference or structured_candidate)):
        try:
            functions = math_verify_functions()
            if functions is None:
                raise RuntimeError('math_verify unavailable')
            parse, verify = functions
            target = parsed_values(parse, candidate)
            if target:
                for reference in references:
                    gold = parsed_values(parse, reference)
                    if gold and verify(gold, target, strict=True, timeout_seconds=3, raise_on_error=False):
                        return {'verdict': 'PASS', 'details': [{'method': 'math_verify_0.9.0_strict', 'reference': str(reference)}]}
        except Exception:
            pass
    return run_fallback(request)

def main() -> int:
    result_stream = os.fdopen(os.dup(sys.stdout.fileno()), 'w', buffering=1)
    os.dup2(sys.stderr.fileno(), sys.stdout.fileno())
    if '--jsonl-worker' in sys.argv[1:]:
        if math_verify_functions() is None:
            raise RuntimeError('formal math verifier requires math_verify')
        for line in sys.stdin:
            try:
                result = verify_request(json.loads(line))
            except Exception as exc:
                result = {'verdict': 'UNCERTAIN', 'details': {'method': 'math_worker_fail_closed', 'error': f'{type(exc).__name__}: {str(exc)[:1000]}'}}
            print(json.dumps(result, ensure_ascii=False, sort_keys=True), file=result_stream, flush=True)
        return 0
    print(json.dumps(verify_request(json.load(sys.stdin)), ensure_ascii=False, sort_keys=True), file=result_stream, flush=True)
    return 0

if __name__ == '__main__':
    raise SystemExit(main())