fix(gsm8k): harden sprint 9 temporal and affine gates

Patch post-merge wrong=0 hazards from #820 review without reverting
Sprint 9 organs: bind affine/temporal answers to asked subjects and body
actors, anchor overtime thresholds to per-shift/day surfaces, fix affine
completeness double-counting, narrow percent_partition group skip, and
expand temporal question/rental verb recognition.
This commit is contained in:
Shay 2026-06-18 10:03:35 -07:00
parent 63a9bea823
commit 13265e1580
4 changed files with 291 additions and 8 deletions

View file

@ -33,6 +33,7 @@ from collections import Counter
from generate.derivation.clauses import segment_clauses
from generate.derivation.extract import extract_quantities
from generate.derivation.model import GroundedDerivation, Quantity, Step
from generate.derivation.state.bind import PRONOUNS, leading_subject_token
from generate.derivation.target import _question_clause
from generate.derivation.verify import Resolution, SelfVerification
from generate.math_candidate_parser import _init_mutation_admitted
@ -51,6 +52,10 @@ _GOAL_INTENT: Final[frozenset[str]] = frozenset(
{"want", "wants", "wanted", "need", "needs", "hoping", "hopes", "plans", "aims", "goal"}
)
_POSSESSION_CUES: Final[frozenset[str]] = frozenset({"has", "have", "had"})
_QUESTION_POSSESSION_RE: Final[re.Pattern[str]] = re.compile(
r"how\s+many\b.+?\bdoes\s+(\w+)\s+have\b",
re.IGNORECASE,
)
def _asks_subject_total(question_clause: str) -> bool:
@ -65,6 +70,43 @@ def _affine_match(problem_text: str) -> re.Match[str] | None:
return matches[0]
def _affine_clause(problem_text: str) -> str | None:
match = _affine_match(problem_text)
if match is None:
return None
for clause in segment_clauses(problem_text):
if match.group(0) in clause:
return clause
return None
def _affine_subject(problem_text: str) -> str | None:
clause = _affine_clause(problem_text)
if clause is None:
return None
subject = leading_subject_token(clause)
return subject.lower() if subject is not None else None
def _question_possession_subject(question_clause: str) -> str | None:
match = _QUESTION_POSSESSION_RE.search(question_clause)
if match is None:
return None
return match.group(1).lower()
def _question_subject_matches_affine(
problem_text: str, question_clause: str
) -> bool:
affine_subject = _affine_subject(problem_text)
asked_subject = _question_possession_subject(question_clause)
if affine_subject is None or asked_subject is None:
return False
if asked_subject in PRONOUNS:
return False
return asked_subject == affine_subject
def _has_hazard_surface(problem_text: str, question_clause: str) -> bool:
if _DECREASE_TO_FRACTION_RE.search(problem_text):
return True
@ -118,6 +160,8 @@ def build_affine_fraction_delta(problem_text: str) -> GroundedDerivation | None:
question_clause = _question_clause(problem_text)
if not _asks_subject_total(question_clause):
return None
if not _question_subject_matches_affine(problem_text, question_clause):
return None
if _has_hazard_surface(problem_text, question_clause):
return None
@ -202,8 +246,6 @@ def _self_verifies_affine_fraction_delta(
+ [step.operand.source_token for step in derivation.steps]
)
if match is not None:
used[match.group(4)] += 1
used[f"{match.group(1)}/{match.group(2)}"] += 1
used[match.group(1)] += 1
used[match.group(2)] += 1
reference_candidate = _reference_candidate(problem_text, match.group(3))

View file

@ -82,7 +82,7 @@ def _total_population(problem_text: str, question_clause: str) -> Quantity | Non
lowered = clause.lower()
if "half" in lowered or _PERCENT_OF_GROUP_RE.search(clause):
continue
if "group" in lowered:
if "group" in lowered and "group of" not in lowered:
continue
quantities = [q for q in extract_quantities(clause) if q.unit]
if len(quantities) == 1:

View file

@ -31,6 +31,7 @@ from collections import Counter
from generate.derivation.clauses import segment_clauses
from generate.derivation.extract import extract_quantities
from generate.derivation.model import GroundedDerivation, Quantity, Step
from generate.derivation.state.bind import PRONOUNS, leading_subject_token
from generate.derivation.target import _question_clause
from generate.derivation.verify import Resolution, SelfVerification
from generate.math_roundtrip import _token_in, _tokens, _value_grounds
@ -40,10 +41,25 @@ _HOURLY_RATE_RE: Final[re.Pattern[str]] = re.compile(
r"\$\s*(\d+(?:\.\d+)?)\s*(?:an\s+hour|/hour|per\s+hour)",
re.IGNORECASE,
)
_THRESHOLD_HOURS_RE: Final[re.Pattern[str]] = re.compile(
r"more\s+than\s+(\d+)\s+hours?",
_THRESHOLD_SHIFT_RE: Final[re.Pattern[str]] = re.compile(
r"more\s+than\s+(\d+)\s+hours?\s+(?:a|an|each|per)\s+(?:day|shift|workday)",
re.IGNORECASE,
)
_THRESHOLD_BAD_ANCHOR_RE: Final[re.Pattern[str]] = re.compile(
r"more\s+than\s+\d+\s+hours?\s+(?:per|a|an|each|for)\s+(?:week|month|year|total)",
re.IGNORECASE,
)
_RENT_VERBS: Final[frozenset[str]] = frozenset({"rent", "rents", "rented", "renting"})
_QUESTION_MONEY_SUBJECT_RE: Final[tuple[re.Pattern[str], ...]] = (
re.compile(
r"how\s+much(?:\s+money)?\s+does\s+(\w+)\s+(?:make|earn)",
re.IGNORECASE,
),
re.compile(r"how\s+much\s+will\s+it\s+cost\s+(\w+)", re.IGNORECASE),
re.compile(r"what\s+does\s+it\s+cost\s+(\w+)", re.IGNORECASE),
re.compile(r"how\s+much\s+does\s+(\w+)\s+pay", re.IGNORECASE),
re.compile(r"what\s+is\s+the\s+total\s+cost(?:\s+for\s+(\w+))?", re.IGNORECASE),
)
_BUNDLE_TARIFF_RE: Final[re.Pattern[str]] = re.compile(
r"\$\s*(\d+(?:\.\d+)?)\s+per\s+day\s+or\s+\$\s*(\d+(?:\.\d+)?)\s+for\s+(\d+)\s+days?",
re.IGNORECASE,
@ -71,11 +87,86 @@ def _asks_money_total(question_clause: str) -> bool:
tokens = _tokens(question_clause)
if "money" in tokens and bool(_EARN_VERBS & tokens):
return True
if "how" in tokens and "much" in tokens and bool(_COST_VERBS & tokens):
if "how" in tokens and "much" in tokens and bool((_COST_VERBS | _EARN_VERBS) & tokens):
return True
if "what" in tokens and bool({"cost", "costs", "charge", "charges", "price", "pay", "paid"} & tokens):
return True
return False
def _question_money_subject(question_clause: str) -> str | None:
for pattern in _QUESTION_MONEY_SUBJECT_RE:
match = pattern.search(question_clause)
if match is None:
continue
subject = match.group(1)
if subject is None:
continue
return subject.lower()
return None
def _subjects_match(body_actor: str | None, asked_subject: str | None) -> bool:
if body_actor is None or asked_subject is None:
return False
if asked_subject in PRONOUNS:
return True
return asked_subject == body_actor.lower()
def _overtime_worker(problem_text: str, question_clause: str) -> str | None:
for clause in segment_clauses(problem_text):
if clause == question_clause:
continue
if _HOURLY_RATE_RE.search(clause) or re.search(
r"makes\s+\$\s*\d", clause, re.IGNORECASE
):
subject = leading_subject_token(clause)
if subject is not None:
return subject.lower()
for clause in segment_clauses(problem_text):
if clause == question_clause:
continue
tokens = _tokens(clause)
if "hours" not in tokens and "hour" not in tokens:
continue
if "every" not in tokens and "per" not in tokens and "each" not in tokens:
continue
subject = leading_subject_token(clause)
if subject is not None and subject.lower() not in {"if", "when"}:
return subject.lower()
return None
def _bundle_renter(problem_text: str, question_clause: str) -> str | None:
for clause in segment_clauses(problem_text):
if clause == question_clause:
continue
tokens = _tokens(clause)
if not (_RENT_VERBS & tokens):
continue
subject = leading_subject_token(clause)
if subject is not None:
return subject.lower()
return None
def _question_target_matches_actor(
problem_text: str, question_clause: str, *, pattern: str
) -> bool:
asked = _question_money_subject(question_clause)
if pattern == "overtime":
body_actor = _overtime_worker(problem_text, question_clause)
else:
body_actor = _bundle_renter(problem_text, question_clause)
if body_actor is None:
return False
if asked is None:
tokens = _tokens(question_clause)
return "what" in tokens and "total" in tokens and "cost" in tokens
return _subjects_match(body_actor, asked)
def _body_text(problem_text: str) -> str:
question_clause = _question_clause(problem_text)
return problem_text.replace(question_clause, "").strip()
@ -113,7 +204,9 @@ def _hourly_rate(problem_text: str) -> Quantity | None:
def _threshold_hours(problem_text: str) -> Quantity | None:
match = _THRESHOLD_HOURS_RE.search(problem_text)
if _THRESHOLD_BAD_ANCHOR_RE.search(problem_text):
return None
match = _THRESHOLD_SHIFT_RE.search(problem_text)
if match is None:
return None
token = match.group(1)
@ -167,7 +260,7 @@ def _rental_days(problem_text: str, question_clause: str) -> Quantity | None:
if clause == question_clause:
continue
tokens = _tokens(clause)
if "rent" not in tokens and "rents" not in tokens:
if not (_RENT_VERBS & tokens):
continue
quantities = [
q
@ -201,6 +294,8 @@ def build_overtime_shift_earnings(problem_text: str) -> GroundedDerivation | Non
question_clause = _question_clause(problem_text)
if not _asks_money_total(question_clause):
return None
if not _question_target_matches_actor(problem_text, question_clause, pattern="overtime"):
return None
if _has_hazard_surface(problem_text, question_clause):
return None
if "overtime" not in _tokens(problem_text):
@ -243,6 +338,8 @@ def build_bundle_overflow_tariff(problem_text: str) -> GroundedDerivation | None
question_clause = _question_clause(problem_text)
if not _asks_money_total(question_clause):
return None
if not _question_target_matches_actor(problem_text, question_clause, pattern="bundle"):
return None
if _has_hazard_surface(problem_text, question_clause):
return None
if "overtime" in _tokens(problem_text):

View file

@ -0,0 +1,144 @@
"""Post-merge wrong=0 hardening for Sprint 9 Gates A2m/A2n (#820)."""
from __future__ import annotations
from generate.math_candidate_graph import parse_and_solve
from generate.derivation.affine_fraction_delta import (
compose_affine_fraction_delta,
resolve_promotable_affine_fraction_delta,
)
from generate.derivation.temporal_tariff import (
compose_bundle_overflow_tariff,
compose_overtime_shift_earnings,
compose_temporal_tariff,
resolve_promotable_temporal_tariff,
)
from generate.derivation.percent_partition import (
compose_percent_partition,
resolve_promotable_percent_partition,
)
_AFFINE_BODY = (
"Yun had 20 paperclips initially, but then lost 12. Marion has 1/4 more than what "
"Yun currently has, plus 7."
)
_OVERTIME_BODY = (
"Tina makes $18.00 an hour. If she works more than 8 hours per shift, she is "
"eligible for overtime, which is paid by your hourly wage + 1/2 your hourly wage. "
"If she works 10 hours every day for 5 days,"
)
_BUNDLE_BODY = (
"Jason has a carriage house that he rents out. He's charging $50.00 per day or "
"$500.00 for 14 days. Eric wants to rent the house for 20 days."
)
_PERCENT_GROUP_OF_100 = (
"A group of 100 students. Half of the students are girls, the other half are boys. "
"20% of the girls have dogs and 10% of the boys have dogs. "
"How many students own dogs?"
)
_PERCENT_SUBGROUP_CONFUSER = (
"Half of the students are girls, the other half are boys. The girls group has 50 students. "
"20% of the girls have dogs and 10% of the boys have dogs. How many students own dogs?"
)
def _run(text: str):
return parse_and_solve(text, sealed=False)
class TestAffineSubjectBinding:
def test_question_subject_mismatch_refuses(self):
text = f"{_AFFINE_BODY} How many paperclips does Alice have?"
assert resolve_promotable_affine_fraction_delta(text) is None
assert _run(text).answer is None
def test_question_reference_subject_refuses(self):
text = f"{_AFFINE_BODY} How many paperclips does Yun have?"
assert resolve_promotable_affine_fraction_delta(text) is None
assert _run(text).answer is None
class TestAffineCompletenessGuard:
def test_duplicate_offset_distractor_refuses(self):
text = (
"Yun had 20 paperclips initially, but then lost 12. There are 7 boxes on the shelf. "
"Marion has 1/4 more than what Yun currently has, plus 7. "
"How many paperclips does Marion have?"
)
assert compose_affine_fraction_delta(text) is None
assert _run(text).answer is None
class TestTemporalActorBinding:
def test_overtime_question_actor_mismatch_refuses(self):
text = f"{_OVERTIME_BODY} how much money does Bob make?"
assert resolve_promotable_temporal_tariff(text) is None
assert _run(text).answer is None
def test_bundle_question_renter_mismatch_refuses(self):
text = f"{_BUNDLE_BODY} What does it cost Alice?"
assert resolve_promotable_temporal_tariff(text) is None
assert _run(text).answer is None
class TestTemporalThresholdAnchoring:
def test_non_shift_threshold_refuses(self):
text = (
"Tina makes $18.00 an hour. If she works more than 8 hours per week, she is "
"eligible for overtime, which is paid by your hourly wage + 1/2 your hourly wage. "
"If she works 10 hours every day for 5 days, how much money does she make?"
)
assert compose_overtime_shift_earnings(text) is None
assert _run(text).answer is None
class TestTemporalRentalVerbVariants:
def test_rented_still_admits(self):
text = (
"Jason has a carriage house that he rents out. He's charging $50.00 per day or "
"$500.00 for 14 days. Eric rented the house for 20 days. "
"How much will it cost him?"
)
res = compose_bundle_overflow_tariff(text)
assert res is not None
assert res.answer == 800.0
def test_renting_still_admits(self):
text = (
"Jason has a carriage house that he rents out. He's charging $50.00 per day or "
"$500.00 for 14 days. Eric is renting the house for 20 days. "
"How much will it cost him?"
)
res = compose_bundle_overflow_tariff(text)
assert res is not None
assert res.answer == 800.0
class TestTemporalQuestionRobustness:
def test_how_much_does_she_earn_admits(self):
text = f"{_OVERTIME_BODY} how much does she earn?"
res = compose_overtime_shift_earnings(text)
assert res is not None
assert res.answer == 990.0
def test_what_is_total_cost_admits(self):
text = f"{_BUNDLE_BODY} What is the total cost?"
res = compose_bundle_overflow_tariff(text)
assert res is not None
assert res.answer == 800.0
class TestPercentPartitionGroupOf:
def test_group_of_total_still_admits(self):
res = compose_percent_partition(_PERCENT_GROUP_OF_100)
assert res is not None
assert res.answer == 15.0
assert _run(_PERCENT_GROUP_OF_100).answer == 15.0
def test_subgroup_group_still_refuses(self):
assert resolve_promotable_percent_partition(_PERCENT_SUBGROUP_CONFUSER) is None
assert _run(_PERCENT_SUBGROUP_CONFUSER).answer is None