Merge pull request #626 from AssetOverflow/feat/r2-solver-verifier

feat(constraint): R2 Pack B — exact integer solver + answer-choice verifier

generate/answer_choices/__init__.py

@@ -0,0 +1,24 @@
+"""Multiple-choice answer verification (off-serving).
+
+Ties a PROVEN value to exactly one labeled option and flags answer-key contradictions — the
+engine asserts the consistent answer and names a wrong key, never silently accepting it. Used
+by the R2 constraint organ (and reusable by any lane that proves an integer answer). Imports no
+``generate.derivation`` / ``core.reliability_gate``.
+"""
+
+from __future__ import annotations
+
+from generate.answer_choices.parse import parse_option_value, parse_options
+from generate.answer_choices.verify import (
+    ChoiceVerdict,
+    VERDICT_STATUSES,
+    verify_answer_choice,
+)
+
+__all__ = [
+    "ChoiceVerdict",
+    "VERDICT_STATUSES",
+    "parse_option_value",
+    "parse_options",
+    "verify_answer_choice",
+]

generate/answer_choices/parse.py

@@ -0,0 +1,49 @@
+"""Parse a multiple-choice option map into normalized integer values (R2 C4).
+
+Options arrive as ``{label: value}``. A value may be a bare integer (the R2 gold form) or a
+string carrying exactly one integer (``"11"``, ``"11 chickens"``, ``"$11"``). A string with
+zero or several integers denotes no single value and REFUSES — the verifier must never guess
+which number an ambiguous option meant. Off-serving; deterministic.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from generate.meaning_graph.reader import Refusal
+
+_INT_RE = re.compile(r"-?\d+")
+
+
+def parse_option_value(value: Any) -> int | None:
+    """The integer an option denotes, or ``None`` if it denotes no single integer.
+
+    An ``int`` is taken verbatim; a ``str`` is accepted iff it carries exactly one integer
+    (so ``"between 5 and 10"`` -> ``None``). ``bool`` is rejected (``True`` is not a count).
+    """
+    if isinstance(value, bool):
+        return None
+    if isinstance(value, int):
+        return value
+    if isinstance(value, str):
+        found = _INT_RE.findall(value)
+        if len(found) == 1:
+            return int(found[0])
+    return None
+
+
+def parse_options(raw: Any) -> dict[str, int] | Refusal:
+    """Normalize ``{label: value}`` into ``{label: int}``; refuse an empty or unparseable map."""
+    if not isinstance(raw, dict) or not raw:
+        return Refusal("no_options")
+    out: dict[str, int] = {}
+    for label, value in raw.items():
+        parsed = parse_option_value(value)
+        if parsed is None:
+            return Refusal("unparseable_option", f"{label}: {value!r}")
+        out[str(label)] = parsed
+    return out
+
+
+__all__ = ["parse_option_value", "parse_options"]

generate/answer_choices/verify.py

@@ -0,0 +1,81 @@
+"""Verify a computed answer against multiple-choice options, flagging key contradictions (R2 C4).
+
+Truth discipline (the user's Phase 5): the engine ties its PROVEN value to exactly one labeled
+option. If a provided answer key disagrees with the proof, that is not a refusal — it is a
+confident **contradiction** verdict ("the math says A; the key says C — the key is wrong"). The
+verifier refuses only when the proof cannot be tied to exactly one option (no match, or a
+duplicate-valued match). Off-serving; deterministic.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from generate.answer_choices.parse import parse_options
+from generate.meaning_graph.reader import Refusal
+
+#: A confident verdict status — NOT a refusal. ``contradiction`` asserts the key is wrong while
+#: the engine's value stands; ``consistent`` confirms (or, with no key, simply labels) it.
+VERDICT_STATUSES = frozenset({"consistent", "contradiction"})
+
+
+@dataclass(frozen=True, slots=True)
+class ChoiceVerdict:
+    """The outcome of tying a proven value to the options. ``computed_label`` is the option the
+    proof matches; ``provided_label`` is the supplied key (or ``None``); ``message`` is the
+    user-facing sentence."""
+
+    computed_value: int
+    computed_label: str
+    provided_label: str | None
+    status: str
+    message: str
+
+
+def _suffix(noun: str) -> str:
+    return f" {noun}" if noun else ""
+
+
+def verify_answer_choice(
+    computed_value: int, options: Any, provided_label: str | None = None, *, noun: str = ""
+) -> ChoiceVerdict | Refusal:
+    """Match the solver's proven value to the options; confirm or contradict a provided key.
+
+    Returns a :class:`ChoiceVerdict` (``consistent`` / ``contradiction``) when the value ties to
+    exactly one option, else a typed :class:`Refusal` (``no_options`` / ``unparseable_option`` /
+    ``no_matching_option`` / ``ambiguous_options`` / ``unknown_provided_label``).
+    """
+    parsed = parse_options(options)
+    if isinstance(parsed, Refusal):
+        return parsed
+    matches = sorted(label for label, value in parsed.items() if value == computed_value)
+    if not matches:
+        return Refusal("no_matching_option", f"no option equals {computed_value}")
+    if len(matches) > 1:
+        return Refusal("ambiguous_options", f"{matches} all equal {computed_value}")
+
+    computed_label = matches[0]
+    suffix = _suffix(noun)
+    if provided_label is None or provided_label == computed_label:
+        return ChoiceVerdict(
+            computed_value,
+            computed_label,
+            provided_label,
+            "consistent",
+            f"The mathematically consistent answer is {computed_label}. {computed_value}{suffix}.",
+        )
+    if provided_label not in parsed:
+        return Refusal("unknown_provided_label", str(provided_label))
+    return ChoiceVerdict(
+        computed_value,
+        computed_label,
+        provided_label,
+        "contradiction",
+        f"The mathematically consistent answer is {computed_label} ({computed_value}{suffix}). "
+        f"The supplied answer key says {provided_label} ({parsed[provided_label]}{suffix}), "
+        f"which contradicts the equations.",
+    )
+
+
+__all__ = ["ChoiceVerdict", "VERDICT_STATUSES", "verify_answer_choice"]

generate/constraint_comprehension/__init__.py

@@ -24,6 +24,12 @@ from generate.constraint_comprehension.model import (
     Domain,
     Unknown,
 )
+from generate.constraint_comprehension.solver import (
+    answer_constraint_problem,
+    solve_constraint_problem,
+    solve_two_var_count_weight,
+    solve_two_var_linear,
+)
 
 __all__ = [
     "AttributeFact",
@@ -34,4 +40,8 @@ __all__ = [
     "LinearExpr",
     "Relation",
     "Unknown",
+    "answer_constraint_problem",
+    "solve_constraint_problem",
+    "solve_two_var_count_weight",
+    "solve_two_var_linear",
 ]

generate/constraint_comprehension/solver.py

@@ -0,0 +1,106 @@
+"""Independent exact integer solver for the R2 two-variable linear system.
+
+Solves a two-variable, two-equation integer linear system by **exact Cramer's rule** — no
+floats, no nearest-option snapping. The R2 analogue of the relational-metric answer oracle:
+an independent decision procedure that consumes the *structured* constraints, never the text.
+
+Refusal-first (the wrong=0 boundary). The four ways a count/weight system has no honest
+nonnegative-integer answer each REFUSE with a typed reason, never a guessed value:
+
+- ``indistinguishable_weights`` — the system is singular (``det == 0``): the two equations
+  cannot separate the unknowns (e.g. equal per-category coefficients), so no unique solution.
+- ``non_integer_solution``      — Cramer's numerator is not divisible by the determinant:
+  no integer solution exists; the solver refuses rather than round.
+- ``negative_solution``         — a solved value is negative: invalid in the count domain.
+- ``verification_failed``       — a defensive re-substitution backstop (an algebraic identity
+  for the closed-form Cramer solution, so unreachable while the derivation is correct; retained
+  as a structural guard against future edits, NOT claimed as an independently-triggerable gate).
+
+The convenience ``solve_two_var_count_weight`` is the canonical ``x + y = N`` /
+``a·x + b·y = T`` specialization; ``solve_constraint_problem`` / ``answer_constraint_problem``
+drive it from a typed :class:`ConstraintProblem`. Off-serving: imports no
+``generate.derivation`` / ``core.reliability_gate``. Deterministic; no clock, no randomness.
+"""
+
+from __future__ import annotations
+
+from generate.constraint_comprehension.expr import LinearConstraint, LinearExpr
+from generate.constraint_comprehension.model import ConstraintProblem
+from generate.meaning_graph.reader import Refusal
+
+
+def _coeffs(constraint: LinearConstraint, x: str, y: str) -> tuple[int, int, int]:
+    """``(coeff_x, coeff_y, rhs - lhs_constant)`` for ``constraint`` over the variables x, y."""
+    cx = cy = 0
+    for symbol, coeff in constraint.lhs.terms:
+        if symbol == x:
+            cx += coeff
+        elif symbol == y:
+            cy += coeff
+    return cx, cy, constraint.rhs - constraint.lhs.constant
+
+
+def solve_two_var_linear(
+    c0: LinearConstraint, c1: LinearConstraint, *, nonnegative: bool = True
+) -> dict[str, int] | Refusal:
+    """Solve a 2-variable, 2-equation integer system over the SAME two symbols by Cramer's rule.
+
+    Precondition (guaranteed upstream by the C2 setup validator / the reader): both constraints
+    are ``eq`` over exactly two shared symbols. Returns ``{symbol: value}`` or a typed
+    :class:`Refusal` carrying one of the four solver reasons.
+    """
+    symbols = sorted({s for c in (c0, c1) for s, _ in c.lhs.terms})
+    if len(symbols) != 2:  # contract violation — upstream must guarantee two variables
+        raise ValueError(f"solver expects exactly two variables; got {symbols}")
+    x, y = symbols
+    p, q, r0 = _coeffs(c0, x, y)
+    r, s, r1 = _coeffs(c1, x, y)
+
+    det = p * s - q * r
+    if det == 0:
+        return Refusal("indistinguishable_weights", f"singular system over {x}/{y}")
+    num_x = r0 * s - q * r1
+    num_y = p * r1 - r0 * r
+    if num_x % det != 0 or num_y % det != 0:
+        return Refusal("non_integer_solution", f"no integer solution for {x}/{y}")
+    vx, vy = num_x // det, num_y // det
+    if nonnegative and (vx < 0 or vy < 0):
+        return Refusal("negative_solution", f"{x}={vx}, {y}={vy}")
+    if p * vx + q * vy != r0 or r * vx + s * vy != r1:  # pragma: no cover - identity backstop
+        return Refusal("verification_failed", "solution failed re-substitution")
+    return {x: vx, y: vy}
+
+
+def solve_two_var_count_weight(
+    x: str, y: str, total_count: int, x_weight: int, y_weight: int, weighted_total: int
+) -> dict[str, int] | Refusal:
+    """The canonical specialization: ``x + y = total_count`` and
+    ``x_weight·x + y_weight·y = weighted_total``. ``x`` / ``y`` are the symbol names."""
+    count = LinearConstraint(LinearExpr(((x, 1), (y, 1))), "eq", total_count)
+    weighted = LinearConstraint(LinearExpr(((x, x_weight), (y, y_weight))), "eq", weighted_total)
+    return solve_two_var_linear(count, weighted)
+
+
+def solve_constraint_problem(problem: ConstraintProblem) -> dict[str, int] | Refusal:
+    """Solve a two-constraint :class:`ConstraintProblem`'s system (order-independent)."""
+    if len(problem.constraints) != 2:  # contract violation — upstream guarantees two
+        raise ValueError(f"solver expects exactly two constraints; got {len(problem.constraints)}")
+    return solve_two_var_linear(problem.constraints[0], problem.constraints[1])
+
+
+def answer_constraint_problem(problem: ConstraintProblem) -> int | Refusal:
+    """Solve, then project to the asked unknown's value (or propagate the refusal)."""
+    solution = solve_constraint_problem(problem)
+    if isinstance(solution, Refusal):
+        return solution
+    if problem.query.symbol not in solution:  # pragma: no cover - query is a category (C2)
+        return Refusal("query_target_unsolved", problem.query.symbol)
+    return solution[problem.query.symbol]
+
+
+__all__ = [
+    "answer_constraint_problem",
+    "solve_constraint_problem",
+    "solve_two_var_count_weight",
+    "solve_two_var_linear",
+]

tests/test_answer_choices.py

@@ -0,0 +1,84 @@
+"""Tests for the R2 multiple-choice verifier (C4).
+
+Pins the truth-discipline behavior: a proven value ties to exactly one option (else refuse),
+and a disagreeing key is flagged as a CONTRADICTION (a confident verdict, not a refusal). Ties
+to the C2 gold + C3 solver end-to-end: every solved fixture solves, ties to its labeled answer,
+and confirms consistent.
+"""
+
+from __future__ import annotations
+
+from evals.constraint_oracle.runner import _load_r2_gold, gold_to_problem
+from generate.answer_choices.parse import parse_option_value, parse_options
+from generate.answer_choices.verify import ChoiceVerdict, verify_answer_choice
+from generate.constraint_comprehension.solver import answer_constraint_problem
+from generate.meaning_graph.reader import Refusal
+
+
+def _solved() -> list[dict]:
+    return [f for f in _load_r2_gold() if f["expect"] == "solved"]
+
+
+def test_parse_option_value_int_and_string() -> None:
+    assert parse_option_value(11) == 11
+    assert parse_option_value("11") == 11
+    assert parse_option_value("11 chickens") == 11
+    assert parse_option_value("$11") == 11
+    assert parse_option_value("between 5 and 10") is None  # two integers -> ambiguous
+    assert parse_option_value(True) is None  # a bool is not a count
+
+
+def test_parse_options_refuses_empty_and_unparseable() -> None:
+    assert isinstance(parse_options({}), Refusal)
+    assert isinstance(parse_options({"A": "lots"}), Refusal)
+    assert parse_options({"A": 2, "B": "3 buses"}) == {"A": 2, "B": 3}
+
+
+def test_every_solved_gold_key_is_consistent() -> None:
+    for fx in _solved():
+        v = verify_answer_choice(fx["gold"], fx["options"], fx["answer"])
+        assert isinstance(v, ChoiceVerdict), fx["id"]
+        assert v.status == "consistent"
+        assert v.computed_label == fx["answer"]
+
+
+def test_solve_then_verify_end_to_end() -> None:
+    # The full off-serving chain that the reader (C5+) will feed: solve -> tie to the option.
+    for fx in _solved():
+        computed = answer_constraint_problem(gold_to_problem(fx))
+        v = verify_answer_choice(computed, fx["options"], fx["answer"], noun=fx["query"]["unit"])
+        assert isinstance(v, ChoiceVerdict) and v.status == "consistent"
+        assert v.computed_value == fx["gold"] and v.computed_label == fx["answer"]
+
+
+def test_disagreeing_key_is_flagged_as_contradiction() -> None:
+    # chickens: proven 11 == option A; a key of "D" (13) contradicts the equations.
+    fx = next(f for f in _solved() if f["id"] == "r2-002-chickens")
+    v = verify_answer_choice(11, fx["options"], "D", noun="animals")
+    assert isinstance(v, ChoiceVerdict)
+    assert v.status == "contradiction"
+    assert v.computed_label == "A" and v.provided_label == "D"
+    # The message names BOTH the consistent answer and the contradicted key.
+    assert "A" in v.message and "11" in v.message and "D" in v.message and "13" in v.message
+    assert "contradicts" in v.message
+
+
+def test_no_matching_option_refuses() -> None:
+    out = verify_answer_choice(99, {"A": 2, "B": 3, "C": 4}, "A")
+    assert isinstance(out, Refusal) and out.reason == "no_matching_option"
+
+
+def test_ambiguous_duplicate_options_refuse() -> None:
+    out = verify_answer_choice(4, {"A": 4, "B": 4}, None)
+    assert isinstance(out, Refusal) and out.reason == "ambiguous_options"
+
+
+def test_unknown_provided_label_refuses() -> None:
+    out = verify_answer_choice(4, {"A": 2, "B": 4}, "Z")
+    assert isinstance(out, Refusal) and out.reason == "unknown_provided_label"
+
+
+def test_consistent_without_a_provided_key_still_labels() -> None:
+    v = verify_answer_choice(4, {"A": 2, "B": 4}, None, noun="buses")
+    assert isinstance(v, ChoiceVerdict) and v.status == "consistent"
+    assert v.computed_label == "B" and "4 buses" in v.message

tests/test_constraint_solver.py

@@ -0,0 +1,98 @@
+"""Tests for the R2 exact integer solver (C3).
+
+Ties the solver to the C2 gold: every ``solved`` fixture computes its ``gold`` and every
+``solver_refuses`` fixture refuses with EXACTLY the reason the gold claims (so the gold's
+stated refusal reason is not just an annotation — the independent solver agrees). Each of the
+three reachable refusals is proven meaningful-fail, and every solution is re-substituted into
+its constraints (the verification backstop, exercised positively).
+"""
+
+from __future__ import annotations
+
+from evals.constraint_oracle.runner import _load_r2_gold, gold_to_problem
+from evals.constraint_oracle.signature import canonical_constraint
+from generate.constraint_comprehension.solver import (
+    answer_constraint_problem,
+    solve_constraint_problem,
+    solve_two_var_count_weight,
+    solve_two_var_linear,
+)
+from generate.meaning_graph.reader import Refusal
+
+
+def _solved() -> list[dict]:
+    return [f for f in _load_r2_gold() if f["expect"] == "solved"]
+
+
+def _solver_refuses() -> list[dict]:
+    return [f for f in _load_r2_gold() if f["expect"] == "solver_refuses"]
+
+
+def test_solver_solves_every_solved_gold_to_its_gold_value() -> None:
+    for fx in _solved():
+        problem = gold_to_problem(fx)
+        got = answer_constraint_problem(problem)
+        assert got == fx["gold"], f"{fx['id']}: got {got!r}, gold {fx['gold']!r}"
+
+
+def test_solver_solution_satisfies_both_constraints() -> None:
+    # The verification backstop, exercised positively: the solved values re-substitute exactly.
+    for fx in _solved():
+        problem = gold_to_problem(fx)
+        sol = solve_constraint_problem(problem)
+        assert isinstance(sol, dict), fx["id"]
+        for c in problem.constraints:
+            terms, _rel, rhs = canonical_constraint(c)
+            assert sum(coeff * sol[s] for s, coeff in terms) == rhs
+
+
+def test_solver_refuses_every_solver_refuse_gold_with_its_claimed_reason() -> None:
+    for fx in _solver_refuses():
+        problem = gold_to_problem(fx)
+        got = answer_constraint_problem(problem)
+        assert isinstance(got, Refusal), f"{fx['id']} should refuse"
+        assert got.reason == fx["solver_reason"], f"{fx['id']}: {got.reason} != {fx['solver_reason']}"
+
+
+def test_count_weight_convenience_matches_buses() -> None:
+    assert solve_two_var_count_weight("large_bus", "small_bus", 6, 50, 30, 260) == {
+        "large_bus": 4,
+        "small_bus": 2,
+    }
+
+
+def test_solver_is_constraint_order_independent() -> None:
+    fx = next(f for f in _solved() if f["id"] == "r2-002-chickens")
+    p = gold_to_problem(fx)
+    swapped = solve_two_var_linear(p.constraints[1], p.constraints[0])
+    assert swapped == solve_two_var_linear(p.constraints[0], p.constraints[1]) == {"chicken": 11, "cow": 7}
+
+
+# --- meaningful-fail: each reachable refusal fires under exactly its violation --------- #
+
+
+def test_indistinguishable_weights_refuses() -> None:
+    # Equal coefficients -> singular system -> no unique solution.
+    out = solve_two_var_count_weight("car", "truck", 8, 4, 4, 32)
+    assert isinstance(out, Refusal) and out.reason == "indistinguishable_weights"
+
+
+def test_non_integer_solution_refuses() -> None:
+    # 3*pen + 5*notebook = 37, pen+notebook=10 -> pen = 6.5: refuse, never round.
+    out = solve_two_var_count_weight("pen", "notebook", 10, 3, 5, 37)
+    assert isinstance(out, Refusal) and out.reason == "non_integer_solution"
+
+
+def test_negative_solution_refuses() -> None:
+    # 50*large + 30*small = 400, large+small=6 -> small=-5: refuse.
+    out = solve_two_var_count_weight("large_bus", "small_bus", 6, 50, 30, 400)
+    assert isinstance(out, Refusal) and out.reason == "negative_solution"
+
+
+def test_exact_integer_path_is_not_rounded() -> None:
+    # A near-miss that would round to a plausible integer: 3x+5y=38, x+y=10 -> x=6 exactly.
+    # (Guards that the solver computes exactly, not by snapping 37/38/39 to the same answer.)
+    assert solve_two_var_count_weight("x", "y", 10, 3, 5, 38) == {"x": 6, "y": 4}
+    assert isinstance(
+        solve_two_var_count_weight("x", "y", 10, 3, 5, 37), Refusal
+    )  # one less dollar -> no integer split