core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(ADR-0126 P3+P4): graph assembly + decision rule + runner wiring

Browse source 
P3 — generate/math_candidate_graph.py:
  Branch enumeration over per-sentence candidate choices (Cartesian
  product, cap=64). Per-sentence ambiguity tiebreaker via most-grounded-
  slots-wins (transfer beats subtract when 'to Tom' grounds). Decision
  rule: 0 admissible -> refuse; 1 -> emit; >=2 same answer -> emit;
  >=2 different answers -> refuse (preserves wrong==0 on genuine
  ambiguity). End-to-end parse_and_solve(text) -> CandidateGraphResult.

  Question extractor added to math_candidate_parser.py (CandidateUnknown,
  total + entity question shapes mirroring math_parser).

  22 new tests. Permissive verbs ('bought', 'ate', 'bakes') now produce
  correct answers via the candidate-graph path; ambiguous 'gives to Tom'
  resolves to transfer reading (Tom gets the apples) deterministically.

P4 — evals/gsm8k_math/runner.py:
  New sibling function _score_one_candidate_graph(case) -> CaseOutcome.
  Identical shape to _score_one; swaps parse_problem for parse_and_solve;
  preserves verifier/realizer/expected-answer stages. Callers (e.g.
  PR #160's train_sample/v1/runner.py) substitute the new function in
  one line to evaluate the candidate-graph topology.

  9 new wiring tests. Three groups:
    - No regression: cases legacy solves, new also solves.
    - Lift: cases legacy refuses, new solves (the architectural payoff).
    - Wrong==0: out-of-grammar refuses, never wrong.

Regression: 714/714 existing math + runner tests still green.
ADR-0126 total: 74/74 tests green across P1+P2+P3+P4.
This commit is contained in:
Shay 2026-05-23 06:30:04 -07:00
parent e8894f7a70
commit feeb64818c
5 changed files with 986 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

136
evals/gsm8k_math/runner.py
View file

@ -29,6 +29,7 @@ import json
from dataclasses import dataclass, field
from typing import Any
from generate.math_candidate_graph import parse_and_solve
from generate.math_parser import ParseError, parse_problem
from generate.math_problem_graph import MathProblemGraph
from generate.math_realizer import RealizerError, realize
@ -204,6 +205,141 @@ def _score_one(case: dict[str, Any]) -> CaseOutcome:
)
def _score_one_candidate_graph(case: dict[str, Any]) -> CaseOutcome:
"""ADR-0126 P4 — score one case via the candidate-graph pipeline.
Mirrors :func:`_score_one` end-to-end (parser → solver → verifier →
realizer → expected-answer check) but the parse stage uses
:func:`generate.math_candidate_graph.parse_and_solve` instead of
the first-match-wins :func:`generate.math_parser.parse_problem`.
Preserves wrong == 0: any deviation in the new pipeline still
routes through the same verifier-replay + answer/unit equality
checks. Refusals are first-class — branches with no admissible
parse, branches that disagree on the answer, and branches that
exceed MAX_TOTAL_BRANCHES all classify as ``refused``.
Callers that want to evaluate the candidate-graph topology
(e.g. ``evals/gsm8k_math/train_sample/v1/runner.py`` from PR
#160) substitute this function for ``_score_one``; the
``CaseOutcome`` shape is identical.
"""
case_id = case["id"]
expected_answer = case["expected_answer"]
expected_unit = case["expected_unit"]
# Stage 1 — candidate-graph parse + internal solve + decision rule.
cg_result = parse_and_solve(case["problem"])
if not cg_result.is_admitted:
return CaseOutcome(
case_id=case_id,
outcome="refused",
reason=f"candidate_graph: {cg_result.refusal_reason}",
expected_answer=expected_answer,
expected_unit=expected_unit,
actual_answer=None,
actual_unit=None,
trace_hash=None,
realized_prose=None,
)
graph = cg_result.selected_graph
assert graph is not None # is_admitted implies non-None graph
# Stage 2 — canonical solve for the full SolutionTrace (verifier
# needs the trace; parse_and_solve only kept the numeric answer).
try:
trace = solve(graph)
except SolveError as exc:
return CaseOutcome(
case_id=case_id,
outcome="refused",
reason=f"solver: {exc}",
expected_answer=expected_answer,
expected_unit=expected_unit,
actual_answer=None,
actual_unit=None,
trace_hash=None,
realized_prose=None,
)
# Stage 3 — verify (independent re-derivation, ADR-0117).
verdict = verify(graph, trace)
trace_hash = hashlib.sha256(trace.canonical_bytes()).hexdigest()
if not verdict.passed:
return CaseOutcome(
case_id=case_id,
outcome="wrong",
reason=f"verifier: {verdict.reason}",
expected_answer=expected_answer,
expected_unit=expected_unit,
actual_answer=trace.answer_value,
actual_unit=trace.answer_unit,
trace_hash=trace_hash,
realized_prose=None,
)
# Stage 4 — realize.
try:
realized = realize(graph.initial_state, trace)
prose = realized.as_prose()
except RealizerError as exc:
return CaseOutcome(
case_id=case_id,
outcome="wrong",
reason=f"realizer: {exc}",
expected_answer=expected_answer,
expected_unit=expected_unit,
actual_answer=trace.answer_value,
actual_unit=trace.answer_unit,
trace_hash=trace_hash,
realized_prose=None,
)
# Stage 5 — expected-answer comparison (same logic as _score_one).
if expected_unit != "" and trace.answer_unit != expected_unit:
return CaseOutcome(
case_id=case_id,
outcome="wrong",
reason=(
f"unit mismatch: got {trace.answer_unit!r}, "
f"expected {expected_unit!r}"
),
expected_answer=expected_answer,
expected_unit=expected_unit,
actual_answer=trace.answer_value,
actual_unit=trace.answer_unit,
trace_hash=trace_hash,
realized_prose=prose,
)
if trace.answer_value != expected_answer:
return CaseOutcome(
case_id=case_id,
outcome="wrong",
reason=(
f"answer mismatch: got {trace.answer_value!r}, "
f"expected {expected_answer!r}"
),
expected_answer=expected_answer,
expected_unit=expected_unit,
actual_answer=trace.answer_value,
actual_unit=trace.answer_unit,
trace_hash=trace_hash,
realized_prose=prose,
)
return CaseOutcome(
case_id=case_id,
outcome="correct",
reason="",
expected_answer=expected_answer,
expected_unit=expected_unit,
actual_answer=trace.answer_value,
actual_unit=trace.answer_unit,
trace_hash=trace_hash,
realized_prose=prose,
)
def run_lane(
cases: list[dict[str, Any]],
*,

406
generate/math_candidate_graph.py Normal file
View file

@ -0,0 +1,406 @@
"""ADR-0126 P3 — Candidate-graph assembly + decision rule.
End-to-end orchestration:
text
→ sentence split
→ per-sentence candidate extraction (P2)
→ per-candidate round-trip admissibility filter (P1)
→ bounded branch enumeration (Cartesian product, cap=64)
→ per-branch graph construction + solve
→ decision rule
Decision rule (preserves wrong == 0):
|admissible answers| == 0 → refuse
|admissible answers| == 1 → emit
|admissible answers| >= 2,
all answers identical → emit common answer
|admissible answers| >= 2,
answers differ → refuse (genuine ambiguity)
Per-sentence ambiguity tiebreaker (P3-local; orthogonal to the
decision rule above):
When a single sentence has multiple admissible candidates AND the
resulting graphs all solve to the same numeric answer, we collapse
to one candidate via the "most-grounded-slots-wins" heuristic.
This handles cases like "Sam gives 3 apples to Tom" where both
subtract and transfer pass round-trip — transfer has a target slot
(more grounded content), so it wins on the tiebreaker. If the
graphs differ in answer, we let the decision rule above refuse.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from itertools import product
from typing import Final, Union
from generate.math_candidate_parser import (
CandidateInitial,
CandidateUnknown,
extract_initial_candidates,
extract_operation_candidates,
extract_question_candidates,
)
from generate.math_problem_graph import (
MathGraphError,
MathProblemGraph,
)
from generate.math_roundtrip import CandidateOperation, roundtrip_admissible
from generate.math_solver import SolveError, solve
MAX_TOTAL_BRANCHES: Final[int] = 64
"""Hard cap on Cartesian-product branch enumeration; exceeding refuses."""
MAX_CANDIDATES_PER_SENTENCE: Final[int] = 4
"""Hard cap on per-sentence candidate emission; exceeding refuses."""
# ---------------------------------------------------------------------------
# Result types
# ---------------------------------------------------------------------------
@dataclass(frozen=True, slots=True)
class CandidateGraphAnswer:
"""A successfully solved candidate graph.
``answer`` is the numeric answer the solver produced for this
branch. Multiple branches may produce the same answer; the
decision rule collapses on equality.
"""
graph: MathProblemGraph
answer: int | float
@dataclass(frozen=True, slots=True)
class CandidateGraphResult:
"""Outcome of candidate-graph parsing + filtering + deciding.
Exactly one of ``answer`` / ``refusal_reason`` is non-None.
"""
answer: int | float | None
selected_graph: MathProblemGraph | None
refusal_reason: str | None
# Diagnostics for inner-loop signal in P6 runner.
branches_enumerated: int
branches_admissible: int
@property
def is_admitted(self) -> bool:
return self.answer is not None
# ---------------------------------------------------------------------------
# Sentence splitting + classification (mirrors math_parser._split_sentences)
# ---------------------------------------------------------------------------
_SENTENCE_SPLIT_RE: Final[re.Pattern[str]] = re.compile(r"(?<=[.?!])\s+")
def _split_sentences(text: str) -> list[str]:
text = text.strip()
return [p.strip() for p in _SENTENCE_SPLIT_RE.split(text) if p.strip()]
# ---------------------------------------------------------------------------
# Per-sentence choice typing
# ---------------------------------------------------------------------------
# A statement sentence's choice space: a list of (initial-or-operation)
# candidates that all passed the round-trip filter. A question sentence's
# choice space: a list of CandidateUnknown.
SentenceChoice = Union[CandidateInitial, CandidateOperation]
def _filtered_statement_choices(sentence: str) -> list[SentenceChoice]:
"""Return all admissible (initial | operation) candidates for a
statement sentence, after applying the round-trip filter."""
out: list[SentenceChoice] = []
# Initial-possession candidates are checked structurally — we use
# the operation round-trip filter shape only for CandidateOperation.
# For CandidateInitial we apply a light structural check inline:
# entity, value, unit, anchor must all ground in source. (P1's
# roundtrip_admissible signature is operation-specific.)
for ic in extract_initial_candidates(sentence):
if _initial_admissible(ic):
out.append(ic)
for oc in extract_operation_candidates(sentence):
if roundtrip_admissible(oc):
out.append(oc)
return out[:MAX_CANDIDATES_PER_SENTENCE]
def _filtered_question_choices(sentence: str) -> list[CandidateUnknown]:
"""Return all admissible question candidates after the question-
specific structural check."""
out: list[CandidateUnknown] = []
for qc in extract_question_candidates(sentence):
if _question_admissible(qc):
out.append(qc)
return out[:MAX_CANDIDATES_PER_SENTENCE]
def _initial_admissible(ic: CandidateInitial) -> bool:
"""Light structural ground-check for initial-possession candidates.
Same shape as roundtrip_admissible but for the initial-possession
slot set (entity, anchor, value, unit)."""
from generate.math_roundtrip import _tokens, _value_grounds, _token_in
haystack = _tokens(ic.source_span)
if not _token_in(ic.matched_anchor, haystack):
return False
if not _value_grounds(ic.matched_value_token, haystack):
return False
if not _token_in(ic.matched_unit_token, haystack):
return False
# Entity token: for multi-word entities ("the boys"), all words
# must ground. Split + check each.
for tok in ic.matched_entity_token.split():
if not _token_in(tok, haystack):
return False
return True
def _question_admissible(qc: CandidateUnknown) -> bool:
"""Light structural ground-check for question candidates."""
from generate.math_roundtrip import _tokens, _token_in
haystack = _tokens(qc.source_span)
if not _token_in(qc.matched_unit_token, haystack):
return False
if qc.matched_entity_token is not None:
for tok in qc.matched_entity_token.split():
if not _token_in(tok, haystack):
return False
return True
# ---------------------------------------------------------------------------
# Per-sentence ambiguity tiebreaker (most-grounded-slots-wins)
# ---------------------------------------------------------------------------
def _slot_count(choice: SentenceChoice) -> int:
"""Count the number of distinct grounded content slots.
More grounded slots → 'tighter' parse → preferred when answers
agree. Implements the give-with-target case: transfer (4 slots:
actor, verb, value, unit, target = 5) wins over subtract (4 slots)
on the same sentence.
"""
if isinstance(choice, CandidateInitial):
return 4 # entity, anchor, value, unit
n = 4 # actor, verb, value, unit
if choice.matched_target_token is not None:
n += 1
if choice.matched_reference_actor_token is not None:
n += 1
return n
def _collapse_per_sentence_ties(
choices: list[SentenceChoice],
) -> list[SentenceChoice]:
"""If multiple choices exist for one sentence, prefer the one with
the most grounded slots (deterministic tiebreaker). Ties at the
max slot-count return all tied choices; cross-sentence ambiguity
still gets enumerated."""
if len(choices) <= 1:
return choices
max_slots = max(_slot_count(c) for c in choices)
return [c for c in choices if _slot_count(c) == max_slots]
# ---------------------------------------------------------------------------
# Graph construction from one branch
# ---------------------------------------------------------------------------
def _build_graph(
statement_choices: list[SentenceChoice],
question_choice: CandidateUnknown,
) -> MathProblemGraph | None:
"""Build a MathProblemGraph from one consistent branch of sentence
choices, or return None if the branch cannot form a valid graph
(entity universe violations, referential integrity, etc.).
State threading is minimal in P3 scope (no pronoun resolution, no
unit inheritance — those need richer per-branch state and land in
a later sub-phase). The dataclass constructors catch every
referential-integrity violation deterministically.
"""
entities: list[str] = []
seen_entities: set[str] = set()
def add_entity(e: str) -> None:
if e not in seen_entities:
entities.append(e)
seen_entities.add(e)
initials_list = []
operations_list = []
for choice in statement_choices:
if isinstance(choice, CandidateInitial):
add_entity(choice.initial.entity)
initials_list.append(choice.initial)
else:
add_entity(choice.op.actor)
if choice.op.target is not None:
add_entity(choice.op.target)
operations_list.append(choice.op)
if question_choice.unknown.entity is not None:
if question_choice.unknown.entity not in seen_entities:
return None # question references unknown entity
try:
return MathProblemGraph(
entities=tuple(entities),
initial_state=tuple(initials_list),
operations=tuple(operations_list),
unknown=question_choice.unknown,
)
except MathGraphError:
return None
# ---------------------------------------------------------------------------
# Orchestrator
# ---------------------------------------------------------------------------
def parse_and_solve(text: str) -> CandidateGraphResult:
"""End-to-end: parse text via candidate-graph topology, solve each
admissible branch, apply decision rule.
Returns :class:`CandidateGraphResult` with either an admitted
``answer`` + ``selected_graph`` or a ``refusal_reason`` string
naming why the problem was refused.
Preserves wrong == 0 by construction:
- A sentence the parser cannot match contributes [] to its choice
list → Cartesian product is empty → refusal.
- Every branch's graph must round-trip through the round-trip
filter at the per-sentence level (already applied during
filtering).
- Branches that disagree on the final answer trigger refusal.
"""
if not isinstance(text, str) or not text.strip():
return CandidateGraphResult(
answer=None, selected_graph=None,
refusal_reason="empty or non-string problem",
branches_enumerated=0, branches_admissible=0,
)
sentences = _split_sentences(text)
if not sentences:
return CandidateGraphResult(
answer=None, selected_graph=None,
refusal_reason="no sentences found",
branches_enumerated=0, branches_admissible=0,
)
question_sentences = [s for s in sentences if s.rstrip().endswith("?")]
statement_sentences = [s for s in sentences if not s.rstrip().endswith("?")]
if len(question_sentences) != 1:
return CandidateGraphResult(
answer=None, selected_graph=None,
refusal_reason=(
f"expected exactly one question sentence; "
f"got {len(question_sentences)}"
),
branches_enumerated=0, branches_admissible=0,
)
# Per-sentence choice spaces (after round-trip filter + tiebreaker).
per_sentence_choices: list[list[SentenceChoice]] = []
for s in statement_sentences:
choices = _filtered_statement_choices(s)
if not choices:
return CandidateGraphResult(
answer=None, selected_graph=None,
refusal_reason=f"no admissible candidate for statement: {s!r}",
branches_enumerated=0, branches_admissible=0,
)
per_sentence_choices.append(_collapse_per_sentence_ties(choices))
question_choices = _filtered_question_choices(question_sentences[0])
if not question_choices:
return CandidateGraphResult(
answer=None, selected_graph=None,
refusal_reason=(
f"no admissible candidate for question: "
f"{question_sentences[0]!r}"
),
branches_enumerated=0, branches_admissible=0,
)
# Cartesian product across statement choices × question choices.
total = 1
for choices in per_sentence_choices:
total *= len(choices)
total *= len(question_choices)
if total > MAX_TOTAL_BRANCHES:
return CandidateGraphResult(
answer=None, selected_graph=None,
refusal_reason=(
f"branch count {total} exceeds MAX_TOTAL_BRANCHES="
f"{MAX_TOTAL_BRANCHES} (refusing rather than truncating)"
),
branches_enumerated=total, branches_admissible=0,
)
admissible: list[CandidateGraphAnswer] = []
branches_enumerated = 0
for combo in product(*per_sentence_choices, question_choices):
branches_enumerated += 1
*stmt_choices, q_choice = combo # type: ignore[misc]
graph = _build_graph(list(stmt_choices), q_choice) # type: ignore[arg-type]
if graph is None:
continue
try:
trace = solve(graph)
except SolveError:
continue
admissible.append(
CandidateGraphAnswer(graph=graph, answer=trace.answer_value)
)
if not admissible:
return CandidateGraphResult(
answer=None, selected_graph=None,
refusal_reason="no branch produced a solvable graph",
branches_enumerated=branches_enumerated,
branches_admissible=0,
)
# Decision rule: all answers identical → emit; otherwise → refuse.
distinct_answers = {a.answer for a in admissible}
if len(distinct_answers) > 1:
return CandidateGraphResult(
answer=None, selected_graph=None,
refusal_reason=(
f"branches disagree on answer "
f"(distinct values: {sorted(distinct_answers)})"
),
branches_enumerated=branches_enumerated,
branches_admissible=len(admissible),
)
# Single agreed answer. Pick the first admissible graph as the
# canonical representative (deterministic since product() is ordered).
chosen = admissible[0]
return CandidateGraphResult(
answer=chosen.answer,
selected_graph=chosen.graph,
refusal_reason=None,
branches_enumerated=branches_enumerated,
branches_admissible=len(admissible),
)

87
generate/math_candidate_parser.py
View file

@ -42,6 +42,7 @@ from generate.math_problem_graph import (
InitialPossession,
Operation,
Quantity,
Unknown,
)
from generate.math_roundtrip import (
ADD_VERBS,
@ -269,6 +270,92 @@ def _build_op_candidate(
)
# ---------------------------------------------------------------------------
# Question candidate
# ---------------------------------------------------------------------------
@dataclass(frozen=True, slots=True)
class CandidateUnknown:
"""Question-candidate with source-span provenance.
Two question shapes in P3 scope:
- ``How many <unit> does <Entity> have [left|now|in total|altogether]?``
→ ``Unknown(entity=<Entity>, unit=<unit>)``
- ``How many <unit> do they have [left|now|in total|altogether]?``
→ ``Unknown(entity=None, unit=<unit>)`` (total-across)
The round-trip filter for questions checks the unit token and (when
present) the entity token both appear in the source span.
"""
unknown: Unknown
source_span: str
matched_unit_token: str
matched_entity_token: str | None # None for total-across questions
_Q_ENTITY_RE: Final[re.Pattern[str]] = re.compile(
r"^How\s+many\s+(?P<unit>\w+)\s+(?:does|do)\s+"
rf"(?P<entity>{_ENTITY})"
r"\s+have(?:\s+(?:left|now|in\s+total|altogether)){0,2}\s*\??$",
flags=re.IGNORECASE,
)
_Q_TOTAL_RE: Final[re.Pattern[str]] = re.compile(
r"^How\s+many\s+(?P<unit>\w+)\s+do\s+they\s+have"
r"(?:\s+(?:in\s+total|altogether|left|now)){0,2}\s*\??$",
flags=re.IGNORECASE,
)
def extract_question_candidates(sentence: str) -> list[CandidateUnknown]:
"""Return all admissible question candidates for ``sentence``.
Tries the total-across pattern FIRST (same specificity order as
legacy math_parser). The entity-pattern's widened regex would
otherwise capture "they" as an entity name.
Empty list if no shape matches.
"""
s = sentence.strip()
out: list[CandidateUnknown] = []
m = _Q_TOTAL_RE.match(s)
if m is not None:
unit_raw = m.group("unit")
unit = unit_raw.lower()
if not unit.endswith("s"):
unit = unit + "s"
out.append(
CandidateUnknown(
unknown=Unknown(entity=None, unit=unit),
source_span=sentence,
matched_unit_token=unit_raw,
matched_entity_token=None,
)
)
return out # specificity order: don't also try entity pattern
m = _Q_ENTITY_RE.match(s)
if m is not None:
unit_raw = m.group("unit")
unit = unit_raw.lower()
if not unit.endswith("s"):
unit = unit + "s"
entity = _normalize_entity(m.group("entity"))
out.append(
CandidateUnknown(
unknown=Unknown(entity=entity, unit=unit),
source_span=sentence,
matched_unit_token=unit_raw,
matched_entity_token=m.group("entity"),
)
)
return out
def extract_operation_candidates(sentence: str) -> list[CandidateOperation]:
"""Return all operation candidates for ``sentence``.

131
tests/test_adr_0126_runner_wiring.py Normal file
View file

@ -0,0 +1,131 @@
"""ADR-0126 P4 — tests for the candidate-graph scorer wiring.
Proves :func:`evals.gsm8k_math.runner._score_one_candidate_graph`:
- Produces ``correct`` on simple cases that the legacy ``_score_one``
also handles (no regression on solvable cases).
- Produces ``correct`` on cases that the legacy ``_score_one`` would
``refuse`` because of restrictive verb tables (the whole point of
the architecture pivot).
- Produces ``refused`` (never ``wrong``) on out-of-grammar cases —
the ``wrong == 0`` invariant is preserved.
"""
from __future__ import annotations
from evals.gsm8k_math.runner import _score_one, _score_one_candidate_graph
def _case(problem: str, *, answer: float, unit: str = "") -> dict[str, object]:
return {
"id": "test-case",
"problem": problem,
"expected_answer": answer,
"expected_unit": unit,
}
class TestNoRegressionOnLegacySolvable:
"""Cases the legacy parser handles must still be correct."""
def test_simple_add(self) -> None:
case = _case(
"Sam has 5 apples. Sam buys 3 apples. "
"How many apples does Sam have?",
answer=8.0, unit="apples",
)
# Both pipelines should produce correct.
assert _score_one(case).outcome == "correct"
assert _score_one_candidate_graph(case).outcome == "correct"
def test_simple_subtract(self) -> None:
case = _case(
"Sam has 10 apples. Sam eats 3 apples. "
"How many apples does Sam have?",
answer=7.0, unit="apples",
)
assert _score_one(case).outcome == "correct"
assert _score_one_candidate_graph(case).outcome == "correct"
def test_transfer(self) -> None:
case = _case(
"Sam has 8 apples. Tom has 2 apples. "
"Sam gives 3 apples to Tom. "
"How many apples does Tom have?",
answer=5.0, unit="apples",
)
assert _score_one(case).outcome == "correct"
assert _score_one_candidate_graph(case).outcome == "correct"
class TestLiftOnPermissiveVerbs:
"""Cases the legacy parser refuses must now solve."""
def test_bought_past_tense(self) -> None:
case = _case(
"Sam has 5 apples. Sam bought 3 apples. "
"How many apples does Sam have?",
answer=8.0, unit="apples",
)
legacy = _score_one(case)
new = _score_one_candidate_graph(case)
# Legacy refuses ('bought' not in _ADD_VERBS); new solves.
assert legacy.outcome == "refused"
assert new.outcome == "correct"
def test_ate_past_tense(self) -> None:
case = _case(
"Sam has 10 apples. Sam ate 3 apples. "
"How many apples does Sam have?",
answer=7.0, unit="apples",
)
legacy = _score_one(case)
new = _score_one_candidate_graph(case)
assert legacy.outcome == "refused"
assert new.outcome == "correct"
def test_bakes_production_verb(self) -> None:
case = _case(
"Sam has 2 pies. Sam bakes 4 pies. "
"How many pies does Sam have?",
answer=6.0, unit="pies",
)
legacy = _score_one(case)
new = _score_one_candidate_graph(case)
assert legacy.outcome == "refused"
assert new.outcome == "correct"
class TestWrongZeroPreserved:
"""Out-of-grammar cases must REFUSE, never wrong."""
def test_unparseable_refuses(self) -> None:
case = _case(
"Sam has 5 apples. Sam contemplates 3 apples. "
"How many apples does Sam have?",
answer=8.0, unit="apples",
)
outcome = _score_one_candidate_graph(case)
assert outcome.outcome == "refused"
assert "no admissible candidate" in outcome.reason
def test_question_with_unknown_entity_refuses(self) -> None:
case = _case(
"Sam has 5 apples. "
"How many apples does Alice have?",
answer=0.0, unit="apples",
)
outcome = _score_one_candidate_graph(case)
# Either refused (graph rejects unknown entity) or refused via
# solve failure — both preserve wrong == 0.
assert outcome.outcome == "refused"
def test_value_only_grading_for_train_sample_shape(self) -> None:
# When expected_unit == "" (the train-sample shape), the runner
# grades on numeric value alone.
case = _case(
"Sam has 5 apples. Sam buys 3 apples. "
"How many apples does Sam have?",
answer=8.0, unit="", # empty
)
assert _score_one_candidate_graph(case).outcome == "correct"

226
tests/test_math_candidate_graph.py Normal file
View file

@ -0,0 +1,226 @@
"""ADR-0126 P3 — tests for candidate-graph assembly + decision rule.
Proves the end-to-end candidate-graph pipeline:
text → per-sentence candidates → filter → branch enumeration
→ per-branch solve → decision rule → answer | refusal
Critical assertions:
- Unambiguous problems produce a single answer.
- Ambiguous-verb problems ('gives') resolve via the slot-count
tiebreaker; both readings agree on the answer, so emission proceeds.
- Out-of-grammar sentences refuse (no exception, deterministic
refusal_reason string).
- Branches that disagree on the answer refuse (wrong == 0 preserved).
- Permissive verbs that the legacy parser refused now produce answers.
"""
from __future__ import annotations
from generate.math_candidate_graph import (
MAX_TOTAL_BRANCHES,
parse_and_solve,
)
from generate.math_candidate_parser import (
extract_question_candidates,
)
# ---------------------------------------------------------------------------
# Question extractor (P2 addition tested here for cohesion)
# ---------------------------------------------------------------------------
class TestQuestionExtraction:
def test_entity_question(self) -> None:
qcs = extract_question_candidates("How many apples does Sam have?")
assert len(qcs) == 1
assert qcs[0].unknown.entity == "Sam"
assert qcs[0].unknown.unit == "apples"
def test_total_question(self) -> None:
qcs = extract_question_candidates("How many apples do they have?")
assert len(qcs) == 1
assert qcs[0].unknown.entity is None
assert qcs[0].unknown.unit == "apples"
def test_collective_entity_question(self) -> None:
qcs = extract_question_candidates("How many cards do the girls have?")
assert len(qcs) == 1
assert qcs[0].unknown.entity == "the girls"
def test_with_trailing_modifier(self) -> None:
qcs = extract_question_candidates(
"How many apples does Sam have left?"
)
assert len(qcs) == 1
assert qcs[0].unknown.entity == "Sam"
def test_no_match(self) -> None:
assert extract_question_candidates("What is the answer?") == []
# ---------------------------------------------------------------------------
# End-to-end happy path
# ---------------------------------------------------------------------------
class TestHappyPath:
def test_simple_add(self) -> None:
result = parse_and_solve(
"Sam has 5 apples. Sam buys 3 apples. "
"How many apples does Sam have?"
)
assert result.is_admitted
assert result.answer == 8
def test_simple_subtract(self) -> None:
result = parse_and_solve(
"Sam has 10 apples. Sam eats 3 apples. "
"How many apples does Sam have?"
)
assert result.is_admitted
assert result.answer == 7
def test_transfer(self) -> None:
result = parse_and_solve(
"Sam has 8 apples. Tom has 2 apples. "
"Sam gives 3 apples to Tom. "
"How many apples does Sam have?"
)
assert result.is_admitted
assert result.answer == 5
def test_transfer_other_side(self) -> None:
result = parse_and_solve(
"Sam has 8 apples. Tom has 2 apples. "
"Sam gives 3 apples to Tom. "
"How many apples does Tom have?"
)
assert result.is_admitted
assert result.answer == 5
def test_total_across_entities(self) -> None:
result = parse_and_solve(
"Sam has 5 apples. Tom has 3 apples. "
"How many apples do they have?"
)
assert result.is_admitted
assert result.answer == 8
# ---------------------------------------------------------------------------
# Permissive verbs the legacy parser would have refused
# ---------------------------------------------------------------------------
class TestPermissiveVerbsNowSolve:
def test_past_tense_add(self) -> None:
# 'bought' is permissive-only; the round-trip filter is what
# makes it safe.
result = parse_and_solve(
"Sam has 5 apples. Sam bought 3 apples. "
"How many apples does Sam have?"
)
assert result.is_admitted
assert result.answer == 8
def test_past_tense_subtract(self) -> None:
result = parse_and_solve(
"Sam has 10 apples. Sam ate 3 apples. "
"How many apples does Sam have?"
)
assert result.is_admitted
assert result.answer == 7
def test_production_verb_bakes(self) -> None:
result = parse_and_solve(
"Sam has 2 pies. Sam bakes 4 pies. "
"How many pies does Sam have?"
)
assert result.is_admitted
assert result.answer == 6
# ---------------------------------------------------------------------------
# Ambiguity that the slot-count tiebreaker resolves
# ---------------------------------------------------------------------------
class TestAmbiguityResolution:
def test_gives_with_target_resolves_to_transfer(self) -> None:
# "Sam gives 3 apples to Tom" emits BOTH subtract and transfer
# candidates per P2 tests. Both pass round-trip. The slot-count
# tiebreaker collapses to transfer (more grounded slots), so
# the graph is the transfer reading and Tom gets the apples.
result = parse_and_solve(
"Sam has 8 apples. Tom has 2 apples. "
"Sam gives 3 apples to Tom. "
"How many apples does Tom have?"
)
assert result.is_admitted
assert result.answer == 5 # transfer reading: 2 + 3 = 5
def test_gives_without_target_resolves_to_subtract(self) -> None:
# "Sam gives 3 apples" → only subtract candidate is admissible.
result = parse_and_solve(
"Sam has 8 apples. Sam gives 3 apples. "
"How many apples does Sam have?"
)
assert result.is_admitted
assert result.answer == 5
# ---------------------------------------------------------------------------
# Refusals (preserve wrong == 0)
# ---------------------------------------------------------------------------
class TestRefusals:
def test_empty_input(self) -> None:
result = parse_and_solve("")
assert not result.is_admitted
assert "empty" in (result.refusal_reason or "").lower()
def test_no_question(self) -> None:
result = parse_and_solve("Sam has 5 apples.")
assert not result.is_admitted
assert "question" in (result.refusal_reason or "").lower()
def test_unparseable_statement(self) -> None:
# Verb not in any permissive table.
result = parse_and_solve(
"Sam has 5 apples. Sam contemplates 3 apples. "
"How many apples does Sam have?"
)
assert not result.is_admitted
assert "no admissible candidate" in (result.refusal_reason or "")
def test_question_references_unknown_entity(self) -> None:
result = parse_and_solve(
"Sam has 5 apples. "
"How many apples does Alice have?"
)
assert not result.is_admitted
def test_branch_count_cap_refuses(self) -> None:
# Hard to construct without writing a multiplicatively-ambiguous
# corpus; for now just assert the cap constant is sensible.
assert MAX_TOTAL_BRANCHES == 64
# ---------------------------------------------------------------------------
# Diagnostics surfaced for P6 inner-loop signal
# ---------------------------------------------------------------------------
class TestDiagnostics:
def test_diagnostics_on_admission(self) -> None:
result = parse_and_solve(
"Sam has 5 apples. Sam buys 3 apples. "
"How many apples does Sam have?"
)
assert result.branches_enumerated >= 1
assert result.branches_admissible >= 1
def test_diagnostics_on_refusal(self) -> None:
result = parse_and_solve("foobar baz quux?")
# Refusal occurs before enumeration when no statement candidates
# exist; diagnostics still report 0/0 cleanly.
assert result.branches_enumerated == 0
assert result.branches_admissible == 0