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

Merge pull request 'feat(trackb): S1–S4 symbolic SME, selector, pure-S1 coverage gain (Inc2)' (#86) from feat/trackb-symbolic-sme-s2s4 into main

Browse source 
Reviewed-on: #86
This commit is contained in:
Joshua Matthew-Catudio Shay 2026-07-20 03:12:47 +00:00
commit d8d62b8ea3
15 changed files with 2521 additions and 393 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

169
docs/research/trackb-inc2-generalization-2026-07-20.md Normal file
View file

@ -0,0 +1,169 @@
# Track B, Increment 2 — Generalization bar (S1–S4, selector, coverage)
**Date**: 2026-07-20
**Branch**: `feat/trackb-symbolic-sme-s2s4`
**Paradigm**: ADR-0252
**Prerequisite**: Increment 1 (PR #85) merged; §5 geometric SME remains NO-GO
**Scope**: Off-serving. S1–S4 canonicals + selector + SM-owned pure-S1 extract. No serving cutover. No organ retirement in this PR.
---
## 1. The bar (ADR-0252 §4 / Increment 2)
| # | Requirement | Result |
| --- | --- | --- |
| 1 | Generalization ratio > 1 on ≥1 canonical (real holdout IDs) | **PASS** — S1 ratio = **9.0** (9 cases / 1 template) |
| 2 | Coverage gain **or** organ retirement | **PASS** — coverage gain: **+4** holdout cases (organ 5 → trackb 9), `wrong=0` |
| 3 | Selector works (S1–S4 held; ties refuse; surface≠structure) | **PASS** — routing table below |
Honest negatives (not failures of the bar):
- S2/S3/S4 holdout ratios = 0 — **parser frontier**: organ yields graphs on only 5/500 holdout cases (all S1).
- S3/S4 multi-register corridor cannot certify pure rebuilds yet (`apply_rate` / `compare_additive` out of scope) — map succeeds, emit refuses (gate held, not weakened).
- Organ retirement **not** performed (off-serving path; coverage-gain chosen).
---
## 2. What shipped
| Piece | Path |
| --- | --- |
| Canonicals S1–S4 | `generate/structure_mapping/canonicals.py` |
| Role schema + `compare_add` | `generate/structure_mapping/role_predicate.py` |
| Converter (incl. additive compare) | `generate/structure_mapping/convert.py` |
| Pure-family mappers S1–S4 | `generate/structure_mapping/mapper.py` |
| Overlapping-waves selector | `generate/structure_mapping/selector.py` |
| Generalized 3-gate solve | `generate/structure_mapping/solve.py` |
| SM-owned pure-S1 text extract | `generate/structure_mapping/text_extract.py` |
| Pipeline (organ parse → extract fallback) | `generate/structure_mapping/pipeline.py` |
| Measure script | `scripts/measure_trackb_inc2.py` |
| Tests | `tests/test_trackb_inc2_structure_mapping.py` (15) + Inc1 suite (20) |
Serving reader (`parse_and_solve` dispatch, organs, `chat/runtime.py`) **untouched**.
---
## 3. Measurements (command + real output)
### 3a. Generalization-ratio table
```text
$ uv run python scripts/measure_trackb_inc2.py --mode ratio
=== generalization ratio (holdout_dev/v1, command-backed) ===
CASE gsm8k-holdout-dev-v1-0101 sid=S1 emit=9.0 gold=9.0 right=True src=organ ...
CASE gsm8k-holdout-dev-v1-0108 sid=S1 emit=110.0 gold=110.0 right=True src=organ ...
CASE gsm8k-holdout-dev-v1-0148 sid=S1 emit=1500.0 gold=1500.0 right=True src=sm_extract pat=s1_p3_day_pair ...
CASE gsm8k-holdout-dev-v1-0228 sid=S1 emit=600.0 gold=600.0 right=True src=sm_extract pat=s1_p2_b_seeded ...
CASE gsm8k-holdout-dev-v1-0234 sid=S1 emit=240.0 gold=240.0 right=True src=sm_extract pat=s1_p2_deaf_blind ...
CASE gsm8k-holdout-dev-v1-0268 sid=S1 emit=16.0 gold=16.0 right=True src=organ ...
CASE gsm8k-holdout-dev-v1-0411 sid=S1 emit=52.0 gold=52.0 right=True src=organ ...
CASE gsm8k-holdout-dev-v1-0441 sid=S1 emit=75.0 gold=75.0 right=True src=sm_extract pat=s1_p2_b_seeded ...
CASE gsm8k-holdout-dev-v1-0453 sid=S1 emit=28.0 gold=28.0 right=True src=organ ...
--- ratio table ---
canonical=S1 templates=1 cases_carried=9 ratio=9.000 ids=[...9 holdout ids...]
canonical=S2 templates=1 cases_carried=0 ratio=0.000 ids=[]
canonical=S3 templates=1 cases_carried=0 ratio=0.000 ids=[]
canonical=S4 templates=1 cases_carried=0 ratio=0.000 ids=[]
SUMMARY active_canonicals=1 total_cases=9 aggregate_ratio=9.000
```
All nine S1 carries include the 3-part proof (right answer, derivation `b=k×a; total=a+b`, multi-register + classical cert). Surface-variant re-extract of the inverted-seed form still maps (Ava/Ben → 48).
### 3b. Coverage gain
```text
$ uv run python scripts/measure_trackb_inc2.py --mode coverage-gain
organ_correct=5 trackb_correct=9 newly_solved=4
NEW gsm8k-holdout-dev-v1-0148 ans=1500.0 gold=1500.0 sid=S1 extract=s1_p3_day_pair
NEW gsm8k-holdout-dev-v1-0228 ans=600.0 gold=600.0 sid=S1 extract=s1_p2_b_seeded
NEW gsm8k-holdout-dev-v1-0234 ans=240.0 gold=240.0 sid=S1 extract=s1_p2_deaf_blind
NEW gsm8k-holdout-dev-v1-0441 ans=75.0 gold=75.0 sid=S1 extract=s1_p2_b_seeded
```
These four are **real surface-distinct holdout cases**, not synthetic clones. The SM-owned extract layer is structure-mapping-owned (regex pure-family templates) and runs only when the serving reader refuses — it does not modify the live reader.
### 3c. Selector
```text
$ uv run python scripts/measure_trackb_inc2.py --mode selector
CASE real_s1_0411 expect=S1 got=S1 refused=False ok=True
CASE real_s2_public_transfer expect=S2 got=S2 refused=False ok=True
CASE struct_s2_not_s1 expect=S2 got=S2 refused=False ok=True
CASE struct_s1 expect=S1 got=S1 refused=False ok=True
CASE empty_ops_refuse expect=None got=None refused=True reason=no_family_matched ok=True
CASE struct_s3_rate expect=S3 got=S3 refused=False ok=True
```
Transfer graphs route to S2 even when the surface text could mention multiplicative language elsewhere; empty ops refuse. Ranking uses fixed systematicity scores, not entity names or numbers.
### 3d. Parser frontier
```text
organ_selected_graph_count=5/500
role_kind_multiset_counts={'compare,contain,total': 5}
--- case 0148 ---
organ_parsed=False
sm_pipeline emit=True answer=1500.0 gold=1500.0 src=sm_extract sid=S1
```
**0148** (the original S1 family namesake) is a **parser-frontier refusal** on the serving reader; the SM-owned extract recovers it. Holdout still has zero organ graphs for transfer/rate/additive-compare families — that is the next structural bottleneck, not the mapper.
### 3e. S2 on real public transfers (not holdout; engineering evidence only)
```text
SUMMARY s2_public_sample: ok=5 scanned_transferish=20
S2 gma-102 ans=54.0 ... transfer 7 Eve→David ...
```
S2 pure-family map+solve works on real public transfer graphs. Not counted in the holdout ratio table (discipline: holdout IDs only).
### 3f. Right-reason summary
```text
SUMMARY right-reason: correct=9 wrong=0 refused=0
SUMMARY surface-variant: ok=True
```
### 3g. Full holdout wrong=0 gate (all 500 — not curated subset)
Regression: a loose inverted-seed regex matched multi-entity zoo case **0361**
(`John has 8 more pandas…`) as pure S1 and the corridor emitted `1.125` vs
gold `114.0`. Fixed by pure-family surface gates (require explicit multiplicative
language; refuse bare `N more` / multi-clause / multi-numeric surfaces).
```text
$ uv run python scripts/measure_trackb_inc2.py --mode full-holdout-wrong0
=== full holdout_dev/v1 wrong=0 gate (all 500 cases) ===
SUMMARY full-holdout: emit_ok=9 wrong=0 refused=491 n=500
PASS: wrong=0 absolute on full holdout_dev/v1
```
---
## 4. Discipline checks
- **Blind**: mappers/selector take only `RoleGraph`; no label parameter. Static audit: `generate/structure_mapping/` does not import scoring labels.
- **wrong=0**: emit only if classical verify **and** multi-register cert **and** agreement; original-graph backstop when a graph is supplied.
- **No crash-as-signal**: corridor exceptions → refuse outcomes.
- **Off-serving**: no organ retirement; no `parse_and_solve` dispatch edits.
---
## 5. Deviations / frontiers (honest)
1. **S2–S4 holdout ratio = 0** until the serving reader (or a reviewed SM extract family) produces pure transfer/rate/additive graphs on holdout. Do not manufacture synthetic holdout clones for ratio.
2. **S3/S4 solve emit** blocked by multi-register scope (`apply_rate`, `compare_additive` not Tier-2a). Gates not weakened — refuse.
3. **SM extract is narrow** (regex pure-S1 templates only, with fail-closed purity gates). It is not a general parser; expanding it is a deliberate future increment. Overmatch risk is gated by `mode full-holdout-wrong0` on all 500 cases.
4. **Organ retirement** deferred; coverage-gain path met bar 2 without serving risk.
5. **0361 pure-S1 overmatch** (fixed in follow-up commit): optional `times` in P2 regex treated additive `8 more` as multiplicative; purity gate + required `times as many` closed it. Full-holdout scan is now mandatory before claiming wrong=0.
---
## 6. STOP
One PR (base `main`). Reviewer verifies:
- blindness + pure-family gates,
- ratio > 1 cases are real holdout IDs (not clones),
- `wrong=0` three-gate path,
- serving reader unchanged.
No S2–S4 serving cutover until a later ruling. No sealed-test run.

4
evals/structure_mapping/scoring/holdout_dev_v1_labels.jsonl
View file

@ -1,5 +1,9 @@
{"case_id": "gsm8k-holdout-dev-v1-0101", "label": "S1", "note": "compare_mult then total; organ cohort"}
{"case_id": "gsm8k-holdout-dev-v1-0108", "label": "S1", "note": "compare_mult then total; organ cohort"}
{"case_id": "gsm8k-holdout-dev-v1-0148", "label": "S1", "note": "pure S1; SM-owned extract (organ parse frontier)"}
{"case_id": "gsm8k-holdout-dev-v1-0228", "label": "S1", "note": "pure S1 b-seeded; SM-owned extract"}
{"case_id": "gsm8k-holdout-dev-v1-0234", "label": "S1", "note": "pure S1 b-seeded; SM-owned extract"}
{"case_id": "gsm8k-holdout-dev-v1-0268", "label": "S1", "note": "compare_mult then total; organ cohort"}
{"case_id": "gsm8k-holdout-dev-v1-0411", "label": "S1", "note": "compare_mult then total; organ cohort"}
{"case_id": "gsm8k-holdout-dev-v1-0441", "label": "S1", "note": "pure S1 b-seeded; SM-owned extract"}
{"case_id": "gsm8k-holdout-dev-v1-0453", "label": "S1", "note": "compare_mult then total; organ cohort"}

6
evals/structure_mapping/scoring/labels.py
View file

@ -18,11 +18,17 @@ from typing import Final, Mapping
# may be labeled S1 in extended JSONL if human-reviewed later.
S1_HOLDOUT_CASE_IDS: Final[frozenset[str]] = frozenset(
{
# Organ cohort (serving reader already solves)
"gsm8k-holdout-dev-v1-0101",
"gsm8k-holdout-dev-v1-0108",
"gsm8k-holdout-dev-v1-0268",
"gsm8k-holdout-dev-v1-0411",
"gsm8k-holdout-dev-v1-0453",
# Increment 2 coverage gain via SM-owned pure-S1 extract (organ misses)
"gsm8k-holdout-dev-v1-0148",
"gsm8k-holdout-dev-v1-0228",
"gsm8k-holdout-dev-v1-0234",
"gsm8k-holdout-dev-v1-0441",
}
)

48
generate/structure_mapping/__init__.py
View file

@ -1,22 +1,37 @@
"""Track B symbolic structure-mapping (ADR-0252 §6 / Increment 1).
"""Track B symbolic structure-mapping (ADR-0252 §6 / Increments 1–2).
Off-serving. Discovers role correspondence via Gentner-style symbolic
relational graph match — not geometric Procrustes (ruled NO-GO: Kabsch
presupposes the correspondence structure-mapping must discover).
Public surface for the S1 vertical slice only. Gold structure labels
(S1–S4) never enter the mapper; scoring helpers live in
``evals/structure_mapping/scoring`` and must not be imported here.
Public surface for S1–S4 + selector. Gold structure labels (S1–S4) never
enter the mapper; scoring helpers live in ``evals/structure_mapping/scoring``
and must not be imported here.
"""
from __future__ import annotations
from generate.structure_mapping.canonicals import S1_CANONICAL, CanonicalStructure
from generate.structure_mapping.canonicals import (
CANONICAL_BY_ID,
CANONICAL_LIBRARY,
S1_CANONICAL,
S2_CANONICAL,
S3_CANONICAL,
S4_CANONICAL,
CanonicalStructure,
)
from generate.structure_mapping.convert import graph_to_role_graph
from generate.structure_mapping.mapper import (
StructureMapRefuse,
StructureMapResult,
map_to_s1,
map_to_s2,
map_to_s3,
map_to_s4,
)
from generate.structure_mapping.pipeline import (
PipelineTrace,
run_structure_mapping_pipeline,
)
from generate.structure_mapping.role_predicate import (
PredicateKind,
@ -24,24 +39,45 @@ from generate.structure_mapping.role_predicate import (
RolePredicate,
RoleTerm,
)
from generate.structure_mapping.solve_s1 import (
from generate.structure_mapping.selector import SelectorResult, select_structure
from generate.structure_mapping.solve import (
S1SolveOutcome,
SolveOutcome,
graph_from_s1_binding,
solve_binding,
solve_s1_binding,
try_s1_structure_map_and_solve,
try_structure_map_and_solve,
)
__all__ = [
"CANONICAL_BY_ID",
"CANONICAL_LIBRARY",
"CanonicalStructure",
"PipelineTrace",
"PredicateKind",
"RoleGraph",
"RolePredicate",
"RoleTerm",
"S1_CANONICAL",
"S1SolveOutcome",
"S2_CANONICAL",
"S3_CANONICAL",
"S4_CANONICAL",
"SelectorResult",
"SolveOutcome",
"StructureMapRefuse",
"StructureMapResult",
"graph_from_s1_binding",
"graph_to_role_graph",
"map_to_s1",
"map_to_s2",
"map_to_s3",
"map_to_s4",
"run_structure_mapping_pipeline",
"select_structure",
"solve_binding",
"solve_s1_binding",
"try_s1_structure_map_and_solve",
"try_structure_map_and_solve",
]

75
generate/structure_mapping/canonicals.py
View file

@ -1,8 +1,7 @@
"""Canonical role-predicate skeletons (expert schema library).
Increment 1 ships **S1 only**: compare-multiplicative-then-total
(``compare(b, a, k) ∧ total(a, b, sum)``). Typed roles; no surface literals
as schema identity. S2–S4 deliberately absent until the S1 slice is reviewed.
Track B Increment 2: S1–S4. Each is a typed role skeleton with abstract
``var`` terms only — no surface literals as schema identity.
"""
from __future__ import annotations
@ -17,8 +16,8 @@ class CanonicalStructure:
"""A named schema with abstract role variables (not surface attributes)."""
structure_id: str
"""Stable id for the schema (e.g. ``S1``). Used only as prior-knowledge
library key — never read from gold labels at match time."""
"""Stable id for the schema (e.g. ``S1``). Library key only — never read
from gold labels at match time."""
description: str
predicates: tuple[RolePredicate, ...]
@ -31,6 +30,7 @@ def _var(name: str) -> RoleTerm:
# S1: "b is k× a; find a+b"
# compare(b, a, k) ∧ contain(a, a_qty) ∧ total(a, b, sum)
# (seed may equivalently be on b; mapper recovers a_value = b_value/k)
S1_CANONICAL: CanonicalStructure = CanonicalStructure(
structure_id="S1",
description=(
@ -52,3 +52,68 @@ S1_CANONICAL: CanonicalStructure = CanonicalStructure(
),
),
)
# S2: transfer-then-query — contain(src), contain(dst), transfer(src→dst, qty);
# query one endpoint after the transfer.
S2_CANONICAL: CanonicalStructure = CanonicalStructure(
structure_id="S2",
description=(
"transfer-then-query: src and dst seeded; qty moves src→dst; "
"query one endpoint's post-transfer amount."
),
predicates=(
RolePredicate(kind="contain", args=(_var("src"), _var("src_qty"))),
RolePredicate(kind="contain", args=(_var("dst"), _var("dst_qty"))),
RolePredicate(
kind="transfer",
args=(_var("src"), _var("dst"), _var("xfer_qty")),
),
),
)
# S3: rate-application — rate × count → total (or invert).
S3_CANONICAL: CanonicalStructure = CanonicalStructure(
structure_id="S3",
description=(
"rate-application: per_unit rate applied over count yields total. "
"Roles: rate_value, count, total_slot."
),
predicates=(
RolePredicate(
kind="rate",
args=(_var("rate_value"), _var("count"), _var("total_slot")),
),
),
)
# S4: additive-comparison-then-total — b = a + delta; query a+b.
# Uses compare_add (additive twin of multiplicative compare).
S4_CANONICAL: CanonicalStructure = CanonicalStructure(
structure_id="S4",
description=(
"additive-comparison-then-total: b = a + delta; query a+b. "
"Roles: a=reference, b=offset entity, delta=additive gap."
),
predicates=(
RolePredicate(kind="contain", args=(_var("a"), _var("a_qty"))),
RolePredicate(
kind="compare_add",
args=(_var("b"), _var("a"), _var("delta")),
),
RolePredicate(
kind="total",
args=(_var("a"), _var("b"), _var("sum")),
),
),
)
CANONICAL_LIBRARY: tuple[CanonicalStructure, ...] = (
S1_CANONICAL,
S2_CANONICAL,
S3_CANONICAL,
S4_CANONICAL,
)
CANONICAL_BY_ID: dict[str, CanonicalStructure] = {
c.structure_id: c for c in CANONICAL_LIBRARY
}

33
generate/structure_mapping/convert.py
View file

@ -28,6 +28,7 @@ def graph_to_role_graph(graph: MathProblemGraph) -> RoleGraph:
- ``initial_state`` → ``contain(entity, qty)``
- ``compare_multiplicative`` → ``compare(actor, reference, factor)``
(actor is k× reference — S1 role order ``compare(b, a, k)``)
- ``compare_additive`` (direction more/less) → ``compare_add(actor, reference, delta)``
- ``transfer`` → ``transfer(src, dst, qty)``
- ``apply_rate`` → ``rate`` with rate value / denominator-as-count placeholder
- ``unknown.entity is None`` → ``total`` over entities that hold ``unknown.unit``
@ -82,6 +83,34 @@ def graph_to_role_graph(graph: MathProblemGraph) -> RoleGraph:
),
)
)
elif op.kind == "compare_additive":
assert isinstance(op.operand, Comparison)
if op.operand.delta is None:
continue
# delta is a Quantity on the Comparison (ADR-0123).
delta_qty = op.operand.delta
if not isinstance(delta_qty, Quantity):
continue
delta = float(delta_qty.value)
# direction "fewer" means actor is below reference → flip sign so
# compare_add always means b = a + delta with delta signed.
if op.operand.direction == "fewer":
delta = -delta
preds.append(
RolePredicate(
kind="compare_add",
args=(
RoleTerm(kind="entity", name=op.actor), # b
RoleTerm(kind="entity", name=op.operand.reference_actor), # a
RoleTerm(
kind="quantity",
name="delta",
value=delta,
unit=delta_qty.unit,
),
),
)
)
elif op.kind == "transfer":
if not isinstance(op.operand, Quantity) or op.target is None:
continue
@ -150,7 +179,9 @@ def _entities_with_unit(graph: MathProblemGraph, unit: str) -> list[str]:
if poss.quantity.unit == unit and poss.entity not in seen:
seen.append(poss.entity)
for op in graph.operations:
if op.kind == "compare_multiplicative" and isinstance(op.operand, Comparison):
if op.kind in ("compare_multiplicative", "compare_additive") and isinstance(
op.operand, Comparison
):
# compare-defined entity inherits the reference unit; include actor
if op.actor not in seen:
seen.append(op.actor)

392
generate/structure_mapping/mapper.py
View file

@ -1,17 +1,11 @@
"""Symbolic relational structure-mapper — S1 vertical slice (Track B).
"""Symbolic relational structure-mapper — S1–S4 pure-family gates (Track B).
Aligns a problem's role-predicate graph to the S1 canonical by **relational
role matching** (predicate kinds + systematicity). Surface attributes
(entity names, numbers) become the binding; they do not drive the match.
Aligns a problem's role-predicate graph to a canonical by **relational role
matching**. Surface attributes become the binding; they do not drive the match.
Blind: never reads gold structure labels. Refuse is a first-class outcome.
Deterministic: single compare-total pattern, fixed role assignment rules.
Pure-S1 gate (Increment 1): the problem must be *exactly* two roles
``a`` (seeded) and ``b`` (defined by ``k×a``) with total over ``{a,b}``.
A total that merely *includes* ``a`` and ``b`` among other parts, or any
seeded ``contain`` on a third entity, is refused — never bound and rebuilt
as a smaller pure-S1 problem (that would emit a wrong certified answer).
Each family has pure-family gates (exact role/part-set matching; refuse
supersets and stray predicates) so none can rebuild a wrong subproblem.
"""
from __future__ import annotations
@ -19,55 +13,54 @@ from __future__ import annotations
from dataclasses import dataclass
from typing import Mapping
from generate.structure_mapping.canonicals import S1_CANONICAL
from generate.structure_mapping.canonicals import (
S1_CANONICAL,
S2_CANONICAL,
S3_CANONICAL,
S4_CANONICAL,
)
from generate.structure_mapping.role_predicate import RoleGraph
@dataclass(frozen=True, slots=True)
class StructureMapResult:
"""Successful map onto S1 with variable binding."""
"""Successful map onto one canonical with variable binding."""
maps_to_s1: bool
"""Always True on this type; present for the public ``(bool, binding)`` contract."""
structure_id: str
"""Matched canonical id (library key from the match, not a gold label)."""
binding: Mapping[str, object]
"""Roles: ``a`` (ref entity), ``b`` (scaled entity), ``k`` (factor float),
``a_value`` (float), ``unit`` (str), ``a_qty_term`` reserved."""
matched_predicates: tuple[str, ...]
"""Predicate kinds that participated (for right-reason audits)."""
# Back-compat with Increment 1 callers
@property
def maps_to_s1(self) -> bool:
return self.structure_id == "S1"
@dataclass(frozen=True, slots=True)
class StructureMapRefuse:
"""No admissible S1 alignment."""
"""No admissible alignment for the requested family (or all families)."""
reason: str
structure_id: str | None = None
def map_to_s1(role_graph: RoleGraph) -> StructureMapResult | StructureMapRefuse:
"""Map ``role_graph`` to the S1 canonical or refuse.
Algorithm (deterministic, systematicity-first, pure-S1 only):
1. Require exactly one ``compare`` predicate (multiplicative factor present).
2. Require exactly one admissible ``total`` whose **part-set equals**
``{a, b}`` (not a superset — extras would be dropped by rebuild solve).
3. Require a ``contain`` on the compare reference entity (role ``a``) with
a known quantity — the seed.
4. Refuse any ``contain`` whose entity is not ``a`` (including ``b`` and
any third entity ``c``). ``b`` is defined only via the compare.
5. Refuse on transfer/rate co-presence (multi-frontier; out of Increment 1).
Gold labels are never consulted.
Pure-S1 gate:
1. Exactly one multiplicative ``compare``; no transfer/rate/compare_add.
2. Exactly one ``total`` whose part-set equals ``{a, b}``.
3. Seed ``contain`` on ``a`` **or** on ``b`` (not both, not a third).
If only ``b`` is seeded, ``a_value = b_value / k`` (inverted seed).
4. No extra contains.
"""
if not isinstance(role_graph, RoleGraph):
raise TypeError(
f"map_to_s1 expects RoleGraph, got {type(role_graph).__name__}"
)
# Deliberately unused: S1_CANONICAL is the prior schema we match *against*
# by construction of the rules below (same predicate multiset). Referencing
# it keeps the library load-bearing and prevents "orphan canonical" drift.
_ = S1_CANONICAL.structure_id
compares = role_graph.of_kind("compare")
@ -75,38 +68,38 @@ def map_to_s1(role_graph: RoleGraph) -> StructureMapResult | StructureMapRefuse:
contains = role_graph.of_kind("contain")
transfers = role_graph.of_kind("transfer")
rates = role_graph.of_kind("rate")
compare_adds = role_graph.of_kind("compare_add")
if transfers:
return StructureMapRefuse(reason="non_s1_has_transfer")
return StructureMapRefuse(reason="non_s1_has_transfer", structure_id="S1")
if rates:
return StructureMapRefuse(reason="non_s1_has_rate")
return StructureMapRefuse(reason="non_s1_has_rate", structure_id="S1")
if compare_adds:
return StructureMapRefuse(reason="non_s1_has_compare_add", structure_id="S1")
if len(compares) != 1:
return StructureMapRefuse(
reason=f"compare_count_not_one:{len(compares)}"
reason=f"compare_count_not_one:{len(compares)}", structure_id="S1"
)
if not totals:
return StructureMapRefuse(reason="missing_total_query")
return StructureMapRefuse(reason="missing_total_query", structure_id="S1")
compare = compares[0]
b_term, a_term, k_term = compare.args
if a_term.kind != "entity" or b_term.kind != "entity":
return StructureMapRefuse(reason="compare_args_not_entities")
return StructureMapRefuse(reason="compare_args_not_entities", structure_id="S1")
if k_term.kind != "quantity" or k_term.value is None:
return StructureMapRefuse(reason="compare_factor_missing")
return StructureMapRefuse(reason="compare_factor_missing", structure_id="S1")
if k_term.value <= 0:
return StructureMapRefuse(reason="compare_factor_non_positive")
return StructureMapRefuse(reason="compare_factor_non_positive", structure_id="S1")
a_name = a_term.name
b_name = b_term.name
if a_name == b_name:
return StructureMapRefuse(reason="compare_self_reference")
return StructureMapRefuse(reason="compare_self_reference", structure_id="S1")
k = float(k_term.value)
expected_parts = frozenset({a_name, b_name})
# Pure S1: total part-set must equal {a, b} exactly — not a superset.
# A total that merely includes a,b among other parts is multi-entity and
# must refuse; rebuilding a+b alone would emit the wrong problem's answer.
exact_total = False
for total in totals:
part_names = frozenset(
@ -116,18 +109,21 @@ def map_to_s1(role_graph: RoleGraph) -> StructureMapResult | StructureMapRefuse:
exact_total = True
break
if not exact_total:
# Distinguish "missing roles" vs "extra parts" for auditability.
any_cover = any(
a_name in (t.name for t in total.args[:-1] if t.kind == "entity")
and b_name in (t.name for t in total.args[:-1] if t.kind == "entity")
for total in totals
)
if any_cover:
return StructureMapRefuse(reason="total_parts_not_exactly_a_b")
return StructureMapRefuse(reason="total_does_not_cover_compare_roles")
return StructureMapRefuse(
reason="total_parts_not_exactly_a_b", structure_id="S1"
)
return StructureMapRefuse(
reason="total_does_not_cover_compare_roles", structure_id="S1"
)
# contain(a, qty) required; no other entity may carry a seed contain.
a_value: float | None = None
b_value: float | None = None
unit: str | None = None
for c in contains:
ent, qty = c.args
@ -139,17 +135,39 @@ def map_to_s1(role_graph: RoleGraph) -> StructureMapResult | StructureMapRefuse:
a_value = float(qty.value)
unit = qty.unit
continue
# Any contain on b or a third entity is not pure S1.
if ent.name == b_name:
return StructureMapRefuse(
reason="b_has_independent_contain_not_pure_s1"
)
b_value = float(qty.value)
if unit is None:
unit = qty.unit
continue
return StructureMapRefuse(
reason=f"extra_contain_not_pure_s1:{ent.name}"
reason=f"extra_contain_not_pure_s1:{ent.name}", structure_id="S1"
)
if a_value is not None and b_value is not None:
# Both seeded: only admissible if consistent with b = k×a.
expected_b = a_value * k
if abs(expected_b - b_value) > 1e-6 * max(1.0, abs(expected_b)):
return StructureMapRefuse(
reason="dual_seed_inconsistent_with_compare", structure_id="S1"
)
# Prefer a-seeded pure form.
b_value = None
if a_value is None and b_value is None:
return StructureMapRefuse(
reason="missing_contain_on_reference_a", structure_id="S1"
)
if a_value is None:
return StructureMapRefuse(reason="missing_contain_on_reference_a")
# Inverted seed: b known, a = b/k.
assert b_value is not None
if abs(k) < 1e-12:
return StructureMapRefuse(reason="compare_factor_near_zero", structure_id="S1")
a_value = float(b_value) / k
seed_mode = "b_seeded"
else:
seed_mode = "a_seeded"
binding: dict[str, object] = {
"a": a_name,
@ -158,15 +176,275 @@ def map_to_s1(role_graph: RoleGraph) -> StructureMapResult | StructureMapRefuse:
"a_value": a_value,
"unit": unit or "",
"canonical_id": S1_CANONICAL.structure_id,
"seed_mode": seed_mode,
}
return StructureMapResult(
maps_to_s1=True,
structure_id="S1",
binding=binding,
matched_predicates=("contain", "compare", "total"),
)
def map_to_s2(role_graph: RoleGraph) -> StructureMapResult | StructureMapRefuse:
"""Pure S2: exactly one transfer, contains on src+dst only, query one endpoint.
No compare / compare_add / rate / total (total would be multi-entity sum;
S2 queries a single entity after transfer).
"""
if not isinstance(role_graph, RoleGraph):
raise TypeError(
f"map_to_s2 expects RoleGraph, got {type(role_graph).__name__}"
)
_ = S2_CANONICAL.structure_id
transfers = role_graph.of_kind("transfer")
contains = role_graph.of_kind("contain")
compares = role_graph.of_kind("compare")
compare_adds = role_graph.of_kind("compare_add")
rates = role_graph.of_kind("rate")
totals = role_graph.of_kind("total")
if compares:
return StructureMapRefuse(reason="non_s2_has_compare", structure_id="S2")
if compare_adds:
return StructureMapRefuse(reason="non_s2_has_compare_add", structure_id="S2")
if rates:
return StructureMapRefuse(reason="non_s2_has_rate", structure_id="S2")
if totals:
return StructureMapRefuse(reason="non_s2_has_total", structure_id="S2")
if len(transfers) != 1:
return StructureMapRefuse(
reason=f"transfer_count_not_one:{len(transfers)}", structure_id="S2"
)
xfer = transfers[0]
src_t, dst_t, qty_t = xfer.args
if src_t.kind != "entity" or dst_t.kind != "entity":
return StructureMapRefuse(reason="transfer_endpoints_not_entities", structure_id="S2")
if qty_t.kind != "quantity" or qty_t.value is None:
return StructureMapRefuse(reason="transfer_qty_missing", structure_id="S2")
if qty_t.value < 0:
return StructureMapRefuse(reason="transfer_qty_negative", structure_id="S2")
src, dst = src_t.name, dst_t.name
if src == dst:
return StructureMapRefuse(reason="transfer_self", structure_id="S2")
seeds: dict[str, tuple[float, str | None]] = {}
for c in contains:
ent, qty = c.args
if ent.kind != "entity" or qty.kind != "quantity" or qty.value is None:
continue
if ent.name not in (src, dst):
return StructureMapRefuse(
reason=f"extra_contain_not_pure_s2:{ent.name}", structure_id="S2"
)
seeds[ent.name] = (float(qty.value), qty.unit)
if src not in seeds or dst not in seeds:
return StructureMapRefuse(reason="missing_src_or_dst_seed", structure_id="S2")
# Query entity: inferred from a single unknown-less graph later; for pure
# role match we require the binding to name both post-transfer values.
# Default query = dst (most common "how many does Y have?" after receive).
# Callers that know the unknown entity may override via binding later.
src_val, unit = seeds[src]
dst_val, unit_d = seeds[dst]
unit = unit or unit_d or ""
xfer_qty = float(qty_t.value)
binding: dict[str, object] = {
"src": src,
"dst": dst,
"xfer_qty": xfer_qty,
"src_value": src_val,
"dst_value": dst_val,
"unit": unit,
"query_entity": dst, # default; solve may override from original graph
"canonical_id": S2_CANONICAL.structure_id,
}
return StructureMapResult(
structure_id="S2",
binding=binding,
matched_predicates=("contain", "contain", "transfer"),
)
def map_to_s3(role_graph: RoleGraph) -> StructureMapResult | StructureMapRefuse:
"""Pure S3: exactly one rate predicate; no transfer/compare/total pollution."""
if not isinstance(role_graph, RoleGraph):
raise TypeError(
f"map_to_s3 expects RoleGraph, got {type(role_graph).__name__}"
)
_ = S3_CANONICAL.structure_id
rates = role_graph.of_kind("rate")
if len(rates) != 1:
return StructureMapRefuse(
reason=f"rate_count_not_one:{len(rates)}", structure_id="S3"
)
if role_graph.of_kind("transfer"):
return StructureMapRefuse(reason="non_s3_has_transfer", structure_id="S3")
if role_graph.of_kind("compare"):
return StructureMapRefuse(reason="non_s3_has_compare", structure_id="S3")
if role_graph.of_kind("compare_add"):
return StructureMapRefuse(reason="non_s3_has_compare_add", structure_id="S3")
if role_graph.of_kind("total"):
return StructureMapRefuse(reason="non_s3_has_total", structure_id="S3")
rate = rates[0]
per, count, total_slot = rate.args
if per.kind != "quantity" or per.value is None:
return StructureMapRefuse(reason="rate_value_missing", structure_id="S3")
# count may be entity or quantity depending on converter
count_value: float | None = None
count_name = count.name
if count.kind == "quantity" and count.value is not None:
count_value = float(count.value)
# Look for contain on count entity for the duration/count seed
if count_value is None and count.kind == "entity":
for c in role_graph.of_kind("contain"):
ent, qty = c.args
if ent.kind == "entity" and ent.name == count.name and qty.value is not None:
count_value = float(qty.value)
break
if count_value is None:
return StructureMapRefuse(reason="rate_count_missing", structure_id="S3")
if per.value <= 0 or count_value <= 0:
return StructureMapRefuse(reason="rate_non_positive", structure_id="S3")
binding: dict[str, object] = {
"rate_value": float(per.value),
"count": count_value,
"count_name": count_name,
"unit": per.unit or (total_slot.unit if total_slot.kind == "quantity" else "") or "",
"canonical_id": S3_CANONICAL.structure_id,
}
return StructureMapResult(
structure_id="S3",
binding=binding,
matched_predicates=("rate",),
)
def map_to_s4(role_graph: RoleGraph) -> StructureMapResult | StructureMapRefuse:
"""Pure S4: one compare_add, total over {a,b}, seed contain on a (or b)."""
if not isinstance(role_graph, RoleGraph):
raise TypeError(
f"map_to_s4 expects RoleGraph, got {type(role_graph).__name__}"
)
_ = S4_CANONICAL.structure_id
if role_graph.of_kind("transfer"):
return StructureMapRefuse(reason="non_s4_has_transfer", structure_id="S4")
if role_graph.of_kind("rate"):
return StructureMapRefuse(reason="non_s4_has_rate", structure_id="S4")
if role_graph.of_kind("compare"):
return StructureMapRefuse(reason="non_s4_has_multiplicative_compare", structure_id="S4")
adds = role_graph.of_kind("compare_add")
totals = role_graph.of_kind("total")
contains = role_graph.of_kind("contain")
if len(adds) != 1:
return StructureMapRefuse(
reason=f"compare_add_count_not_one:{len(adds)}", structure_id="S4"
)
if not totals:
return StructureMapRefuse(reason="missing_total_query", structure_id="S4")
cad = adds[0]
b_term, a_term, d_term = cad.args
if a_term.kind != "entity" or b_term.kind != "entity":
return StructureMapRefuse(reason="compare_add_args_not_entities", structure_id="S4")
if d_term.kind != "quantity" or d_term.value is None:
return StructureMapRefuse(reason="compare_add_delta_missing", structure_id="S4")
a_name, b_name = a_term.name, b_term.name
if a_name == b_name:
return StructureMapRefuse(reason="compare_add_self_reference", structure_id="S4")
delta = float(d_term.value)
expected_parts = frozenset({a_name, b_name})
if not any(
frozenset(t.name for t in total.args[:-1] if t.kind == "entity") == expected_parts
for total in totals
):
return StructureMapRefuse(reason="total_parts_not_exactly_a_b", structure_id="S4")
a_value: float | None = None
b_value: float | None = None
unit: str | None = None
for c in contains:
ent, qty = c.args
if ent.kind != "entity" or qty.kind != "quantity" or qty.value is None:
continue
if ent.name == a_name:
a_value = float(qty.value)
unit = qty.unit
elif ent.name == b_name:
b_value = float(qty.value)
if unit is None:
unit = qty.unit
else:
return StructureMapRefuse(
reason=f"extra_contain_not_pure_s4:{ent.name}", structure_id="S4"
)
if a_value is None and b_value is None:
return StructureMapRefuse(reason="missing_seed_for_s4", structure_id="S4")
if a_value is None:
assert b_value is not None
a_value = b_value - delta
seed_mode = "b_seeded"
elif b_value is None:
seed_mode = "a_seeded"
else:
if abs((a_value + delta) - b_value) > 1e-6 * max(1.0, abs(b_value)):
return StructureMapRefuse(
reason="dual_seed_inconsistent_with_compare_add", structure_id="S4"
)
seed_mode = "a_seeded"
binding: dict[str, object] = {
"a": a_name,
"b": b_name,
"delta": delta,
"a_value": a_value,
"unit": unit or "",
"canonical_id": S4_CANONICAL.structure_id,
"seed_mode": seed_mode,
}
return StructureMapResult(
structure_id="S4",
binding=binding,
matched_predicates=("contain", "compare_add", "total"),
)
def binding_roles_only(binding: Mapping[str, object]) -> dict[str, object]:
"""Return the solve-relevant subset of a binding (no label fields)."""
keys = ("a", "b", "k", "a_value", "unit")
keys = (
"a",
"b",
"k",
"a_value",
"unit",
"src",
"dst",
"xfer_qty",
"src_value",
"dst_value",
"query_entity",
"rate_value",
"count",
"delta",
)
return {k: binding[k] for k in keys if k in binding}
MAPPERS = {
"S1": map_to_s1,
"S2": map_to_s2,
"S3": map_to_s3,
"S4": map_to_s4,
}

83
generate/structure_mapping/pipeline.py Normal file
View file

@ -0,0 +1,83 @@
"""End-to-end off-serving structure-mapping pipeline.
``text → (serving parse | SM-owned extract) → role graph → selector → solve``
Serving reader is tried first; SM-owned extractors run only on refuse, and
never feed the live organ path. Gold labels never enter.
"""
from __future__ import annotations
from dataclasses import dataclass
from generate.math_candidate_graph import parse_and_solve
from generate.math_problem_graph import MathProblemGraph
from generate.structure_mapping.convert import graph_to_role_graph
from generate.structure_mapping.solve import SolveOutcome, try_structure_map_and_solve
from generate.structure_mapping.text_extract import extract_pure_s1
@dataclass(frozen=True, slots=True)
class PipelineTrace:
"""Diagnostics for measurement scripts (not a gold label)."""
organ_parsed: bool
organ_answer: float | None
organ_refusal: str | None
extract_used: bool
extract_pattern: str | None
graph_source: str # "organ" | "sm_extract" | "none"
def run_structure_mapping_pipeline(
problem: str,
*,
families: tuple[str, ...] = ("S1", "S2", "S3", "S4"),
allow_sm_extract: bool = True,
) -> tuple[SolveOutcome, PipelineTrace, MathProblemGraph | None]:
"""Full off-serving path for one problem text.
Returns ``(outcome, trace, graph_used)``.
"""
organ = parse_and_solve(problem)
graph: MathProblemGraph | None = organ.selected_graph
extract_used = False
extract_pattern: str | None = None
graph_source = "organ" if graph is not None else "none"
if graph is None and allow_sm_extract:
ext = extract_pure_s1(problem)
if ext.graph is not None:
graph = ext.graph
extract_used = True
extract_pattern = ext.pattern_id
graph_source = "sm_extract"
trace = PipelineTrace(
organ_parsed=organ.selected_graph is not None,
organ_answer=organ.answer,
organ_refusal=organ.refusal_reason,
extract_used=extract_used,
extract_pattern=extract_pattern,
graph_source=graph_source,
)
if graph is None:
return (
SolveOutcome(
emitted=False,
answer=None,
refusal_reason="parser_frontier:no_graph",
binding=None,
derivation=None,
multi_register_certified=False,
classical_verified=False,
),
trace,
None,
)
# Optional: convert for hash; solve path re-converts.
_ = graph_to_role_graph(graph)
out = try_structure_map_and_solve(graph, families=families)
return out, trace, graph

6
generate/structure_mapping/role_predicate.py
View file

@ -17,12 +17,13 @@ PredicateKind = Literal[
"contain",
"transfer",
"compare",
"compare_add",
"rate",
"total",
]
VALID_PREDICATE_KINDS: Final[frozenset[str]] = frozenset(
{"contain", "transfer", "compare", "rate", "total"}
{"contain", "transfer", "compare", "compare_add", "rate", "total"}
)
@ -61,6 +62,7 @@ class RolePredicate:
- contain(entity, qty)
- transfer(src, dst, qty)
- compare(b, a, k) — b is k× a (actor, reference, factor)
- compare_add(b, a, delta) — b = a + delta (actor, reference, gap)
- rate(per_unit, count, total)
- total(part_1, ..., part_n, sum_query)
"""
@ -79,6 +81,8 @@ class RolePredicate:
raise ValueError("transfer requires (src, dst, qty)")
if self.kind == "compare" and len(self.args) != 3:
raise ValueError("compare requires (b, a, k)")
if self.kind == "compare_add" and len(self.args) != 3:
raise ValueError("compare_add requires (b, a, delta)")
if self.kind == "rate" and len(self.args) != 3:
raise ValueError("rate requires (per_unit, count, total)")
if self.kind == "total" and len(self.args) < 2:

106
generate/structure_mapping/selector.py Normal file
View file

@ -0,0 +1,106 @@
"""Overlapping-waves selector (ADR-0252 stage 5 / revive ADR-0174).
Given a role-graph, hold candidate canonical mappings, rank them, pick the
best, **refuse on ambiguity/tie**, and suppress surface-similarity as the
decision driver (structure predicates only).
Blind to gold labels. Deterministic ranking.
"""
from __future__ import annotations
from dataclasses import dataclass
from generate.structure_mapping.mapper import (
MAPPERS,
StructureMapRefuse,
StructureMapResult,
)
from generate.structure_mapping.role_predicate import RoleGraph
# Systematicity / specificity scores: more constraining pure-family matches
# rank higher. Ties at the same score → refuse.
_FAMILY_SCORE: dict[str, int] = {
# Multi-predicate pure families outrank single-predicate ones.
"S1": 30, # compare + contain + total
"S4": 30, # compare_add + contain + total
"S2": 25, # transfer + 2 contains
"S3": 10, # single rate
}
@dataclass(frozen=True, slots=True)
class SelectorResult:
"""Selected canonical mapping, or a refuse with diagnostics."""
selected: StructureMapResult | None
refused: bool
reason: str | None
candidates: tuple[StructureMapResult, ...]
refused_families: tuple[tuple[str, str], ...]
"""(structure_id, refuse_reason) for families that did not match."""
def select_structure(
role_graph: RoleGraph,
*,
families: tuple[str, ...] = ("S1", "S2", "S3", "S4"),
) -> SelectorResult:
"""Rank pure-family mappers; pick unique best; refuse on empty or tie.
Surface attributes (entity names, numbers) never affect ranking — only
which pure-family mappers admit the role graph and their fixed scores.
"""
if not isinstance(role_graph, RoleGraph):
raise TypeError(
f"select_structure expects RoleGraph, got {type(role_graph).__name__}"
)
hits: list[StructureMapResult] = []
refuses: list[tuple[str, str]] = []
for fam in families:
mapper = MAPPERS.get(fam)
if mapper is None:
refuses.append((fam, "unknown_family"))
continue
out = mapper(role_graph)
if isinstance(out, StructureMapResult):
hits.append(out)
else:
assert isinstance(out, StructureMapRefuse)
refuses.append((fam, out.reason))
if not hits:
return SelectorResult(
selected=None,
refused=True,
reason="no_family_matched",
candidates=(),
refused_families=tuple(refuses),
)
# Rank by systematicity score; secondary key = structure_id for stability.
ranked = sorted(
hits,
key=lambda m: (-_FAMILY_SCORE.get(m.structure_id, 0), m.structure_id),
)
best = ranked[0]
best_score = _FAMILY_SCORE.get(best.structure_id, 0)
tied = [m for m in ranked if _FAMILY_SCORE.get(m.structure_id, 0) == best_score]
if len(tied) > 1:
ids = ",".join(sorted(m.structure_id for m in tied))
return SelectorResult(
selected=None,
refused=True,
reason=f"ambiguous_tie:{ids}",
candidates=tuple(ranked),
refused_families=tuple(refuses),
)
return SelectorResult(
selected=best,
refused=False,
reason=None,
candidates=tuple(ranked),
refused_families=tuple(refuses),
)

498
generate/structure_mapping/solve.py Normal file
View file

@ -0,0 +1,498 @@
"""Solve structure-mapped bindings through the certificate corridor.
Generalizes Increment 1's three-gate ``wrong=0`` pattern:
1. Map-time pure-family exact gate (in mapper)
2. Classical ``solve``+``verify`` AND multi-register certificate agreement
3. Original-graph answer agreement backstop when the original graph is supplied
Off-serving. Does not touch the live reader dispatch.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Mapping
from evals.multi_register_program import (
MultiRegisterError,
compile_multi_register_program,
execute_multi_register_program,
)
from generate.math_problem_graph import (
Comparison,
InitialPossession,
MathProblemGraph,
Operation,
Quantity,
Rate,
Unknown,
)
from generate.math_solver import SolveError, solve
from generate.math_verifier import VerificationError, verify
from generate.structure_mapping.convert import graph_to_role_graph
from generate.structure_mapping.mapper import StructureMapRefuse, StructureMapResult
from generate.structure_mapping.role_predicate import RoleGraph
from generate.structure_mapping.selector import SelectorResult, select_structure
@dataclass(frozen=True, slots=True)
class SolveOutcome:
"""Emit or refuse for one structure-mapped problem."""
emitted: bool
answer: float | None
refusal_reason: str | None
binding: Mapping[str, object] | None
derivation: str | None
multi_register_certified: bool
classical_verified: bool
structure_id: str | None = None
source_graph_hash: str | None = None
# Back-compat alias used by Increment 1 imports/tests.
S1SolveOutcome = SolveOutcome
def graph_from_s1_binding(binding: Mapping[str, object]) -> MathProblemGraph:
"""Rebuild a pure S1 :class:`MathProblemGraph` from a role binding."""
a = str(binding["a"])
b = str(binding["b"])
k = float(binding["k"]) # type: ignore[arg-type]
a_value = float(binding["a_value"]) # type: ignore[arg-type]
unit = str(binding.get("unit") or "units")
return MathProblemGraph(
entities=(a, b),
initial_state=(
InitialPossession(entity=a, quantity=Quantity(value=a_value, unit=unit)),
),
operations=(
Operation(
actor=b,
kind="compare_multiplicative",
operand=Comparison(
reference_actor=a,
delta=None,
factor=k,
direction="times",
),
),
),
unknown=Unknown(entity=None, unit=unit),
)
def graph_from_s2_binding(binding: Mapping[str, object]) -> MathProblemGraph:
src = str(binding["src"])
dst = str(binding["dst"])
xfer = float(binding["xfer_qty"]) # type: ignore[arg-type]
src_val = float(binding["src_value"]) # type: ignore[arg-type]
dst_val = float(binding["dst_value"]) # type: ignore[arg-type]
unit = str(binding.get("unit") or "units")
query = str(binding.get("query_entity") or dst)
return MathProblemGraph(
entities=(src, dst),
initial_state=(
InitialPossession(entity=src, quantity=Quantity(value=src_val, unit=unit)),
InitialPossession(entity=dst, quantity=Quantity(value=dst_val, unit=unit)),
),
operations=(
Operation(
actor=src,
kind="transfer",
operand=Quantity(value=xfer, unit=unit),
target=dst,
),
),
unknown=Unknown(entity=query, unit=unit),
)
def graph_from_s3_binding(binding: Mapping[str, object]) -> MathProblemGraph:
rate_value = float(binding["rate_value"]) # type: ignore[arg-type]
count = float(binding["count"]) # type: ignore[arg-type]
unit = str(binding.get("unit") or "units")
actor = str(binding.get("count_name") or "actor")
return MathProblemGraph(
entities=(actor,),
initial_state=(
InitialPossession(
entity=actor, quantity=Quantity(value=count, unit="units")
),
),
operations=(
Operation(
actor=actor,
kind="apply_rate",
operand=Rate(
value=rate_value,
numerator_unit=unit,
denominator_unit="units",
),
),
),
unknown=Unknown(entity=actor, unit=unit),
)
def graph_from_s4_binding(binding: Mapping[str, object]) -> MathProblemGraph:
a = str(binding["a"])
b = str(binding["b"])
delta = float(binding["delta"]) # type: ignore[arg-type]
a_value = float(binding["a_value"]) # type: ignore[arg-type]
unit = str(binding.get("unit") or "units")
direction = "more" if delta >= 0 else "fewer"
abs_delta = abs(delta)
return MathProblemGraph(
entities=(a, b),
initial_state=(
InitialPossession(entity=a, quantity=Quantity(value=a_value, unit=unit)),
),
operations=(
Operation(
actor=b,
kind="compare_additive",
operand=Comparison(
reference_actor=a,
delta=Quantity(value=abs_delta, unit=unit),
factor=None,
direction=direction,
),
),
),
unknown=Unknown(entity=None, unit=unit),
)
def _graph_from_binding(
structure_id: str, binding: Mapping[str, object]
) -> MathProblemGraph:
if structure_id == "S1":
return graph_from_s1_binding(binding)
if structure_id == "S2":
return graph_from_s2_binding(binding)
if structure_id == "S3":
return graph_from_s3_binding(binding)
if structure_id == "S4":
return graph_from_s4_binding(binding)
raise ValueError(f"unknown structure_id for rebuild: {structure_id!r}")
def _derivation(structure_id: str, binding: Mapping[str, object]) -> str:
if structure_id == "S1":
a, b = binding["a"], binding["b"]
k, a_value = float(binding["k"]), float(binding["a_value"]) # type: ignore[arg-type]
total = a_value * (1.0 + k)
return (
f"S1: {b} = {k} × {a}; {a} = {a_value}; "
f"total = {a} + {b} = {a_value} + {k}*{a_value} = {total}"
)
if structure_id == "S2":
src, dst = binding["src"], binding["dst"]
xfer = float(binding["xfer_qty"]) # type: ignore[arg-type]
src_v, dst_v = float(binding["src_value"]), float(binding["dst_value"]) # type: ignore[arg-type]
q = binding.get("query_entity") or dst
post_src, post_dst = src_v - xfer, dst_v + xfer
ans = post_dst if q == dst else post_src
return (
f"S2: transfer {xfer} {src}→{dst}; "
f"pre ({src}={src_v}, {dst}={dst_v}); "
f"post ({src}={post_src}, {dst}={post_dst}); query {q}={ans}"
)
if structure_id == "S3":
r, c = float(binding["rate_value"]), float(binding["count"]) # type: ignore[arg-type]
return f"S3: total = rate × count = {r} × {c} = {r * c}"
if structure_id == "S4":
a, b = binding["a"], binding["b"]
d, a_value = float(binding["delta"]), float(binding["a_value"]) # type: ignore[arg-type]
b_value = a_value + d
total = a_value + b_value
return (
f"S4: {b} = {a} + {d}; {a} = {a_value}; {b} = {b_value}; "
f"total = {a_value}+{b_value} = {total}"
)
return f"{structure_id}: (no derivation template)"
def solve_binding(
structure_id: str, binding: Mapping[str, object]
) -> SolveOutcome:
"""Solve via multi-register + classical corridors; refuse if uncertified."""
try:
graph = _graph_from_binding(structure_id, binding)
except (KeyError, TypeError, ValueError) as exc:
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason=f"binding_incomplete:{type(exc).__name__}:{exc}",
binding=dict(binding),
derivation=None,
multi_register_certified=False,
classical_verified=False,
structure_id=structure_id,
)
derivation = _derivation(structure_id, binding)
classical_verified = False
classical_answer: float | None = None
try:
trace = solve(graph)
verdict = verify(graph, trace)
classical_verified = bool(verdict.passed)
classical_answer = float(trace.answer_value)
except (SolveError, VerificationError, ValueError) as exc:
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason=f"classical_solve_or_verify_failed:{type(exc).__name__}:{exc}",
binding=dict(binding),
derivation=derivation,
multi_register_certified=False,
classical_verified=False,
structure_id=structure_id,
)
if not classical_verified:
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason="classical_verifier_rejected",
binding=dict(binding),
derivation=derivation,
multi_register_certified=False,
classical_verified=False,
structure_id=structure_id,
)
mr_certified = False
mr_answer: float | None = None
try:
program = compile_multi_register_program(graph)
outcome = execute_multi_register_program(program)
mr_certified = bool(outcome.certified)
mr_answer = float(outcome.answer)
except MultiRegisterError as exc:
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason=f"multi_register_error:{exc.reason}",
binding=dict(binding),
derivation=derivation,
multi_register_certified=False,
classical_verified=True,
structure_id=structure_id,
)
except Exception as exc: # noqa: BLE001 — refuse, never crash-as-signal
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason=f"multi_register_unexpected:{type(exc).__name__}:{exc}",
binding=dict(binding),
derivation=derivation,
multi_register_certified=False,
classical_verified=True,
structure_id=structure_id,
)
if not mr_certified:
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason="multi_register_chain_not_certified",
binding=dict(binding),
derivation=derivation,
multi_register_certified=False,
classical_verified=True,
structure_id=structure_id,
)
if classical_answer is None or mr_answer is None:
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason="missing_answer_after_cert",
binding=dict(binding),
derivation=derivation,
multi_register_certified=mr_certified,
classical_verified=classical_verified,
structure_id=structure_id,
)
if abs(classical_answer - mr_answer) > 1e-6 * max(1.0, abs(classical_answer)):
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason=(
f"corridor_disagreement:classical={classical_answer},mr={mr_answer}"
),
binding=dict(binding),
derivation=derivation,
multi_register_certified=mr_certified,
classical_verified=classical_verified,
structure_id=structure_id,
)
return SolveOutcome(
emitted=True,
answer=classical_answer,
refusal_reason=None,
binding=dict(binding),
derivation=derivation,
multi_register_certified=True,
classical_verified=True,
structure_id=structure_id,
)
def solve_s1_binding(binding: Mapping[str, object]) -> SolveOutcome:
"""Increment 1 API: solve an S1 binding."""
return solve_binding("S1", binding)
def _original_graph_backstop(
out: SolveOutcome,
graph: MathProblemGraph,
role_graph: RoleGraph,
) -> SolveOutcome:
if not (out.emitted and out.answer is not None):
return SolveOutcome(
emitted=out.emitted,
answer=out.answer,
refusal_reason=out.refusal_reason,
binding=out.binding,
derivation=out.derivation,
multi_register_certified=out.multi_register_certified,
classical_verified=out.classical_verified,
structure_id=out.structure_id,
source_graph_hash=role_graph.source_graph_hash,
)
try:
original_trace = solve(graph)
original_verdict = verify(graph, original_trace)
except (SolveError, VerificationError, ValueError) as exc:
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason=f"original_graph_solve_failed:{type(exc).__name__}:{exc}",
binding=out.binding,
derivation=out.derivation,
multi_register_certified=out.multi_register_certified,
classical_verified=False,
structure_id=out.structure_id,
source_graph_hash=role_graph.source_graph_hash,
)
if not original_verdict.passed:
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason="original_graph_verifier_rejected",
binding=out.binding,
derivation=out.derivation,
multi_register_certified=out.multi_register_certified,
classical_verified=False,
structure_id=out.structure_id,
source_graph_hash=role_graph.source_graph_hash,
)
orig_ans = float(original_trace.answer_value)
if abs(orig_ans - float(out.answer)) > 1e-6 * max(1.0, abs(orig_ans)):
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason=(
f"original_graph_disagreement:original={orig_ans},"
f"rebuild={out.answer}"
),
binding=out.binding,
derivation=out.derivation,
multi_register_certified=out.multi_register_certified,
classical_verified=out.classical_verified,
structure_id=out.structure_id,
source_graph_hash=role_graph.source_graph_hash,
)
return SolveOutcome(
emitted=out.emitted,
answer=out.answer,
refusal_reason=out.refusal_reason,
binding=out.binding,
derivation=out.derivation,
multi_register_certified=out.multi_register_certified,
classical_verified=out.classical_verified,
structure_id=out.structure_id,
source_graph_hash=role_graph.source_graph_hash,
)
def try_structure_map_and_solve(
graph: MathProblemGraph | None = None,
*,
role_graph: RoleGraph | None = None,
families: tuple[str, ...] = ("S1", "S2", "S3", "S4"),
) -> SolveOutcome:
"""Convert (if needed) → selector → solve corridor.
Never takes a gold structure label. When the original graph is supplied,
the original-graph integrity backstop runs after a successful emit.
For S2, if the original graph's ``unknown.entity`` is set, it overrides
the default query endpoint before solve.
"""
if role_graph is None:
if graph is None:
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason="no_graph_provided",
binding=None,
derivation=None,
multi_register_certified=False,
classical_verified=False,
)
role_graph = graph_to_role_graph(graph)
sel: SelectorResult = select_structure(role_graph, families=families)
if sel.refused or sel.selected is None:
return SolveOutcome(
emitted=False,
answer=None,
refusal_reason=f"structure_map_refuse:{sel.reason}",
binding=None,
derivation=None,
multi_register_certified=False,
classical_verified=False,
source_graph_hash=role_graph.source_graph_hash,
)
mapped: StructureMapResult = sel.selected
binding = dict(mapped.binding)
# S2: prefer original unknown entity when present.
if mapped.structure_id == "S2" and graph is not None and graph.unknown.entity:
binding["query_entity"] = graph.unknown.entity
out = solve_binding(mapped.structure_id, binding)
if graph is not None:
return _original_graph_backstop(out, graph, role_graph)
return SolveOutcome(
emitted=out.emitted,
answer=out.answer,
refusal_reason=out.refusal_reason,
binding=out.binding,
derivation=out.derivation,
multi_register_certified=out.multi_register_certified,
classical_verified=out.classical_verified,
structure_id=out.structure_id,
source_graph_hash=role_graph.source_graph_hash,
)
def try_s1_structure_map_and_solve(
graph: MathProblemGraph | None = None,
*,
role_graph: RoleGraph | None = None,
) -> SolveOutcome:
"""Increment 1 API: S1-only vertical slice (selector restricted to S1)."""
return try_structure_map_and_solve(
graph, role_graph=role_graph, families=("S1",)
)

339
generate/structure_mapping/solve_s1.py
View file

@ -1,330 +1,23 @@
"""Solve an S1 binding through the existing multi-register certificate corridor.
"""Back-compat re-exports for Increment 1 import paths.
Reuses ``evals.multi_register_program`` (ADR-0249/0250 Hamiltonian path):
compile graph → execute with ``relax_to_ground`` → require ``certified``.
Emit only when the certificate chain verifies; else refuse.
Also cross-checks classical ``math_solver.solve`` + ``math_verifier.verify``
so derivation is independently replayable (right-for-right-reason).
Off-serving. Does not touch the live reader dispatch.
Implementation lives in ``generate.structure_mapping.solve`` (generalized
S1–S4 corridor). This module preserves the Increment 1 public names.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Mapping
from evals.multi_register_program import (
MultiRegisterError,
compile_multi_register_program,
execute_multi_register_program,
from generate.structure_mapping.solve import (
S1SolveOutcome,
SolveOutcome,
graph_from_s1_binding,
solve_s1_binding,
try_s1_structure_map_and_solve,
)
from generate.math_problem_graph import (
Comparison,
InitialPossession,
MathProblemGraph,
Operation,
Quantity,
Unknown,
)
from generate.math_solver import SolveError, solve
from generate.math_verifier import VerificationError, verify
from generate.structure_mapping.convert import graph_to_role_graph
from generate.structure_mapping.mapper import (
StructureMapRefuse,
StructureMapResult,
map_to_s1,
)
from generate.structure_mapping.role_predicate import RoleGraph
@dataclass(frozen=True, slots=True)
class S1SolveOutcome:
"""Emit or refuse for one S1-mapped problem."""
emitted: bool
answer: float | None
refusal_reason: str | None
binding: Mapping[str, object] | None
derivation: str | None
"""Human-readable derivation for right-reason audits."""
multi_register_certified: bool
classical_verified: bool
source_graph_hash: str | None = None
def graph_from_s1_binding(binding: Mapping[str, object]) -> MathProblemGraph:
"""Rebuild a pure S1 :class:`MathProblemGraph` from a role binding.
Shape: contain(a)=a_value; compare_multiplicative b = k×a; unknown total.
"""
a = str(binding["a"])
b = str(binding["b"])
k = float(binding["k"]) # type: ignore[arg-type]
a_value = float(binding["a_value"]) # type: ignore[arg-type]
unit = str(binding.get("unit") or "units")
return MathProblemGraph(
entities=(a, b),
initial_state=(
InitialPossession(entity=a, quantity=Quantity(value=a_value, unit=unit)),
),
operations=(
Operation(
actor=b,
kind="compare_multiplicative",
operand=Comparison(
reference_actor=a,
delta=None,
factor=k,
direction="times",
),
),
),
unknown=Unknown(entity=None, unit=unit),
)
def solve_s1_binding(binding: Mapping[str, object]) -> S1SolveOutcome:
"""Solve via multi-register certificate corridor; refuse if uncertified."""
required = ("a", "b", "k", "a_value")
missing = [k for k in required if k not in binding]
if missing:
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason=f"binding_incomplete:{','.join(missing)}",
binding=dict(binding),
derivation=None,
multi_register_certified=False,
classical_verified=False,
)
graph = graph_from_s1_binding(binding)
a = str(binding["a"])
b = str(binding["b"])
k = float(binding["k"]) # type: ignore[arg-type]
a_value = float(binding["a_value"]) # type: ignore[arg-type]
expected_formula = a_value * (1.0 + k)
derivation = (
f"S1: {b} = {k} × {a}; {a} = {a_value}; "
f"total = {a} + {b} = {a_value} + {k}*{a_value} = {expected_formula}"
)
# Classical solve + independent verify (replay certificate).
classical_verified = False
classical_answer: float | None = None
try:
trace = solve(graph)
verdict = verify(graph, trace)
classical_verified = bool(verdict.passed)
classical_answer = float(trace.answer_value)
except (SolveError, VerificationError, ValueError) as exc:
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason=f"classical_solve_or_verify_failed:{type(exc).__name__}:{exc}",
binding=dict(binding),
derivation=derivation,
multi_register_certified=False,
classical_verified=False,
)
if not classical_verified:
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason="classical_verifier_rejected",
binding=dict(binding),
derivation=derivation,
multi_register_certified=False,
classical_verified=False,
)
# Hamiltonian multi-register corridor (ADR-0250).
mr_certified = False
mr_answer: float | None = None
try:
program = compile_multi_register_program(graph)
outcome = execute_multi_register_program(program)
mr_certified = bool(outcome.certified)
mr_answer = float(outcome.answer)
except MultiRegisterError as exc:
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason=f"multi_register_error:{exc.reason}",
binding=dict(binding),
derivation=derivation,
multi_register_certified=False,
classical_verified=True,
)
except Exception as exc: # noqa: BLE001 — refuse, never crash-as-signal
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason=f"multi_register_unexpected:{type(exc).__name__}:{exc}",
binding=dict(binding),
derivation=derivation,
multi_register_certified=False,
classical_verified=True,
)
if not mr_certified:
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason="multi_register_chain_not_certified",
binding=dict(binding),
derivation=derivation,
multi_register_certified=False,
classical_verified=True,
)
# Agreement gate: classical and multi-register must agree (wrong=0).
if classical_answer is None or mr_answer is None:
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason="missing_answer_after_cert",
binding=dict(binding),
derivation=derivation,
multi_register_certified=mr_certified,
classical_verified=classical_verified,
)
if abs(classical_answer - mr_answer) > 1e-6 * max(1.0, abs(classical_answer)):
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason=(
f"corridor_disagreement:classical={classical_answer},mr={mr_answer}"
),
binding=dict(binding),
derivation=derivation,
multi_register_certified=mr_certified,
classical_verified=classical_verified,
)
# Emit the classically verified scalar once the multi-register chain is
# certified and both corridors agree within relative 1e-6. Geometric
# relaxation leaves ~1e-8 relative float noise on the MR decode; the
# classical solver answer is the exact integer arithmetic of the binding
# and is independently replay-verified. Refusing to emit the noisier MR
# float is not a shortcut around the certificate — ``mr_certified`` must
# still be True.
return S1SolveOutcome(
emitted=True,
answer=classical_answer,
refusal_reason=None,
binding=dict(binding),
derivation=derivation,
multi_register_certified=True,
classical_verified=True,
source_graph_hash=None,
)
def try_s1_structure_map_and_solve(
graph: MathProblemGraph | None = None,
*,
role_graph: RoleGraph | None = None,
) -> S1SolveOutcome:
"""Convert (if needed) → map_to_s1 → solve corridor. Full S1 vertical slice.
Accepts either a surface graph or a prebuilt role graph. Never takes a
gold structure label.
When the original ``MathProblemGraph`` is supplied, an additional
integrity gate runs after a successful binding solve: the original graph
must classically solve to the same answer as the pure-S1 rebuild. This
is a fail-closed backstop so a mapper bug that drops entities cannot
emit a certified wrong answer for a different problem.
"""
if role_graph is None:
if graph is None:
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason="no_graph_provided",
binding=None,
derivation=None,
multi_register_certified=False,
classical_verified=False,
)
role_graph = graph_to_role_graph(graph)
mapped = map_to_s1(role_graph)
if isinstance(mapped, StructureMapRefuse):
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason=f"structure_map_refuse:{mapped.reason}",
binding=None,
derivation=None,
multi_register_certified=False,
classical_verified=False,
source_graph_hash=role_graph.source_graph_hash,
)
assert isinstance(mapped, StructureMapResult)
out = solve_s1_binding(mapped.binding)
# Original-graph integrity backstop (when available).
if (
out.emitted
and out.answer is not None
and graph is not None
):
try:
original_trace = solve(graph)
original_verdict = verify(graph, original_trace)
except (SolveError, VerificationError, ValueError) as exc:
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason=(
f"original_graph_solve_failed:{type(exc).__name__}:{exc}"
),
binding=out.binding,
derivation=out.derivation,
multi_register_certified=out.multi_register_certified,
classical_verified=False,
source_graph_hash=role_graph.source_graph_hash,
)
if not original_verdict.passed:
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason="original_graph_verifier_rejected",
binding=out.binding,
derivation=out.derivation,
multi_register_certified=out.multi_register_certified,
classical_verified=False,
source_graph_hash=role_graph.source_graph_hash,
)
orig_ans = float(original_trace.answer_value)
if abs(orig_ans - float(out.answer)) > 1e-6 * max(1.0, abs(orig_ans)):
return S1SolveOutcome(
emitted=False,
answer=None,
refusal_reason=(
f"original_graph_disagreement:original={orig_ans},"
f"s1_rebuild={out.answer}"
),
binding=out.binding,
derivation=out.derivation,
multi_register_certified=out.multi_register_certified,
classical_verified=out.classical_verified,
source_graph_hash=role_graph.source_graph_hash,
)
return S1SolveOutcome(
emitted=out.emitted,
answer=out.answer,
refusal_reason=out.refusal_reason,
binding=out.binding,
derivation=out.derivation,
multi_register_certified=out.multi_register_certified,
classical_verified=out.classical_verified,
source_graph_hash=role_graph.source_graph_hash,
)
__all__ = [
"S1SolveOutcome",
"SolveOutcome",
"graph_from_s1_binding",
"solve_s1_binding",
"try_s1_structure_map_and_solve",
]

328
generate/structure_mapping/text_extract.py Normal file
View file

@ -0,0 +1,328 @@
"""Structure-mapping-owned pure-family text → MathProblemGraph extractors.
Off-serving only. Used when the live ``parse_and_solve`` reader refuses a
problem that is still a pure S1/S2/S3/S4 surface form. Does **not** modify
the serving reader or organ dispatch.
Deterministic regex extractors — no LLM, no gold labels. Refuse is the
default; only narrow pure-family patterns emit a graph.
Pure-family surface gates (fail-closed — wrong=0 absolute):
- Exactly one multiplicative compare phrase (``times as many/much``,
``twice|double|triple|thrice as``, ``times the number/size of``).
- Never match bare additive language (``N more``, ``N fewer``) as a factor —
that was the 0361 overmatch that emitted a certified wrong answer.
- Refuse multi-clause / multi-relation / fraction / percent surfaces.
- At most two numeric seeds and two named roles for inverted/day templates.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from generate.math_problem_graph import (
Comparison,
InitialPossession,
MathProblemGraph,
Operation,
Quantity,
Unknown,
)
# Word → factor for multiplicative language.
_WORD_FACTORS: dict[str, float] = {
"twice": 2.0,
"double": 2.0,
"two": 2.0,
"three": 3.0,
"triple": 3.0,
"thrice": 3.0,
"four": 4.0,
"five": 5.0,
"six": 6.0,
"seven": 7.0,
"eight": 8.0,
"nine": 9.0,
"ten": 10.0,
}
# Multiplicative compare surface (must be present for pure S1 extract).
# Intentionally does NOT match bare "N more" / "N fewer" (additive).
_MULT_COMPARE = re.compile(
r"(?:"
r"twice\s+as\s+(?:many|much)|"
r"twice\s+the\s+(?:total\s+)?(?:number|amount|size)|"
r"double\s+(?:what|the)|"
r"triple\s+(?:what|the)|"
r"thrice\s+as\s+(?:many|much)|"
r"(?:two|three|four|five|six|seven|eight|nine|ten|\d+)\s+times\s+"
r"(?:as\s+(?:many|much)|the\s+(?:total\s+)?(?:number|size|amount)|more\s+than)|"
r"times\s+(?:as\s+(?:many|much)|the\s+(?:total\s+)?(?:number|size|amount)\s+of)"
r")",
re.IGNORECASE,
)
# Additive / multi-step markers that disqualify pure-S1 extract.
_IMPURE_MARKERS = re.compile(
r"(?:"
r"\bfewer\b|"
r"\bless than\b|"
r"\bmore than\b|" # additive "more than" without being part of "times more than"
r"\b\d+\s+more\b|"
r"\b\d+\s+fewer\b|"
r"\b\d+\s+less\b|"
r"\b1/\d+\b|"
r"\bhalf as\b|"
r"\ba third\b|"
r"\bpercent\b|"
r"%|"
r"\band half\b|"
r"\bbut\b|"
r"\bthen\b|"
r"\bafter\b|"
r"\bbought\b|"
r"\bsold\b|"
r"\bgave\b|"
r"\bgives\b|"
r"\blost\b|"
r"\beach of\b|"
r"\bevery\b"
r")",
re.IGNORECASE,
)
# "times more than" is multiplicative; strip those before impure "more than" check.
_TIMES_MORE_THAN = re.compile(
r"(?:twice|thrice|\d+|two|three|four|five|six|seven|eight|nine|ten)\s+times\s+more\s+than",
re.IGNORECASE,
)
@dataclass(frozen=True, slots=True)
class TextExtractResult:
graph: MathProblemGraph | None
reason: str | None
pattern_id: str | None = None
def _factor_from_token(tok: str) -> float | None:
t = tok.strip().lower()
if t in _WORD_FACTORS:
return _WORD_FACTORS[t]
try:
v = float(t)
except ValueError:
return None
return v if v > 0 else None
def _purity_gate(text: str) -> str | None:
"""Return a refuse reason if text is not a pure single-relation S1 surface."""
# Count multiplicative compare phrases — pure S1 has exactly one.
mults = list(_MULT_COMPARE.finditer(text))
if len(mults) == 0:
return "no_multiplicative_compare"
if len(mults) > 1:
return "multi_multiplicative_compare"
# Impure markers after masking legitimate "times more than".
masked = _TIMES_MORE_THAN.sub("TIMES_MORE_THAN", text)
# Also mask the single mult compare span so "more" inside "times as many"
# is not confused — the mult phrase itself is clean.
for m in mults:
masked = masked[: m.start()] + ("X" * (m.end() - m.start())) + masked[m.end() :]
if _IMPURE_MARKERS.search(masked):
return "impure_multi_clause_markers"
# Too many independent numeric seeds → multi-step (pure S1: 1 seed + factor
# word, or 1 seed with numeric factor already in mult phrase).
nums = re.findall(r"\b\d+(?:\.\d+)?\b", text)
if len(nums) > 2:
return "too_many_numeric_seeds"
return None
def _s1_graph(
*,
a: str,
b: str,
k: float,
a_value: float,
unit: str,
) -> MathProblemGraph:
return MathProblemGraph(
entities=(a, b),
initial_state=(
InitialPossession(entity=a, quantity=Quantity(value=a_value, unit=unit)),
),
operations=(
Operation(
actor=b,
kind="compare_multiplicative",
operand=Comparison(
reference_actor=a,
delta=None,
factor=k,
direction="times",
),
),
),
unknown=Unknown(entity=None, unit=unit),
)
def extract_pure_s1(text: str) -> TextExtractResult:
"""Extract pure S1 graphs from narrow surface templates.
Patterns (all require a total/altogether/combined/two-days query and
pass the pure-family surface gate):
P2 — scaled entity seeded (inverted):
``<B> wrote|has <k> times as many <unit> as <A>. If <B> ... <n> ... altogether``
P2b — warehouse / first-second form with ``twice|N times as many ... as``
P2c — deaf/blind population form
P3 — day pair: first day = k × second day; second seeded; total two days
"""
if not isinstance(text, str) or not text.strip():
return TextExtractResult(None, "empty_text")
t = " ".join(text.split())
refuse = _purity_gate(t)
if refuse is not None:
return TextExtractResult(None, f"not_pure_s1:{refuse}")
# --- P3: people-counting / day1 = k × day2, day2 seeded (case 0148 family)
# Requires explicit "times" when kword is numeric; "twice/double/triple/thrice"
# are inherently multiplicative.
m = re.search(
r"(?:the\s+)?(?:number of\s+\w+\s+counted on\s+)?the\s+first\s+day\s+was\s+"
r"(?P<kword>twice|double|triple|thrice|"
r"(?:two|three|four|five|six|seven|eight|nine|ten|\d+)\s+times)"
r"\s+(?:the\s+)?(?:total\s+)?(?:number\s+)?(?:counted\s+)?on\s+the\s+second\s+day"
r".{0,80}?"
r"(?P<n>\d+(?:\.\d+)?)\s+\w+\s+(?:were\s+)?counted\s+on\s+the\s+second\s+day"
r".{0,40}?"
r"(?:how many|total).{0,40}?(?:two days|both days|in total)",
t,
flags=re.IGNORECASE | re.DOTALL,
)
if m:
k_raw = m.group("kword")
# Strip trailing "times" for numeric/word factors
k_tok = re.sub(r"\s+times$", "", k_raw.strip(), flags=re.IGNORECASE)
k = _factor_from_token(k_tok)
n = float(m.group("n"))
if k is not None:
graph = _s1_graph(
a="second day", b="first day", k=k, a_value=n, unit="people"
)
return TextExtractResult(graph, None, pattern_id="s1_p3_day_pair")
# --- P2: inverted seed — B = k×A, B has n, total A+B
# REQUIRE "times as many/much" or word-factor "twice as many" — never bare "N more".
m = re.search(
r"(?P<b>[A-Z][a-z]+)\s+"
r"(?:wrote|has|have|scored|taught|caught|produced)\s+"
r"(?P<kword>twice|double|triple|thrice|"
r"(?:two|three|four|five|six|seven|eight|nine|ten|\d+)\s+times)\s+"
r"(?:as many|as much)\s+"
r"(?P<unit>\w+)\s+as\s+"
r"(?P<a>[A-Z][a-z]+)"
r"\.\s+"
r"(?:If\s+)?(?P=b)\s+(?:wrote|has|have|scored|taught|caught|produced)\s+"
r"(?P<n>\d+(?:\.\d+)?)"
r".{0,80}?"
r"(?:altogether|in total|combined)",
t,
flags=re.IGNORECASE | re.DOTALL,
)
if m:
k_tok = re.sub(
r"\s+times$", "", m.group("kword").strip(), flags=re.IGNORECASE
)
k = _factor_from_token(k_tok)
n = float(m.group("n"))
a = m.group("a").strip()
b = m.group("b").strip()
unit = m.group("unit").lower()
# Entity names must be single proper nouns (reject "he does lions").
if (
k is not None
and a.lower() != b.lower()
and a[0].isupper()
and b[0].isupper()
and " " not in a
and " " not in b
):
graph = _s1_graph(a=a, b=b, k=k, a_value=n / k, unit=unit)
return TextExtractResult(graph, None, pattern_id="s1_p2_b_seeded")
# Warehouse form: first/second + "twice|N times as many UNIT as"
m = re.search(
r"(?:the\s+)?(?P<b>first\s+\w+)\s+has\s+"
r"(?P<kword>twice|double|triple|thrice|"
r"(?:two|three|four|five|six|seven|eight|nine|ten|\d+)\s+times)\s+"
r"as many\s+(?P<unit>\w+)\s+as\s+"
r"(?:the\s+)?(?P<a>second\s+\w+)"
r".{0,80}?"
r"(?:if\s+)?(?:the\s+)?(?P=b)\s+has\s+(?P<n>\d+(?:\.\d+)?)"
r".{0,80}?"
r"(?:combined|altogether|in total|how many)",
t,
flags=re.IGNORECASE | re.DOTALL,
)
if m:
k_tok = re.sub(
r"\s+times$", "", m.group("kword").strip(), flags=re.IGNORECASE
)
k = _factor_from_token(k_tok)
n = float(m.group("n"))
a = re.sub(r"\s+", " ", m.group("a")).strip()
b = re.sub(r"\s+", " ", m.group("b")).strip()
unit = m.group("unit").lower()
if k is not None and a.lower() != b.lower():
graph = _s1_graph(a=a, b=b, k=k, a_value=n / k, unit=unit)
return TextExtractResult(graph, None, pattern_id="s1_p2_b_seeded_has")
# Deaf/blind population form (fixed surface; still purity-gated).
m = re.search(
r"(?P<b>deaf\s+student\s+population)\s+"
r"(?P<kword>twice|double|triple|thrice|"
r"(?:two|three|four|five|six|seven|eight|nine|ten|\d+)\s+times)\s+"
r"(?:the size of|as many as)\s+"
r"(?P<a>blind\s+student\s+population)"
r".{0,80}?"
r"(?:number of\s+deaf\s+students\s+is|deaf\s+students\s+is)\s+(?P<n>\d+(?:\.\d+)?)"
r".{0,60}?"
r"(?:altogether|in total|how many students)",
t,
flags=re.IGNORECASE | re.DOTALL,
)
if m:
k_tok = re.sub(
r"\s+times$", "", m.group("kword").strip(), flags=re.IGNORECASE
)
k = _factor_from_token(k_tok)
n = float(m.group("n"))
if k is not None:
graph = _s1_graph(
a="blind students",
b="deaf students",
k=k,
a_value=n / k,
unit="students",
)
return TextExtractResult(graph, None, pattern_id="s1_p2_deaf_blind")
return TextExtractResult(None, "no_pure_s1_pattern")
def extract_role_graph_from_text(text: str) -> TextExtractResult:
"""Best-effort SM-owned extract. Currently pure-S1 patterns only."""
return extract_pure_s1(text)

500
scripts/measure_trackb_inc2.py Normal file
View file

@ -0,0 +1,500 @@
"""Blind measurement for Track B Increment 2 (S1–S4 + selector + generalization).
Usage::
uv run python scripts/measure_trackb_inc2.py --mode all
uv run python scripts/measure_trackb_inc2.py --mode ratio
uv run python scripts/measure_trackb_inc2.py --mode coverage-gain
uv run python scripts/measure_trackb_inc2.py --mode selector
uv run python scripts/measure_trackb_inc2.py --mode parser-frontier
uv run python scripts/measure_trackb_inc2.py --mode right-reason
Labels load only in scoring branches. Mapper/selector never see gold labels.
"""
from __future__ import annotations
import argparse
import json
import sys
from collections import defaultdict
from pathlib import Path
from generate.math_candidate_graph import parse_and_solve
from generate.math_problem_graph import (
Comparison,
InitialPossession,
MathProblemGraph,
Operation,
Quantity,
Rate,
Unknown,
)
from generate.structure_mapping.convert import graph_to_role_graph
from generate.structure_mapping.mapper import (
StructureMapRefuse,
StructureMapResult,
map_to_s1,
map_to_s2,
)
from generate.structure_mapping.pipeline import run_structure_mapping_pipeline
from generate.structure_mapping.selector import select_structure
from generate.structure_mapping.solve import try_structure_map_and_solve
HOLDOUT_CASES = Path("evals/gsm8k_math/holdout_dev/v1/cases.jsonl")
PUBLIC_CASES = Path("evals/gsm8k_math/public/cases.jsonl")
# S1 organ cohort (already carried on serving reader).
S1_ORGAN_IDS = frozenset(
{
"gsm8k-holdout-dev-v1-0101",
"gsm8k-holdout-dev-v1-0108",
"gsm8k-holdout-dev-v1-0268",
"gsm8k-holdout-dev-v1-0411",
"gsm8k-holdout-dev-v1-0453",
}
)
# Pure-S1 holdout cases the organ misses; SM-owned extract targets.
S1_COVERAGE_CANDIDATES = frozenset(
{
"gsm8k-holdout-dev-v1-0148",
"gsm8k-holdout-dev-v1-0228",
"gsm8k-holdout-dev-v1-0234",
"gsm8k-holdout-dev-v1-0441",
}
)
def _load_cases(path: Path = HOLDOUT_CASES) -> dict[str, dict]:
out: dict[str, dict] = {}
for line in path.read_text(encoding="utf-8").splitlines():
row = json.loads(line)
out[row["id"]] = row
return out
def mode_ratio() -> None:
"""Generalization-ratio table on real holdout case IDs."""
print("=== generalization ratio (holdout_dev/v1, command-backed) ===")
cases = _load_cases()
by_canonical: dict[str, list[dict]] = defaultdict(list)
target_ids = S1_ORGAN_IDS | S1_COVERAGE_CANDIDATES
for cid in sorted(target_ids):
row = cases[cid]
gold = float(row["expected_answer"])
out, trace, _graph = run_structure_mapping_pipeline(row["problem"])
right = (
out.emitted
and out.answer is not None
and abs(float(out.answer) - gold) <= 1e-6 * max(1.0, abs(gold))
and out.derivation is not None
and out.multi_register_certified
and out.classical_verified
)
rec = {
"id": cid,
"structure_id": out.structure_id,
"emitted": out.emitted,
"answer": out.answer,
"gold": gold,
"right_reason": right,
"derivation": out.derivation,
"graph_source": trace.graph_source,
"extract_pattern": trace.extract_pattern,
"refusal": out.refusal_reason,
"organ_parsed": trace.organ_parsed,
}
print(
f"CASE {cid} sid={out.structure_id} emit={out.answer} gold={gold} "
f"right={right} src={trace.graph_source} pat={trace.extract_pattern} "
f"refuse={out.refusal_reason}"
)
if out.derivation:
print(f" derivation={out.derivation!r}")
if right and out.structure_id:
by_canonical[out.structure_id].append(rec)
print("--- ratio table ---")
n_templates = 0
for sid in ("S1", "S2", "S3", "S4"):
carried = by_canonical.get(sid, [])
# One canonical template written per family in the library.
templates = 1 if sid in ("S1", "S2", "S3", "S4") else 0
n_templates += templates
ids = [c["id"] for c in carried]
ratio = (len(carried) / templates) if templates and carried else 0.0
print(
f"canonical={sid} templates=1 cases_carried={len(carried)} "
f"ratio={ratio:.3f} ids={ids}"
)
# Overall for families that carried ≥1
active = {k: v for k, v in by_canonical.items() if v}
if active:
total_cases = sum(len(v) for v in active.values())
total_tmpl = len(active)
print(
f"SUMMARY active_canonicals={total_tmpl} total_cases={total_cases} "
f"aggregate_ratio={total_cases / total_tmpl:.3f}"
)
else:
print("SUMMARY no_cases_carried ratio=0")
def mode_coverage_gain() -> None:
"""Cases newly solved vs organ (holdout)."""
print("=== coverage gain vs organ (holdout_dev/v1) ===")
cases = _load_cases()
organ_ok = 0
tb_ok = 0
newly = []
both = []
for cid, row in cases.items():
gold = float(row["expected_answer"])
organ = parse_and_solve(row["problem"])
o_ok = (
organ.answer is not None
and abs(float(organ.answer) - gold) <= 1e-6 * max(1.0, abs(gold))
)
out, trace, _ = run_structure_mapping_pipeline(row["problem"])
t_ok = (
out.emitted
and out.answer is not None
and abs(float(out.answer) - gold) <= 1e-6 * max(1.0, abs(gold))
and out.multi_register_certified
)
if o_ok:
organ_ok += 1
if t_ok:
tb_ok += 1
if t_ok and not o_ok:
newly.append((cid, out.answer, gold, out.structure_id, trace.extract_pattern))
if t_ok and o_ok:
both.append(cid)
print(f"organ_correct={organ_ok} trackb_correct={tb_ok} newly_solved={len(newly)}")
for cid, ans, gold, sid, pat in newly:
print(f" NEW {cid} ans={ans} gold={gold} sid={sid} extract={pat}")
print(f"both_correct_count={len(both)}")
print(
"NOTE: organ_retirement not performed (off-serving; coverage-gain path). "
"Serving reader unchanged."
)
def mode_selector() -> None:
"""Selector routing table: real + targeted synthetic."""
print("=== selector routing (blind; synthetic + real graphs) ===")
def show(label: str, graph: MathProblemGraph, expect_sid: str | None) -> None:
rg = graph_to_role_graph(graph)
sel = select_structure(rg)
got = sel.selected.structure_id if sel.selected else None
ok = got == expect_sid
print(
f"CASE {label} expect={expect_sid} got={got} refused={sel.refused} "
f"reason={sel.reason} ok={ok} candidates="
f"{[c.structure_id for c in sel.candidates]}"
)
# Real S1 organ graph
s1_text = (
"In a class, there were 13 female students. There were three times as many "
"male students in this class. How many students were in the class?"
)
g = parse_and_solve(s1_text).selected_graph
assert g is not None
show("real_s1_0411", g, "S1")
# Real S2 transfer from public corpus
t2 = "Eve has 15 coins. David has 47 coins. Eve hands 7 coins to David. How many coins does David have?"
g2 = parse_and_solve(t2).selected_graph
assert g2 is not None
show("real_s2_public_transfer", g2, "S2")
# Lexically S1-looking but structurally S2: has "times" in entity names? Better:
# transfer problem with multiplicative words in fluff is rare; instead use a
# pure S2 vs pure S1 structural pair and a deliberately ambiguous dual graph.
# Surface-similarity trap: text mentions "twice" but graph is transfer-only
# (built synthetic role graph with transfer only).
g_s2_trap = MathProblemGraph(
entities=("Alice", "Bob"),
initial_state=(
InitialPossession(entity="Alice", quantity=Quantity(value=10, unit="apples")),
InitialPossession(entity="Bob", quantity=Quantity(value=5, unit="apples")),
),
operations=(
Operation(
actor="Alice",
kind="transfer",
operand=Quantity(value=3, unit="apples"),
target="Bob",
),
),
unknown=Unknown(entity="Bob", unit="apples"),
)
show("struct_s2_not_s1", g_s2_trap, "S2")
# S1 pure
g_s1 = MathProblemGraph(
entities=("Ann", "Bea"),
initial_state=(
InitialPossession(entity="Ann", quantity=Quantity(value=7, unit="pts")),
),
operations=(
Operation(
actor="Bea",
kind="compare_multiplicative",
operand=Comparison(
reference_actor="Ann",
delta=None,
factor=3.0,
direction="times",
),
),
),
unknown=Unknown(entity=None, unit="pts"),
)
show("struct_s1", g_s1, "S1")
# Ambiguous: empty role graph → refuse
g_empty = MathProblemGraph(
entities=("X",),
initial_state=(
InitialPossession(entity="X", quantity=Quantity(value=1, unit="u")),
),
operations=(),
unknown=Unknown(entity="X", unit="u"),
)
show("empty_ops_refuse", g_empty, None)
# Rate graph → S3 map may succeed but note MR frontier later
g_rate = MathProblemGraph(
entities=("worker",),
initial_state=(
InitialPossession(entity="worker", quantity=Quantity(value=5, unit="hours")),
),
operations=(
Operation(
actor="worker",
kind="apply_rate",
operand=Rate(value=10.0, numerator_unit="pages", denominator_unit="hours"),
),
),
unknown=Unknown(entity="worker", unit="pages"),
)
show("struct_s3_rate", g_rate, "S3")
def mode_parser_frontier() -> None:
"""Parse→role coverage per family on holdout + note 0148."""
print("=== parser-frontier report (holdout_dev/v1) ===")
cases = _load_cases()
organ_graphs = 0
by_kinds: dict[str, int] = defaultdict(int)
for cid, row in cases.items():
organ = parse_and_solve(row["problem"])
if organ.selected_graph is None:
continue
organ_graphs += 1
rg = graph_to_role_graph(organ.selected_graph)
kinds = ",".join(sorted(rg.kinds()))
by_kinds[kinds] += 1
mapped = []
for fam, fn in (
("S1", map_to_s1),
("S2", map_to_s2),
):
m = fn(rg)
if isinstance(m, StructureMapResult):
mapped.append(fam)
if mapped:
print(f" organ_graph {cid} kinds={kinds} maps={mapped}")
print(f"organ_selected_graph_count={organ_graphs}/500")
print(f"role_kind_multiset_counts={dict(by_kinds)}")
# 0148 status
row = cases["gsm8k-holdout-dev-v1-0148"]
organ = parse_and_solve(row["problem"])
out, trace, _ = run_structure_mapping_pipeline(row["problem"])
print("--- case 0148 ---")
print(f"organ_parsed={organ.selected_graph is not None} organ_answer={organ.answer}")
print(
f"sm_pipeline emit={out.emitted} answer={out.answer} gold={row['expected_answer']} "
f"src={trace.graph_source} pat={trace.extract_pattern} sid={out.structure_id}"
)
print(
"NOTE: holdout non-S1 organ graphs remain near-zero; S2/S3/S4 holdout "
"coverage is parser-frontier limited. S2 is demonstrated on public "
"transfer graphs (real, not clones) in selector/right-reason modes."
)
def mode_right_reason() -> None:
"""3-part right-reason on carried S1 holdout + surface variant."""
print("=== right-for-right-reason (S1 holdout + coverage) ===")
cases = _load_cases()
correct = wrong = refused = 0
for cid in sorted(S1_ORGAN_IDS | S1_COVERAGE_CANDIDATES):
row = cases[cid]
gold = float(row["expected_answer"])
out, trace, _ = run_structure_mapping_pipeline(row["problem"])
if not out.emitted or out.answer is None:
refused += 1
print(f"CASE {cid} REFUSED {out.refusal_reason} src={trace.graph_source}")
continue
match = abs(float(out.answer) - gold) <= 1e-6 * max(1.0, abs(gold))
if match and out.derivation and out.multi_register_certified:
correct += 1
flag = False
else:
wrong += 1
flag = True
print(
f"CASE {cid} emit={out.answer} gold={gold} match={match} wrong_flag={flag} "
f"src={trace.graph_source} mr={out.multi_register_certified} "
f"derivation={out.derivation!r}"
)
print(f"SUMMARY right-reason: correct={correct} wrong={wrong} refused={refused}")
# Surface-variant on a coverage case (0148-style rephrase)
print("=== surface-variant (re-parse / re-extract) ===")
base = cases["gsm8k-holdout-dev-v1-0441"]["problem"]
variant = (
"Ava wrote five times as many essays as Ben. If Ava wrote 40 essays, "
"how many essays did they write altogether?"
)
out_b, tr_b, _ = run_structure_mapping_pipeline(base)
out_v, tr_v, _ = run_structure_mapping_pipeline(variant)
print(f"BASE emit={out_b.answer} sid={out_b.structure_id} src={tr_b.graph_source}")
print(
f"VARIANT emit={out_v.answer} expected=48.0 sid={out_v.structure_id} "
f"src={tr_v.graph_source} binding={out_v.binding}"
)
ok = (
out_v.emitted
and out_v.answer is not None
and abs(float(out_v.answer) - 48.0) < 1e-6
and out_v.structure_id == "S1"
)
print(f"SUMMARY surface-variant: ok={ok}")
# S2 public real transfer right-reason
print("=== S2 public transfer right-reason (real corpus, not holdout) ===")
pub = _load_cases(PUBLIC_CASES) if PUBLIC_CASES.exists() else {}
# scan a few known transfers
s2_n = s2_ok = 0
for cid, row in list(pub.items())[:80]:
organ = parse_and_solve(row["problem"])
if organ.selected_graph is None:
continue
rg = graph_to_role_graph(organ.selected_graph)
if "transfer" not in rg.kinds():
continue
s2_n += 1
out = try_structure_map_and_solve(organ.selected_graph, families=("S1", "S2", "S3", "S4"))
gold = float(row["expected_answer"])
if (
out.emitted
and out.answer is not None
and abs(float(out.answer) - gold) <= 1e-6 * max(1.0, abs(gold))
and out.structure_id == "S2"
):
s2_ok += 1
if s2_ok <= 3:
print(f" S2 {cid} ans={out.answer} gold={gold} der={out.derivation!r}")
if s2_n >= 20:
break
print(f"SUMMARY s2_public_sample: ok={s2_ok} scanned_transferish={s2_n}")
def mode_full_holdout_wrong0() -> None:
"""Absolute wrong=0 gate over **all** holdout_dev/v1 cases (not curated subset).
Any pipeline emit whose answer disagrees with gold is a hard failure of the
wrong=0 bar. Overmatching SM extract that certifies garbage is the hazard
this mode is designed to catch (regression: 0361).
"""
print("=== full holdout_dev/v1 wrong=0 gate (all 500 cases) ===")
cases = _load_cases()
emit_ok = 0
wrong: list[tuple[str, float | None, float, str | None, str | None]] = []
refused = 0
for cid, row in cases.items():
gold = float(row["expected_answer"])
out, trace, _ = run_structure_mapping_pipeline(row["problem"])
if not out.emitted or out.answer is None:
refused += 1
continue
if abs(float(out.answer) - gold) <= 1e-6 * max(1.0, abs(gold)):
emit_ok += 1
else:
wrong.append(
(
cid,
out.answer,
gold,
trace.graph_source,
trace.extract_pattern,
)
)
print(
f"SUMMARY full-holdout: emit_ok={emit_ok} wrong={len(wrong)} "
f"refused={refused} n={len(cases)}"
)
for cid, ans, gold, src, pat in wrong:
print(
f" WRONG {cid} emit={ans} gold={gold} src={src} extract={pat}"
)
if wrong:
print("FAIL: wrong≠0 on holdout_dev/v1 — pure-family extract overmatch")
else:
print("PASS: wrong=0 absolute on full holdout_dev/v1")
def mode_all() -> None:
mode_ratio()
print()
mode_coverage_gain()
print()
mode_selector()
print()
mode_parser_frontier()
print()
mode_right_reason()
print()
mode_full_holdout_wrong0()
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser()
p.add_argument(
"--mode",
choices=(
"ratio",
"coverage-gain",
"selector",
"parser-frontier",
"right-reason",
"full-holdout-wrong0",
"all",
),
default="all",
)
args = p.parse_args(argv)
{
"ratio": mode_ratio,
"coverage-gain": mode_coverage_gain,
"selector": mode_selector,
"parser-frontier": mode_parser_frontier,
"right-reason": mode_right_reason,
"full-holdout-wrong0": mode_full_holdout_wrong0,
"all": mode_all,
}[args.mode]()
return 0
if __name__ == "__main__":
sys.exit(main())

327
tests/test_trackb_inc2_structure_mapping.py Normal file
View file

@ -0,0 +1,327 @@
"""Track B Increment 2 — S2–S4 canonicals, selector, coverage extract, gates."""
from __future__ import annotations
import ast
import inspect
from pathlib import Path
import pytest
from generate.math_candidate_graph import parse_and_solve
from generate.math_problem_graph import (
Comparison,
InitialPossession,
MathProblemGraph,
Operation,
Quantity,
Rate,
Unknown,
)
from generate.structure_mapping.canonicals import (
CANONICAL_LIBRARY,
S1_CANONICAL,
S2_CANONICAL,
S3_CANONICAL,
S4_CANONICAL,
)
from generate.structure_mapping.convert import graph_to_role_graph
from generate.structure_mapping.mapper import (
StructureMapRefuse,
StructureMapResult,
map_to_s1,
map_to_s2,
map_to_s3,
map_to_s4,
)
from generate.structure_mapping.pipeline import run_structure_mapping_pipeline
from generate.structure_mapping.selector import select_structure
from generate.structure_mapping.solve import solve_binding, try_structure_map_and_solve
from generate.structure_mapping.text_extract import extract_pure_s1
# Real holdout pure-S1 coverage targets (organ misses; SM extract carries).
COVERAGE_CASES = (
(
"gsm8k-holdout-dev-v1-0148",
"At a people counting station, the number of people counted on the first day "
"was twice the total number counted on the second day. If 500 people were "
"counted on the second day, how many people were counted on the two days?",
1500.0,
),
(
"gsm8k-holdout-dev-v1-0228",
"There are two warehouses. The first warehouse has twice as many boxes as "
"the second warehouse. If the first warehouse has 400 boxes, how many boxes "
"are there in both warehouses combined?",
600.0,
),
(
"gsm8k-holdout-dev-v1-0234",
"A special school for deaf and blind students has a deaf student population "
"three times the size of blind student population. If the number of deaf "
"students is 180, how many students are there altogether?",
240.0,
),
(
"gsm8k-holdout-dev-v1-0441",
"Zig wrote four times as many books as Flo. If Zig wrote 60 books, how many "
"books did they write altogether?",
75.0,
),
)
def test_canonical_library_has_s1_through_s4():
ids = {c.structure_id for c in CANONICAL_LIBRARY}
assert ids == {"S1", "S2", "S3", "S4"}
for c in (S1_CANONICAL, S2_CANONICAL, S3_CANONICAL, S4_CANONICAL):
assert all(t.kind == "var" for p in c.predicates for t in p.args)
def test_s2_transfer_map_and_solve_public_shape():
text = (
"Eve has 15 coins. David has 47 coins. Eve hands 7 coins to David. "
"How many coins does David have?"
)
organ = parse_and_solve(text)
assert organ.selected_graph is not None
rg = graph_to_role_graph(organ.selected_graph)
mapped = map_to_s2(rg)
assert isinstance(mapped, StructureMapResult)
assert mapped.structure_id == "S2"
out = try_structure_map_and_solve(organ.selected_graph)
assert out.emitted
assert out.answer == pytest.approx(54.0)
assert out.structure_id == "S2"
assert out.multi_register_certified
assert "transfer" in (out.derivation or "")
def test_s2_refuses_when_compare_present():
g = MathProblemGraph(
entities=("A", "B"),
initial_state=(
InitialPossession(entity="A", quantity=Quantity(value=1, unit="u")),
InitialPossession(entity="B", quantity=Quantity(value=1, unit="u")),
),
operations=(
Operation(
actor="A",
kind="transfer",
operand=Quantity(value=1, unit="u"),
target="B",
),
Operation(
actor="B",
kind="compare_multiplicative",
operand=Comparison(
reference_actor="A", delta=None, factor=2.0, direction="times"
),
),
),
unknown=Unknown(entity="B", unit="u"),
)
mapped = map_to_s2(graph_to_role_graph(g))
assert isinstance(mapped, StructureMapRefuse)
def test_s1_inverted_seed_binding():
"""b seeded, a = b/k — pure S1."""
g = MathProblemGraph(
entities=("Flo", "Zig"),
initial_state=(
InitialPossession(entity="Zig", quantity=Quantity(value=60, unit="books")),
),
operations=(
Operation(
actor="Zig",
kind="compare_multiplicative",
operand=Comparison(
reference_actor="Flo", delta=None, factor=4.0, direction="times"
),
),
),
unknown=Unknown(entity=None, unit="books"),
)
# contain only on b; convert will emit contain(Zig) + compare + total
rg = graph_to_role_graph(g)
mapped = map_to_s1(rg)
assert isinstance(mapped, StructureMapResult)
assert mapped.binding["a_value"] == pytest.approx(15.0)
assert mapped.binding["seed_mode"] == "b_seeded"
out = solve_binding("S1", mapped.binding)
assert out.emitted
assert out.answer == pytest.approx(75.0)
def test_selector_picks_s1_over_refuse_others():
g = MathProblemGraph(
entities=("Ann", "Bea"),
initial_state=(
InitialPossession(entity="Ann", quantity=Quantity(value=7, unit="pts")),
),
operations=(
Operation(
actor="Bea",
kind="compare_multiplicative",
operand=Comparison(
reference_actor="Ann", delta=None, factor=3.0, direction="times"
),
),
),
unknown=Unknown(entity=None, unit="pts"),
)
sel = select_structure(graph_to_role_graph(g))
assert not sel.refused
assert sel.selected is not None
assert sel.selected.structure_id == "S1"
def test_selector_picks_s2_not_s1_on_transfer():
g = MathProblemGraph(
entities=("Alice", "Bob"),
initial_state=(
InitialPossession(entity="Alice", quantity=Quantity(value=10, unit="apples")),
InitialPossession(entity="Bob", quantity=Quantity(value=5, unit="apples")),
),
operations=(
Operation(
actor="Alice",
kind="transfer",
operand=Quantity(value=3, unit="apples"),
target="Bob",
),
),
unknown=Unknown(entity="Bob", unit="apples"),
)
sel = select_structure(graph_to_role_graph(g))
assert not sel.refused
assert sel.selected is not None
assert sel.selected.structure_id == "S2"
# S1 must have refused this graph
assert any(f == "S1" for f, _ in sel.refused_families)
def test_selector_refuses_empty():
g = MathProblemGraph(
entities=("X",),
initial_state=(
InitialPossession(entity="X", quantity=Quantity(value=1, unit="u")),
),
operations=(),
unknown=Unknown(entity="X", unit="u"),
)
sel = select_structure(graph_to_role_graph(g))
assert sel.refused
assert sel.selected is None
def test_s3_maps_but_mr_frontier_refuses_emit():
"""S3 pure map succeeds; multi-register out-of-scope → refuse (gate holds)."""
g = MathProblemGraph(
entities=("worker",),
initial_state=(
InitialPossession(entity="worker", quantity=Quantity(value=5, unit="hours")),
),
operations=(
Operation(
actor="worker",
kind="apply_rate",
operand=Rate(
value=10.0, numerator_unit="pages", denominator_unit="hours"
),
),
),
unknown=Unknown(entity="worker", unit="pages"),
)
mapped = map_to_s3(graph_to_role_graph(g))
assert isinstance(mapped, StructureMapResult)
out = try_structure_map_and_solve(g)
# Either S3 selected and MR refuses, or emit only if corridor supports it.
if out.emitted:
assert out.multi_register_certified
assert out.answer == pytest.approx(50.0)
else:
assert out.refusal_reason is not None
assert "multi_register" in out.refusal_reason or "structure_map" in (
out.refusal_reason or ""
)
@pytest.mark.parametrize("case_id,text,gold", COVERAGE_CASES)
def test_coverage_gain_holdout_pure_s1(case_id: str, text: str, gold: float):
organ = parse_and_solve(text)
assert organ.selected_graph is None or organ.answer is None
out, trace, _ = run_structure_mapping_pipeline(text)
assert trace.extract_used
assert out.emitted
assert out.answer == pytest.approx(gold)
assert out.structure_id == "S1"
assert out.multi_register_certified
assert out.classical_verified
assert out.derivation is not None
def test_surface_variant_inverted_s1():
variant = (
"Ava wrote five times as many essays as Ben. If Ava wrote 40 essays, "
"how many essays did they write altogether?"
)
out, trace, _ = run_structure_mapping_pipeline(variant)
assert out.emitted
assert out.answer == pytest.approx(48.0)
assert out.structure_id == "S1"
assert trace.graph_source == "sm_extract"
def test_extract_refuses_multi_entity_zoo_0361():
"""Regression: 0361 overmatch must refuse, never emit wrong certified answer.
Prior loose P2 regex matched 'John has 8 more pandas than he does lions'
as pure S1 and the corridor emitted 1.125 vs gold 114.0 (wrong≠0).
"""
text = (
"John wants to start a zoo. He has 15 snakes. He has twice as many "
"monkeys as he does snakes. He has 5 fewer lions than he does monkeys. "
"John has 8 more pandas than he does lions. John has 1/3 as many dogs as "
"he does pandas. How many total animals does John have?"
)
from generate.structure_mapping.text_extract import extract_pure_s1
ext = extract_pure_s1(text)
assert ext.graph is None, f"must refuse extract, got {ext}"
assert ext.reason is not None
assert "not_pure_s1" in ext.reason or "no_pure" in ext.reason
out, trace, _ = run_structure_mapping_pipeline(text)
assert not out.emitted
assert out.answer is None
assert not trace.extract_used
def test_extract_refuses_bare_n_more_as_factor():
"""Bare additive 'N more' must never be parsed as multiplicative factor."""
from generate.structure_mapping.text_extract import extract_pure_s1
text = (
"John has 8 more pandas than he does lions. If John has 8 pandas, "
"how many pandas and lions do they have altogether?"
)
ext = extract_pure_s1(text)
assert ext.graph is None
def test_mapper_modules_do_not_import_scoring_labels():
root = Path("generate/structure_mapping")
for path in root.glob("*.py"):
tree = ast.parse(path.read_text(encoding="utf-8"))
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and node.module:
assert "structure_mapping.scoring" not in node.module
assert "holdout_dev_v1_labels" not in (node.module or "")
def test_select_structure_signature_blind():
sig = inspect.signature(select_structure)
assert "label" not in sig.parameters
assert "gold" not in sig.parameters