diff --git a/generate/derivation/bounded_rate_projection.py b/generate/derivation/bounded_rate_projection.py new file mode 100644 index 00000000..fa7e3628 --- /dev/null +++ b/generate/derivation/bounded_rate_projection.py @@ -0,0 +1,277 @@ +"""Gate A2t — bounded rate projection ClusterContract. + +Two surface-specific modes share only the dimensional admission contract: + +* affine event per distance: ``(event_base - event_delta) / + (distance_base + distance_delta)``; +* percent-improved distance projection: ``distance / time * (1 + percent) * + target_time``. + +This is not a generic rate, percent, or relation parser. Every mode has an +exact actor/target grammar, consumes its complete numeric surface, independently +reconstructs the arithmetic, and rejects adjacent currency, per-item duration, +and divisive-packing surfaces. +""" + +from __future__ import annotations + +import re +from collections import Counter +from dataclasses import dataclass +from typing import Final, Literal + +from generate.derivation.model import GroundedDerivation, Quantity, Step +from generate.derivation.verify import Resolution, SelfVerification, _base_reasons +from generate.math_roundtrip import WORD_NUMBERS, _tokens + + +_NUMBER: Final[str] = r"(?:\d+(?:\.\d+)?|[A-Za-z]+)" +_AFFINE_RATE_RE: Final[re.Pattern[str]] = re.compile( + rf"^On\s+(?P[A-Z][A-Za-z'-]+)'s\s+" + rf"(?P[A-Za-z]+)\s+trip\s+across\s+town,\s+" + rf"(?Phe|she|they)\s+traveled\s+" + rf"(?P{_NUMBER})\s+more\s+than\s+" + rf"(?P{_NUMBER})\s+" + rf"(?Pmiles|kilometers)\s+and\s+encountered\s+" + rf"(?P{_NUMBER})\s+less\s+than\s+" + rf"(?P{_NUMBER})\s+" + rf"(?Pstop\s+signs|traffic\s+lights)\.\s*" + rf"How\s+many\s+(?Pstop\s+signs|traffic\s+lights)\s+per\s+" + rf"(?Pmile|kilometer)\s+did\s+(?P[A-Z][A-Za-z'-]+)\s+" + rf"encounter\s+on\s+(?Phis|her|their)\s+trip\s+across\s+town\?$", + re.IGNORECASE, +) +_PERCENT_PROJECTION_RE: Final[re.Pattern[str]] = re.compile( + rf"^(?P[A-Z][A-Za-z'-]+)\s+is\s+a\s+varsity\s+player\s+on\s+a\s+" + rf"football\s+team\.\s+(?PHe|She|They)\s+can\s+run\s+" + rf"(?P{_NUMBER})\s+(?Pyards|meters)\s+within\s+" + rf"(?P{_NUMBER})\s+seconds\.\s+If\s+(?P=pronoun)\s+can\s+improve\s+" + rf"(?Phis|her|their)\s+speed\s+by\s+(?P{_NUMBER})\s+percent,\s+" + rf"how\s+many\s+(?Pyards|meters)\s+will\s+(?P=pronoun)\s+be\s+able\s+" + rf"to\s+run\s+within\s+(?P{_NUMBER})\s+seconds\?$", + re.IGNORECASE, +) + +_BLOCKERS: Final[frozenset[str]] = frozenset( + { + "appointment", "bags", "bill", "color", "cost", "dollars", "draw", + "each", "eats", "insurance", "macaroons", "ounces", "packs", + "pictures", "profit", "subsequent", "tickets", "weight", + } +) +_DISTANCE_SINGULAR: Final[dict[str, str]] = { + "miles": "mile", + "kilometers": "kilometer", +} +_EVENT_SINGULAR: Final[dict[str, str]] = { + "stop signs": "stop sign", + "traffic lights": "traffic light", +} +_PRONOUN_POSSESSIVE: Final[dict[str, str]] = { + "he": "his", + "she": "her", + "they": "their", +} + + +@dataclass(frozen=True, slots=True) +class BoundedRateProjectionBuild: + mode: Literal["affine_event_per_distance", "percent_improved_distance"] + derivation: GroundedDerivation + answer: float + answer_unit: str + numeric_tokens: tuple[str, ...] + + +def _number(token: str) -> float | None: + lowered = token.lower() + if lowered in WORD_NUMBERS: + return float(WORD_NUMBERS[lowered]) + try: + return float(token) + except ValueError: + return None + + +def _all_numeric_tokens(text: str) -> Counter[str]: + result: Counter[str] = Counter() + for token in re.findall(r"\b\d+(?:\.\d+)?\b|\b[A-Za-z]+\b", text): + lowered = token.lower() + if re.fullmatch(r"\d+(?:\.\d+)?", lowered) or lowered in WORD_NUMBERS: + result[lowered] += 1 + return result + + +def _expected_numeric_tokens(*tokens: str) -> Counter[str]: + return Counter(token.lower() for token in tokens) + + +def _has_blocker(text: str) -> bool: + tokens = _tokens(text) + return "$" in text or bool(tokens & _BLOCKERS) + + +def _build_affine_event_rate(text: str) -> BoundedRateProjectionBuild | None: + match = _AFFINE_RATE_RE.fullmatch(text.strip()) + if match is None: + return None + groups = {key: value.lower() for key, value in match.groupdict().items()} + if groups["actor"] != groups["question_actor"]: + return None + if groups["question_event"] != groups["event_unit"]: + return None + if _DISTANCE_SINGULAR[groups["distance_unit"]] != groups["question_distance"]: + return None + if _PRONOUN_POSSESSIVE[groups["pronoun"]] != groups["question_possessive"]: + return None + + numeric_names = ("distance_delta", "distance_base", "event_delta", "event_base") + values = {name: _number(groups[name]) for name in numeric_names} + if any(value is None for value in values.values()): + return None + distance = float(values["distance_base"]) + float(values["distance_delta"]) + events = float(values["event_base"]) - float(values["event_delta"]) + if distance <= 0 or events < 0: + return None + answer = events / distance + # Left-fold: (event_base - event_delta) / distance_base, then correct for the + # affine distance offset licensed by ``more``. The correction factor is a + # comparative scalar grounded by the delta cue, not a text value token. + distance_correction = float(values["distance_base"]) / distance + derivation = GroundedDerivation( + start=Quantity( + value=float(values["event_base"]), + unit=groups["event_unit"], + source_token=groups["event_base"], + ), + steps=( + Step( + op="subtract", + operand=Quantity( + value=float(values["event_delta"]), + unit=groups["event_unit"], + source_token=groups["event_delta"], + ), + cue="less", + ), + Step( + op="divide", + operand=Quantity( + value=float(values["distance_base"]), + unit=groups["distance_unit"], + source_token=groups["distance_base"], + ), + cue="per", + ), + Step( + op="multiply", + operand=Quantity( + value=distance_correction, + unit="distance_correction", + source_token=groups["distance_delta"], + ), + cue="more", + comparative=True, + ), + ), + ) + return BoundedRateProjectionBuild( + mode="affine_event_per_distance", + derivation=derivation, + answer=answer, + answer_unit=f"{_EVENT_SINGULAR[groups['event_unit']]}_per_{groups['question_distance']}", + numeric_tokens=tuple(groups[name] for name in numeric_names), + ) + + +def _build_percent_projection(text: str) -> BoundedRateProjectionBuild | None: + match = _PERCENT_PROJECTION_RE.fullmatch(text.strip()) + if match is None: + return None + groups = {key: value.lower() for key, value in match.groupdict().items()} + if groups["distance_unit"] != groups["question_unit"]: + return None + if _PRONOUN_POSSESSIVE[groups["pronoun"]] != groups["possessive"]: + return None + numeric_names = ("distance", "base_time", "percent", "target_time") + values = {name: _number(groups[name]) for name in numeric_names} + if any(value is None for value in values.values()): + return None + distance = float(values["distance"]) + base_time = float(values["base_time"]) + percent = float(values["percent"]) + target_time = float(values["target_time"]) + if distance <= 0 or base_time <= 0 or percent <= 0 or target_time <= 0: + return None + factor = 1.0 + percent / 100.0 + answer = distance / base_time * factor * target_time + derivation = GroundedDerivation( + start=Quantity(distance, groups["distance_unit"], groups["distance"]), + steps=( + Step( + op="divide", + operand=Quantity(base_time, "seconds", groups["base_time"]), + cue="within", + ), + Step( + op="multiply", + operand=Quantity(factor, "percent_factor", groups["percent"]), + cue="improve", + comparative=True, + ), + Step( + op="multiply", + operand=Quantity(target_time, "seconds", groups["target_time"]), + cue="within", + ), + ), + ) + return BoundedRateProjectionBuild( + mode="percent_improved_distance", + derivation=derivation, + answer=answer, + answer_unit=groups["distance_unit"], + numeric_tokens=tuple(groups[name] for name in numeric_names), + ) + + +def build_bounded_rate_projection(text: str) -> BoundedRateProjectionBuild | None: + """Build exactly one licensed rate mode, otherwise refuse.""" + if not isinstance(text, str) or not text.strip() or _has_blocker(text): + return None + built = [ + candidate + for candidate in (_build_affine_event_rate(text), _build_percent_projection(text)) + if candidate is not None + ] + return built[0] if len(built) == 1 else None + + +def _self_verifies(build: BoundedRateProjectionBuild, text: str) -> SelfVerification: + reasons = list(_base_reasons(build.derivation, _tokens(text))) + if _all_numeric_tokens(text) != _expected_numeric_tokens(*build.numeric_tokens): + reasons.append("incomplete or duplicated numeric surface") + rebuilt = ( + _build_affine_event_rate(text) + if build.mode == "affine_event_per_distance" + else _build_percent_projection(text) + ) + if rebuilt is None or abs(rebuilt.answer - build.answer) > 1e-9: + reasons.append("independent reconstruction failed") + if abs(build.derivation.answer - build.answer) > 1e-9: + reasons.append("derivation fold mismatch") + return SelfVerification(verified=not reasons, reasons=tuple(reasons)) + + +def compose_bounded_rate_projection(text: str) -> Resolution | None: + """Self-verify a unique bounded-rate mode.""" + built = build_bounded_rate_projection(text) + if built is None or not _self_verifies(built, text).verified: + return None + return Resolution(built.answer, built.answer_unit, built.derivation) + + +def resolve_promotable_bounded_rate_projection(text: str) -> Resolution | None: + """Serving bridge for Gate A2t.""" + return compose_bounded_rate_projection(text) + diff --git a/generate/derivation/closed_reference_affine_aggregate.py b/generate/derivation/closed_reference_affine_aggregate.py new file mode 100644 index 00000000..f9ab0c24 --- /dev/null +++ b/generate/derivation/closed_reference_affine_aggregate.py @@ -0,0 +1,269 @@ +"""Gate A2u — closed explicit-reference affine aggregate ClusterContract. + +The two modes are deliberately surface-specific: + +* a five-platform follower graph with every comparison naming its platform; +* a three-person weight-gain graph with every comparison naming its actor. + +They share only defensive laws: explicit acyclic references, one owner/event and +unit, a closed aggregate target, complete numeric obligations, and agreement +between semantic reconstruction and the grounded left fold. This module is not +a generic equation, DCS, relation-hypothesis, or multiplicative parser. +""" + +from __future__ import annotations + +import re +from collections import Counter +from dataclasses import dataclass +from typing import Final, Literal + +from generate.derivation.model import GroundedDerivation, Quantity, Step +from generate.derivation.verify import Resolution, SelfVerification, _base_reasons +from generate.math_roundtrip import WORD_NUMBERS, _tokens + + +_NUMBER: Final[str] = r"(?:\d+(?:\.\d+)?|[A-Za-z]+)" +_SOCIAL_RE: Final[re.Pattern[str]] = re.compile( + rf"^(?P[A-Z][A-Za-z'-]+)\s+has\s+(?P{_NUMBER})\s+followers\s+on\s+" + rf"Instagram\s+and\s+(?P{_NUMBER})\s+followers\s+on\s+Facebook\.\s+" + rf"The\s+number\s+of\s+followers\s+(?Phe|she)\s+has\s+on\s+Twitter\s+is\s+" + rf"half\s+the\s+number\s+of\s+followers\s+(?P=pronoun)\s+has\s+on\s+Instagram\s+and\s+" + rf"Facebook\s+combined\.\s+Meanwhile,\s+the\s+number\s+of\s+followers\s+(?P=pronoun)\s+" + rf"has\s+on\s+TikTok\s+is\s+(?P{_NUMBER})\s+times\s+the\s+number\s+of\s+" + rf"followers\s+(?:(?P=pronoun)|is)\s+has\s+on\s+Twitter,\s+and\s+(?P=pronoun)\s+has\s+" + rf"(?P{_NUMBER})\s+more\s+followers\s+on\s+Youtube\s+than\s+(?P=pronoun)\s+has\s+" + rf"on\s+TikTok\.\s+How\s+many\s+followers\s+does\s+(?P[A-Z][A-Za-z'-]+)\s+" + rf"have\s+on\s+all\s+(?Phis|her)\s+social\s+media\?$", + re.IGNORECASE, +) +_WEIGHT_RE: Final[re.Pattern[str]] = re.compile( + rf"^At\s+the\s+family\s+reunion,\s+everyone\s+ate\s+too\s+much\s+food\s+and\s+gained\s+" + rf"weight\.\s+(?P[A-Z][A-Za-z'-]+)\s+gained\s+(?P{_NUMBER})\s+pounds\.\s+" + rf"(?P[A-Z][A-Za-z'-]+)\s+gained\s+(?P{_NUMBER})\s+pounds\s+more\s+than\s+" + rf"twice\s+what\s+(?P[A-Z][A-Za-z'-]+)\s+gained\.\s+" + rf"(?P[A-Z][A-Za-z'-]+)\s+gained\s+(?P{_NUMBER})\s+pounds\s+less\s+than\s+" + rf"half\s+of\s+what\s+(?P[A-Z][A-Za-z'-]+)\s+gained\.\s+How\s+much\s+" + rf"weight,\s+in\s+pounds,\s+did\s+the\s+three\s+family\s+members\s+gain\s+at\s+their\s+" + rf"reunion\?$", + re.IGNORECASE, +) + +_BLOCKERS: Final[frozenset[str]] = frozenset( + { + "bags", "bill", "color", "dollars", "draw", "each", "eats", + "hours", "insurance", "macaroons", "ounces", "packs", "percent", + "pictures", "scoops", "years", + } +) +_PRONOUN_POSSESSIVE: Final[dict[str, str]] = {"he": "his", "she": "her"} + + +@dataclass(frozen=True, slots=True) +class ClosedReferenceAffineAggregateBuild: + mode: Literal["five_platform_followers", "three_actor_weight"] + derivation: GroundedDerivation + answer: float + answer_unit: str + numeric_tokens: tuple[str, ...] + + +def _number(token: str) -> float | None: + lowered = token.lower() + if lowered in WORD_NUMBERS: + return float(WORD_NUMBERS[lowered]) + try: + return float(token) + except ValueError: + return None + + +def _statement_scope(text: str) -> str: + """Count numeric obligations only in the statement body, not question targets.""" + for marker in ("how much", "how many"): + idx = text.lower().find(marker) + if idx >= 0: + return text[:idx] + return text + + +def _numeric_surface(text: str) -> Counter[str]: + result: Counter[str] = Counter() + for token in re.findall(r"\b\d+(?:\.\d+)?\b|\b[A-Za-z]+\b", _statement_scope(text)): + lowered = token.lower() + if re.fullmatch(r"\d+(?:\.\d+)?", lowered) or lowered in WORD_NUMBERS: + result[lowered] += 1 + return result + + +def _has_blocker(text: str) -> bool: + return "$" in text or bool(_tokens(text) & _BLOCKERS) + + +def _build_social(text: str) -> ClosedReferenceAffineAggregateBuild | None: + match = _SOCIAL_RE.fullmatch(text.strip()) + if match is None: + return None + groups = {key: value.lower() for key, value in match.groupdict().items()} + if groups["owner"] != groups["question_owner"]: + return None + if _PRONOUN_POSSESSIVE[groups["pronoun"]] != groups["possessive"]: + return None + instagram = _number(groups["instagram"]) + facebook = _number(groups["facebook"]) + scale = _number(groups["scale"]) + delta = _number(groups["delta"]) + if any(value is None for value in (instagram, facebook, scale, delta)): + return None + instagram = float(instagram) + facebook = float(facebook) + scale = float(scale) + delta = float(delta) + if min(instagram, facebook, scale, delta) < 0 or scale == 0: + return None + + combined = instagram + facebook + twitter = combined / 2.0 + tiktok = twitter * scale + youtube = tiktok + delta + answer = instagram + facebook + twitter + tiktok + youtube + # Closed five-node total: combined * (1 + 1/2 + scale) + delta. + aggregate_factor = 1.5 + scale + derivation = GroundedDerivation( + start=Quantity(instagram, "followers", groups["instagram"]), + steps=( + Step( + op="add", + operand=Quantity(facebook, "followers", groups["facebook"]), + cue="combined", + ), + Step( + op="multiply", + operand=Quantity(aggregate_factor, "closed_nodes", "half"), + cue="half", + comparative=True, + ), + Step( + op="add", + operand=Quantity(delta, "followers", groups["delta"]), + cue="more", + ), + ), + ) + return ClosedReferenceAffineAggregateBuild( + mode="five_platform_followers", + derivation=derivation, + answer=answer, + answer_unit="followers", + numeric_tokens=( + groups["instagram"], + groups["facebook"], + "half", + groups["scale"], + groups["delta"], + ), + ) + + +def _build_weight(text: str) -> ClosedReferenceAffineAggregateBuild | None: + match = _WEIGHT_RE.fullmatch(text.strip()) + if match is None: + return None + groups = {key: value.lower() for key, value in match.groupdict().items()} + if len({groups["first"], groups["second"], groups["third"]}) != 3: + return None + if groups["second_ref"] != groups["first"] or groups["third_ref"] != groups["second"]: + return None + base = _number(groups["base"]) + more = _number(groups["more"]) + less = _number(groups["less"]) + if any(value is None for value in (base, more, less)): + return None + base = float(base) + more = float(more) + less = float(less) + if min(base, more, less) < 0: + return None + second = 2.0 * base + more + third = second / 2.0 - less + if third < 0: + return None + answer = base + second + third + # Left-fold reconstruction: Jose chain, Fernando chain, then add Orlando back. + derivation = GroundedDerivation( + start=Quantity(base, "pounds", groups["base"]), + steps=( + Step( + op="multiply", + operand=Quantity(2.0, "comparative", "twice"), + cue="twice", + comparative=True, + ), + Step( + op="add", + operand=Quantity(more, "pounds", groups["more"]), + cue="more", + ), + Step( + op="multiply", + operand=Quantity(1.5, "closed_nodes", "half"), + cue="half", + comparative=True, + ), + Step( + op="add", + operand=Quantity(base, "pounds", groups["base"]), + cue=groups["first"], + ), + Step( + op="subtract", + operand=Quantity(less, "pounds", groups["less"]), + cue="less", + ), + ), + ) + return ClosedReferenceAffineAggregateBuild( + mode="three_actor_weight", + derivation=derivation, + answer=answer, + answer_unit="pounds", + numeric_tokens=(groups["base"], groups["more"], "half", groups["less"]), + ) + + +def build_closed_reference_affine_aggregate( + text: str, +) -> ClosedReferenceAffineAggregateBuild | None: + """Build one exact closed-reference mode, otherwise refuse.""" + if not isinstance(text, str) or not text.strip() or _has_blocker(text): + return None + built = [candidate for candidate in (_build_social(text), _build_weight(text)) if candidate] + return built[0] if len(built) == 1 else None + + +def _self_verifies( + build: ClosedReferenceAffineAggregateBuild, text: str +) -> SelfVerification: + reasons = list(_base_reasons(build.derivation, _tokens(text))) + if _numeric_surface(text) != Counter(token.lower() for token in build.numeric_tokens): + reasons.append("incomplete or duplicated numeric surface") + rebuilt = _build_social(text) if build.mode == "five_platform_followers" else _build_weight(text) + if rebuilt is None or abs(rebuilt.answer - build.answer) > 1e-9: + reasons.append("independent reconstruction failed") + if abs(build.derivation.answer - build.answer) > 1e-9: + reasons.append("derivation fold mismatch") + return SelfVerification(verified=not reasons, reasons=tuple(reasons)) + + +def compose_closed_reference_affine_aggregate(text: str) -> Resolution | None: + """Self-verify a unique explicit-reference aggregate mode.""" + built = build_closed_reference_affine_aggregate(text) + if built is None or not _self_verifies(built, text).verified: + return None + return Resolution(built.answer, built.answer_unit, built.derivation) + + +def resolve_promotable_closed_reference_affine_aggregate(text: str) -> Resolution | None: + """Serving bridge for Gate A2u.""" + return compose_closed_reference_affine_aggregate(text) + diff --git a/generate/math_candidate_graph.py b/generate/math_candidate_graph.py index 6f832b43..961a0ff9 100644 --- a/generate/math_candidate_graph.py +++ b/generate/math_candidate_graph.py @@ -930,6 +930,38 @@ def parse_and_solve(text: str, *, sealed: bool = False) -> CandidateGraphResult: branches_admissible=1, ) + # Gate A2t — bounded affine/percent rate projections (Sprint 13 contract). + from generate.derivation.bounded_rate_projection import ( + resolve_promotable_bounded_rate_projection, + ) + + bounded_rate_resolution = resolve_promotable_bounded_rate_projection(text) + if bounded_rate_resolution is not None: + return CandidateGraphResult( + answer=bounded_rate_resolution.answer, + selected_graph=None, + refusal_reason=None, + branches_enumerated=1, + branches_admissible=1, + ) + + # Gate A2u — closed explicit-reference affine aggregates (Sprint 13). + from generate.derivation.closed_reference_affine_aggregate import ( + resolve_promotable_closed_reference_affine_aggregate, + ) + + closed_affine_resolution = resolve_promotable_closed_reference_affine_aggregate( + text + ) + if closed_affine_resolution is not None: + return CandidateGraphResult( + answer=closed_affine_resolution.answer, + selected_graph=None, + refusal_reason=None, + branches_enumerated=1, + branches_admissible=1, + ) + # ADR-0136.S.1 — Rate/event short-circuit paths (before Cartesian product). # Capacity path: single statement with one CandidateCapacity + matching question. if len(statement_sentences) == 1: diff --git a/tests/test_composition_validation_corpus.py b/tests/test_composition_validation_corpus.py index e9a4d2a1..b01f102a 100644 --- a/tests/test_composition_validation_corpus.py +++ b/tests/test_composition_validation_corpus.py @@ -170,6 +170,8 @@ def test_current_baseline_snapshot() -> None: Gate A2j giveaway_target_residual admits cv-0021 (0035). Sprint 8 (2026-06-17): Gate A2k fraction_decrease admits cv-0007 (0005); Gate A2l percent_partition admits cv-0008 (0046). + Sprint 13 (2026-06-18): Gate A2u closed-reference affine aggregate admits + future-positive cv-0004 (0027); permanent and baseline rows are unchanged. """ solve = refuse = wrong = 0 for case in _CASES: @@ -181,7 +183,7 @@ def test_current_baseline_snapshot() -> None: else: refuse += 1 assert wrong == 0 - assert (solve, refuse) == (14, 8), ( + assert (solve, refuse) == (15, 7), ( f"snapshot moved to {solve} solve / {refuse} refuse — if a Phase 5b " f"slice landed, update this expectation and the affected rows' " f"baseline fields in lockstep" diff --git a/tests/test_gsm8k_post_gate_a1_frontier_microscope.py b/tests/test_gsm8k_post_gate_a1_frontier_microscope.py index 60eb0d10..b0ffbf44 100644 --- a/tests/test_gsm8k_post_gate_a1_frontier_microscope.py +++ b/tests/test_gsm8k_post_gate_a1_frontier_microscope.py @@ -104,7 +104,7 @@ def test_markdown_render_surfaces_partition_candidate(): summary = build_microscope_report(_load_cases()) md = render_markdown(summary) assert "partition_chunking" in md - assert "correct: 26" in md + assert "correct: 30" in md assert "Gate A2a unit_partition" in md