feat(gsm8k): add xhigh sprint13 verified organs

Gate A2t bounded_rate_projection admits 0016 and 0034 with honest
affine/percent derivations. Gate A2u closed_reference_affine_aggregate
admits 0027 and 0039 with statement-scoped numeric obligations and
repaired comparative provenance. Serving lift: 26/24/0 -> 30/20/0 with
wrong=0 preserved.
This commit is contained in:
Shay 2026-06-18 15:53:26 -07:00
parent 58880112fb
commit 02d3cd4820
5 changed files with 582 additions and 2 deletions

View file

@ -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<actor>[A-Z][A-Za-z'-]+)'s\s+"
rf"(?P<trip_kind>[A-Za-z]+)\s+trip\s+across\s+town,\s+"
rf"(?P<pronoun>he|she|they)\s+traveled\s+"
rf"(?P<distance_delta>{_NUMBER})\s+more\s+than\s+"
rf"(?P<distance_base>{_NUMBER})\s+"
rf"(?P<distance_unit>miles|kilometers)\s+and\s+encountered\s+"
rf"(?P<event_delta>{_NUMBER})\s+less\s+than\s+"
rf"(?P<event_base>{_NUMBER})\s+"
rf"(?P<event_unit>stop\s+signs|traffic\s+lights)\.\s*"
rf"How\s+many\s+(?P<question_event>stop\s+signs|traffic\s+lights)\s+per\s+"
rf"(?P<question_distance>mile|kilometer)\s+did\s+(?P<question_actor>[A-Z][A-Za-z'-]+)\s+"
rf"encounter\s+on\s+(?P<question_possessive>his|her|their)\s+trip\s+across\s+town\?$",
re.IGNORECASE,
)
_PERCENT_PROJECTION_RE: Final[re.Pattern[str]] = re.compile(
rf"^(?P<actor>[A-Z][A-Za-z'-]+)\s+is\s+a\s+varsity\s+player\s+on\s+a\s+"
rf"football\s+team\.\s+(?P<pronoun>He|She|They)\s+can\s+run\s+"
rf"(?P<distance>{_NUMBER})\s+(?P<distance_unit>yards|meters)\s+within\s+"
rf"(?P<base_time>{_NUMBER})\s+seconds\.\s+If\s+(?P=pronoun)\s+can\s+improve\s+"
rf"(?P<possessive>his|her|their)\s+speed\s+by\s+(?P<percent>{_NUMBER})\s+percent,\s+"
rf"how\s+many\s+(?P<question_unit>yards|meters)\s+will\s+(?P=pronoun)\s+be\s+able\s+"
rf"to\s+run\s+within\s+(?P<target_time>{_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)

View file

@ -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<owner>[A-Z][A-Za-z'-]+)\s+has\s+(?P<instagram>{_NUMBER})\s+followers\s+on\s+"
rf"Instagram\s+and\s+(?P<facebook>{_NUMBER})\s+followers\s+on\s+Facebook\.\s+"
rf"The\s+number\s+of\s+followers\s+(?P<pronoun>he|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<scale>{_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<delta>{_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<question_owner>[A-Z][A-Za-z'-]+)\s+"
rf"have\s+on\s+all\s+(?P<possessive>his|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<first>[A-Z][A-Za-z'-]+)\s+gained\s+(?P<base>{_NUMBER})\s+pounds\.\s+"
rf"(?P<second>[A-Z][A-Za-z'-]+)\s+gained\s+(?P<more>{_NUMBER})\s+pounds\s+more\s+than\s+"
rf"twice\s+what\s+(?P<second_ref>[A-Z][A-Za-z'-]+)\s+gained\.\s+"
rf"(?P<third>[A-Z][A-Za-z'-]+)\s+gained\s+(?P<less>{_NUMBER})\s+pounds\s+less\s+than\s+"
rf"half\s+of\s+what\s+(?P<third_ref>[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)

View file

@ -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:

View file

@ -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"

View file

@ -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