npc-reason/verifier/step_verifier.py

"""NPC Reason — mechanical step verifier (THE core artifact).

PURE CODE. No model, no LLM judgment anywhere. This is what makes the
"verifiable-rate" metric un-fakeable, and it is reused verbatim as the RL reward
signal in a later dispatch — so it is frozen (VERIFIER.lock) the moment its tests pass.

DEFINITION (committed; see reports/PREREG.md and VERIFIER.lock):
  - A reasoning chain is a sequence of steps ending in a final answer.
  - A load-bearing numeric step carries an inline checkable assertion in the canonical
    form  <<EXPR = RESULT>>  where EXPR is an arithmetic/algebraic expression over
    numbers and earlier-BOUND variables, and RESULT is the claimed value.
  - A step is VERIFIED if a SymPy evaluation of EXPR equals RESULT within tolerance
    (exact for integers/rationals; 1e-6 relative for floats).
  - A chain is VERIFIABLE iff:
      (a) it has >=1 load-bearing  <<...>>  assertion (no bare asserted numbers drive it),
      (b) every such assertion VERIFIES, and
      (c) the final answer equals the RESULT of the last load-bearing step
          (the chain COMPOSES to its conclusion).
  - A chain is CORRECT iff its final answer equals the gold answer. CORRECT and
    VERIFIABLE are INDEPENDENT axes.

TOLERANCE POLICY (explicit, frozen):
  - Parse EXPR and RESULT with sympy.sympify (after a small, fixed normalization).
  - Exact branch: if simplify(EXPR_value - RESULT_value) == 0  -> VERIFIED.
  - Float branch: else, if both are finite numbers and
        |EXPR_value - RESULT_value| <= 1e-6 * max(1, |RESULT_value|)  -> VERIFIED.
  - Otherwise NOT verified.

FAIL-CLOSED (conservative by design):
  - Unparseable EXPR or RESULT            -> NOT verified (reason recorded).
  - EXPR references an UNBOUND symbol      -> NOT verified (cannot confirm).
  - Any exception during evaluation        -> NOT verified.
  A chain only counts VERIFIABLE if the checker can ACTUALLY confirm every load-bearing step.

VARIABLE BINDING (v1, documented):
  - A step may bind a name:  `let total = <<3*8 = 24>>`  or  `total = <<3*8 = 24>>`.
    The name binds to the (parsed) RESULT value and is usable by later EXPRs.
  - Binding tests internal consistency: each later step is checked against the values
    the chain itself previously stated.
  - V1 SCOPE: assertions are concrete-valued (EXPR evaluates to a number once prior
    bindings are substituted). Algebra with FREE variables / equation-solving such as
    `<<x**2 - 4 = 0>>` with x unbound is OUT OF SCOPE for v1 and FAILS CLOSED
    (recorded reason: "unbound symbol"). It is noted as a verifier-v2 extension.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from typing import Optional

import sympy
from sympy import Rational, simplify, sympify

# ----------------------------------------------------------------------------- #
# Patterns
# ----------------------------------------------------------------------------- #

# Optional binding name, then  << EXPR = RESULT >>. Non-greedy EXPR stops at the
# FIRST '=' inside the brackets. Binding name must start with a letter/underscore,
# so numeric tokens to the left (e.g. "2 + 2 =") are never captured as a name.
ASSERTION = re.compile(
    r"(?:(?:let\s+)?([A-Za-z_]\w*)\s*=\s*)?<<\s*(.+?)\s*=\s*(.+?)\s*>>"
)

# Final-answer extractors, tried in priority order.
_BOXED = re.compile(r"\\boxed\{\s*([^{}]+?)\s*\}")
_GSM = re.compile(r"####\s*([^\n]+)")
_ANSWER_IS = re.compile(
    r"(?:final answer|the answer is|answer\s*[:=])\s*\$?\\?\(?\s*"
    r"([+-]?[\d.,/eE^*+\-() ]*\d)",
    re.IGNORECASE,
)

# Characters/sequences to normalize before sympify.
_THOUSANDS = re.compile(r"(?<=\d),(?=\d{3}\b)")
_NORM_REPLACE = (
    (r"\times", "*"),
    (r"\cdot", "*"),
    (r"\div", "/"),
    (r"\left", ""),
    (r"\right", ""),
    ("^", "**"),
    ("%", "/100"),
    ("$", ""),
    ("×", "*"),
    ("÷", "/"),
    ("−", "-"),  # unicode minus
)


@dataclass
class StepResult:
    expr: str
    claimed: str
    ok: bool
    reason: str
    binding: Optional[str] = None


@dataclass
class ChainRecord:
    n_assertions: int = 0
    n_verified: int = 0
    has_loadbearing_assertions: bool = False
    all_assertions_verified: bool = False
    final_answer: Optional[str] = None
    final_answer_value: Optional[object] = None
    composes_to_final: bool = False
    verifiable: bool = False
    correct: Optional[bool] = None
    verified_and_correct: Optional[bool] = None
    steps: list = field(default_factory=list)
    failures: list = field(default_factory=list)

    def as_dict(self) -> dict:
        return {
            "n_assertions": self.n_assertions,
            "n_verified": self.n_verified,
            "has_loadbearing_assertions": self.has_loadbearing_assertions,
            "all_assertions_verified": self.all_assertions_verified,
            "final_answer": self.final_answer,
            "composes_to_final": self.composes_to_final,
            "verifiable": self.verifiable,
            "correct": self.correct,
            "verified_and_correct": self.verified_and_correct,
            "failures": self.failures,
            "steps": [
                {"expr": s.expr, "claimed": s.claimed, "ok": s.ok,
                 "reason": s.reason, "binding": s.binding}
                for s in self.steps
            ],
        }


# ----------------------------------------------------------------------------- #
# Numeric core
# ----------------------------------------------------------------------------- #

def _normalize(raw: str) -> str:
    s = raw.strip()
    s = _THOUSANDS.sub("", s)          # 1,234 -> 1234  (drop thousands separators)
    for a, b in _NORM_REPLACE:
        s = s.replace(a, b)
    # \frac{a}{b} -> ((a)/(b))
    s = re.sub(r"\\d?frac\{([^{}]+)\}\{([^{}]+)\}", r"((\1)/(\2))", s)
    s = s.replace("\\", " ")
    return s.strip()


def _to_sympy(raw: str, bindings: dict):
    """sympify a normalized token; raises on failure (caller fails closed)."""
    expr = sympify(_normalize(raw), locals=bindings, rational=True)
    return expr


def _values_match(a, b) -> bool:
    """Exact for rationals/integers; 1e-6 relative for floats."""
    diff = simplify(a - b)
    if diff == 0:
        return True
    try:
        if diff.free_symbols:
            return False
        fa, fb = float(a), float(b)
        return abs(fa - fb) <= 1e-6 * max(1.0, abs(fb))
    except (TypeError, ValueError):
        return False


def verify_assertion(expr: str, claimed: str, bindings: dict) -> StepResult:
    """Evaluate one  <<EXPR = RESULT>>  against current bindings. Fail-closed."""
    try:
        e = _to_sympy(expr, bindings)
    except Exception as ex:  # noqa: BLE001 — fail closed on any parse error
        return StepResult(expr, claimed, False, f"unparseable expr: {ex}")
    try:
        c = _to_sympy(claimed, bindings)
    except Exception as ex:  # noqa: BLE001
        return StepResult(expr, claimed, False, f"unparseable result: {ex}")

    # EXPR must reduce to a concrete value (v1 scope: no free variables).
    if getattr(e, "free_symbols", set()):
        unbound = ", ".join(sorted(str(s) for s in e.free_symbols))
        return StepResult(expr, claimed, False, f"unbound symbol(s): {unbound}")

    try:
        ok = _values_match(e, c)
    except Exception as ex:  # noqa: BLE001
        return StepResult(expr, claimed, False, f"comparison error: {ex}")
    reason = "verified" if ok else f"mismatch: {expr} -> {e} != {claimed}"
    return StepResult(expr, claimed, ok, reason)


# ----------------------------------------------------------------------------- #
# Final answer
# ----------------------------------------------------------------------------- #

def extract_final_answer(text: str) -> Optional[str]:
    """Last \\boxed{}, else last ####, else 'the answer is X'. None if absent."""
    boxed = _BOXED.findall(text)
    if boxed:
        return boxed[-1].strip()
    gsm = _GSM.findall(text)
    if gsm:
        return gsm[-1].strip()
    m = list(_ANSWER_IS.finditer(text))
    if m:
        return m[-1].group(1).strip().rstrip(".")
    return None


def _safe_value(raw: str, bindings: dict):
    try:
        v = _to_sympy(raw, bindings)
        return None if getattr(v, "free_symbols", set()) else v
    except Exception:  # noqa: BLE001
        return None


# ----------------------------------------------------------------------------- #
# Chain
# ----------------------------------------------------------------------------- #

def verify_chain(text: str, gold_answer=None) -> dict:
    """Mechanically derive every field. No judgment. Returns ChainRecord.as_dict()."""
    rec = ChainRecord()
    bindings: dict = {}

    for m in ASSERTION.finditer(text):
        name, expr, claimed = m.group(1), m.group(2), m.group(3)
        res = verify_assertion(expr, claimed, bindings)
        res.binding = name
        rec.steps.append(res)
        rec.n_assertions += 1
        if res.ok:
            rec.n_verified += 1
        else:
            rec.failures.append({"expr": expr, "claimed": claimed, "reason": res.reason})
        # Bind the name to the CLAIMED result value (internal-consistency semantics),
        # whenever the result parses to a concrete value — even if the step failed,
        # so downstream reasons are about the downstream step, not a cascade.
        if name:
            v = _safe_value(claimed, bindings)
            if v is not None:
                bindings[name] = v

    rec.has_loadbearing_assertions = rec.n_assertions > 0
    rec.all_assertions_verified = (
        rec.has_loadbearing_assertions and rec.n_verified == rec.n_assertions
    )

    # Final answer + composition.
    fa = extract_final_answer(text)
    rec.final_answer = fa
    fa_val = _safe_value(fa, bindings) if fa is not None else None
    rec.final_answer_value = fa_val

    if rec.has_loadbearing_assertions and fa_val is not None:
        last_claimed = rec.steps[-1].claimed
        last_val = _safe_value(last_claimed, bindings)
        if last_val is not None:
            try:
                rec.composes_to_final = _values_match(fa_val, last_val)
            except Exception:  # noqa: BLE001
                rec.composes_to_final = False
    if not rec.composes_to_final and rec.has_loadbearing_assertions:
        rec.failures.append({"reason": "final answer does not compose from last step"})

    rec.verifiable = (
        rec.has_loadbearing_assertions
        and rec.all_assertions_verified
        and rec.composes_to_final
    )

    # Correctness (independent axis).
    if gold_answer is not None:
        gold_val = _safe_value(str(gold_answer), {})
        if fa_val is not None and gold_val is not None:
            try:
                rec.correct = _values_match(fa_val, gold_val)
            except Exception:  # noqa: BLE001
                rec.correct = False
        else:
            # Fall back to normalized string compare (handles non-numeric MATH answers).
            rec.correct = (
                fa is not None
                and _normalize(fa).replace(" ", "") == _normalize(str(gold_answer)).replace(" ", "")
            )
        rec.verified_and_correct = bool(rec.verifiable and rec.correct)

    return rec.as_dict()


if __name__ == "__main__":
    import json
    import sys

    blob = sys.stdin.read()
    print(json.dumps(verify_chain(blob), indent=2, default=str))