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

feat(ADR-0163.D.2): parsed_anchors → MathProblemGraph state — discrete_count_statement injection v1 (#315)

Browse source 
First PR plumbing recognizer parsed_anchors into the candidate-graph as
typed CandidateInitial primitives. Scope limited to discrete_count_statement;
other five round-2 categories route to the round-2 skip-only fallback until
follow-up D.2.x PRs.

Five-layer wrong=0 safety net:
1. Matcher narrowness — _try_extract_discrete_count_anchor refuses on any
   ambiguity (multi-subject, pronoun subject, non-possession verb,
   multi-count, clause-split, unobserved counted_noun, unobserved
   count_kind).
2. Extraction correctness — refusal-preferring; populated parsed_anchors
   only when ALL narrowness rules hold.
3. Injection correctness — _initial_admissible gates every constructed
   CandidateInitial; failure to ground returns () (under-admit).
4. Replay gate — propose-time admissibility_replay_gate auto-rejects any
   matcher change that would lift GSM8K wrong count.
5. Multi-branch decision rule — injected candidate disagreeing with
   another branch triggers refuse path.

Re-baseline (GSM8K train_sample v1):
- Old (#309 alone): correct=3 refused=47 wrong=0
- New (#309 + D.2 v1): correct=3 refused=47 wrong=0
- Empirical lift in v1 = 0 cases; framework operational. No GSM8K
  train_sample case has a discrete_count statement that simultaneously
  meets all narrowness rules AND is missed by the existing parser.
  Bottleneck moves to other recognizer categories (D.2.2+).

Validation:
- tests/test_adr_0163_d2_discrete_count_injection.py: 34 passed
- tests/test_recognizer_match.py + test_candidate_graph_recognizer_wiring
  + test_admissibility_replay_gate: 27 passed
- adr_0131_* (G1..G5 + S1 wrong=0 invariant): 222 passed / 2 pre-existing
  report-comparison failures / 3 skipped — byte-identical to pre-D.2
- Solver code: unchanged

Operator caveat: round-1's ratified discrete_count_statement spec is
unchanged. Matcher behavior on the spec's canonical_pattern has been
extended from detection-only to populated parsed_anchors. Re-ratification
is not required; if policy requires it on matcher-behavior changes, the
registry digest provides byte-stable provenance.
This commit is contained in:
Shay 2026-05-26 18:32:05 -07:00 committed by GitHub
parent 573fed073b
commit da70919f94
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 999 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
docs/decisions/ADR-0163-gsm8k-path-to-mastery.md
View file

@ -407,6 +407,69 @@ by running `evals.gsm8k_math.train_sample.v1.runner` against
See [SESSION-2026-05-26-corridor-closure.md](../sessions/SESSION-2026-05-26-corridor-closure.md)
for the full session ledger.
## Phase D.2 amendment — discrete_count_statement injection v1
Phase D.2 v1 plumbs `parsed_anchors` from one round-2 recognizer
(`discrete_count_statement`) into the candidate-graph as
`CandidateInitial`. The wiring is the first PR where a recognizer's
matcher output becomes solver input; wrong=0 moves from "skip-only by
construction" to **five layered safety nets** that all must hold:
1. **Matcher narrowness** — `_try_extract_discrete_count_anchor` refuses
on ambiguity: requires a single proper-noun subject, a closed
possession-verb whitelist (`has`/`have`/`had`), exactly one numeric
token, `count_kind ∈ observed_count_kinds`, `counted_noun ∈
observed_counted_nouns`, no clause-split connectives.
2. **Extraction correctness** — the recognizer's match returns
`parsed_anchors=()` (detection-only fallback) when the narrowness
rules fail; the per-category injector returns `()` on any
construction failure.
3. **Injection correctness** — the built `CandidateInitial` is gated by
`_initial_admissible` upstream of the Cartesian product; failures
under-admit (return `()`) rather than over-admit.
4. **Replay gate** — propose-time `run_admissibility_replay_gate`
auto-rejects extraction changes that lift GSM8K wrong count.
5. **Multi-branch decision rule** — when an injected candidate
disagrees with another branch's answer, the candidate-graph
refuses.
**Re-baseline (GSM8K train_sample v1, post-D.2 v1):**
`correct=3, refused=47, wrong=0` — **identical to the pre-D.2 baseline**.
The framework lands and is operational, but no GSM8K train_sample case
has a discrete_count statement that simultaneously (a) the existing
parser misses, (b) carries a counted_noun in the spec's observed lemma
set, (c) carries exactly one numeric token, and (d) carries no
clause-split connectives. Empirical lift in v1 = 0 cases; the bottleneck
is **other recognizer categories** (rate_with_currency, temporal_aggregation,
multiplicative_aggregation, currency_amount) whose injectors return `()`
(skip-only fallback) until follow-up PRs D.2.2..D.2.5 plumb them.
**Operator caveat — matcher behavior, not canonical_pattern.**
Round-1's ratified `discrete_count_statement` spec is unchanged. The
matcher's behavior on the spec's `canonical_pattern` has been extended
from detection-only to populated `parsed_anchors`. Re-ratification is
not required for this extension; if policy requires re-ratification
when matcher behavior changes, the registry digest provides byte-stable
provenance.
**G1..G5 + S1 wrong=0 invariant:** 222 passed / 2 pre-existing
report-comparison failures / 3 skipped — byte-identical to pre-D.2.
**Solver code: unchanged.** The injector returns the same
`CandidateInitial` type the existing parser produces; the solver runs
unchanged.
**Follow-up PRs (D.2.x):**
- D.2.2 — `rate_with_currency` parsed_anchors → solver state
- D.2.3 — `temporal_aggregation` parsed_anchors → solver state
- D.2.4 — `multiplicative_aggregation` parsed_anchors → solver state
- D.2.5 — `currency_amount` parsed_anchors → solver state
Each ships in its own PR after the operator reviews D.2 v1's framework
and empirical lift; the dispatch table in
`generate/recognizer_anchor_inject.py` is the single registration site.
---
## Acceptance criteria

35
generate/math_candidate_graph.py
View file

@ -504,8 +504,39 @@ def parse_and_solve(text: str) -> CandidateGraphResult:
if not choices:
if _ratified_registry:
from generate.recognizer_match import match as _recognizer_match
if _recognizer_match(s, _ratified_registry) is not None:
# Recognized — skip the sentence, do not refuse.
recognizer_match = _recognizer_match(s, _ratified_registry)
if recognizer_match is not None:
# ADR-0163.D.2 — per-category anchor injection.
# The matcher may carry populated parsed_anchors that
# an injector turns into typed solver primitives
# (CandidateInitial / CandidateOperation). When the
# injector returns a non-empty tuple, the recognized
# statement contributes math state to the Cartesian
# product the same way the existing parser's output
# does — and every constructed candidate has already
# passed _initial_admissible upstream of this call.
# When the injector returns () (skip-only fallback —
# the round-2 default and the only path for v1
# categories without an injector), the statement is
# dropped from per_sentence_choices, preserving the
# wrong=0 safety net by construction.
from generate.recognizer_anchor_inject import (
inject_from_match,
)
injected = inject_from_match(recognizer_match, s)
if injected:
admitted: list[SentenceChoice] = [
c for c in injected if _initial_admissible(c)
]
if len(admitted) == len(injected) and admitted:
per_sentence_choices.append(
_collapse_per_sentence_ties(admitted)
)
continue
# Recognized but no injection — skip the sentence, do
# not refuse. Identical to the round-2 skip-only
# wiring; preserves wrong=0 because zero math state
# is contributed.
continue
return CandidateGraphResult(
answer=None, selected_graph=None,

249
generate/recognizer_anchor_inject.py Normal file
View file

@ -0,0 +1,249 @@
"""ADR-0163.D.2 — per-category recognizer anchor injection.
When the candidate-graph pipeline's existing parser yields no candidates
for a statement AND the ratified recognizer registry recognizes the
statement, this module is consulted to build typed solver primitives
(``CandidateInitial`` / future ``CandidateOperation`` values) from the
recognizer's ``parsed_anchors``. The output extends ``per_sentence_choices``
the same way the existing parser's output does, so the downstream
solver runs unchanged.
Doctrine
--------
- Pure, deterministic injectors. Same ``(match, sentence)`` → same
``SentenceChoice`` tuple, byte-equal.
- Refusal-preferring: each injector returns ``()`` when it cannot build
a primitive that passes the existing ``_initial_admissible``
structural check (the wrong=0 safety net the candidate-graph already
enforces).
- No LLM / embeddings / learned classifiers; the injection is rules-only
same discipline as Phase A/C/D detection.
- Per-category boundary: v1 implements only ``discrete_count_statement``.
Every other category routes to the empty-tuple fallback (skip-only,
identical to the round-2 Phase D wiring) and lands in follow-up
D.2.x PRs after the framework's empirical lift is operator-reviewed.
Five-layer wrong=0 safety net (the Phase D.2 brief's load-bearing
section) is preserved across this module:
1. Matcher narrowness — ``recognizer_match._try_extract_discrete_count_anchor``
refuses on any ambiguity.
2. Extraction correctness — anchor fields ground in the literal
statement surface.
3. Injection correctness — the per-category injector returns a
``CandidateInitial`` that passes ``_initial_admissible``; failure
to ground yields ``()``.
4. Replay gate — propose-time ``run_admissibility_replay_gate``
auto-rejects any extraction change that lifts the GSM8K wrong
count.
5. Multi-branch decision rule — when an injected candidate disagrees
with another branch's answer, the candidate-graph refuses.
"""
from __future__ import annotations
from typing import Mapping
from evals.refusal_taxonomy.shape_categories import ShapeCategory
from generate.math_candidate_parser import CandidateInitial
from generate.math_problem_graph import InitialPossession, MathGraphError, Quantity
from generate.recognizer_match import RecognizerMatch
# ---------------------------------------------------------------------------
# Public surface
# ---------------------------------------------------------------------------
def inject_from_match(
match: RecognizerMatch,
sentence: str,
) -> tuple[CandidateInitial, ...]:
"""Dispatch a recognizer match to its per-category injector.
Returns an empty tuple when the category has no v1 injector or when
the v1 injector refused. Skip-only behavior (the round-2 default)
is the empty-tuple result.
"""
injector = _INJECTORS.get(match.category)
if injector is None:
return ()
return injector(match, sentence)
# ---------------------------------------------------------------------------
# Per-category injectors
# ---------------------------------------------------------------------------
def inject_discrete_count_statement(
match: RecognizerMatch,
sentence: str,
) -> tuple[CandidateInitial, ...]:
"""Build CandidateInitial(s) from ``discrete_count`` parsed anchors.
v1 narrowness: the matcher emits at most one anchor (further anchors
refuse extraction). When the anchor is absent (detection-only
fallback), the injector returns ``()`` and the candidate-graph
continues with the round-2 skip-only behavior.
"""
if not match.parsed_anchors:
return ()
out: list[CandidateInitial] = []
for anchor in match.parsed_anchors:
cand = _build_initial_from_discrete_count(anchor, sentence)
if cand is None:
# Under-admit on any failure to construct. The other
# already-built candidates for this sentence remain
# admissible only if they all pass; partial admission would
# mean the downstream Cartesian product enumerates a graph
# missing state — under-admit instead.
return ()
out.append(cand)
return tuple(out)
# ---------------------------------------------------------------------------
# Internals
# ---------------------------------------------------------------------------
def _build_initial_from_discrete_count(
anchor: Mapping[str, object],
sentence: str,
) -> CandidateInitial | None:
"""Construct one CandidateInitial from a discrete_count anchor.
Refuses (returns ``None``) when any field cannot be coerced or when
the constructed value would violate ``CandidateInitial`` /
``InitialPossession`` invariants. The resulting CandidateInitial is
structurally verified upstream by ``_initial_admissible``.
Anchor schema:
{
"kind": "discrete_count",
"subject_role": <str>,
"count_token": <str>, # '20' or 'two'
"count_kind": <"integer"|"word">,
"counted_noun": <str>, # 'paperclips' / 'Pokemon cards'
}
"""
subject_role = anchor.get("subject_role")
count_token = anchor.get("count_token")
count_kind = anchor.get("count_kind")
counted_noun = anchor.get("counted_noun")
if (
not isinstance(subject_role, str) or not subject_role
or not isinstance(count_token, str) or not count_token
or not isinstance(count_kind, str)
or not isinstance(counted_noun, str) or not counted_noun
):
return None
# Resolve the count token to a numeric value. v1 supports integer
# and single-word cardinals; hyphenated compounds defer to a follow-up
# PR because their resolution requires the language pack's
# parse_compound_cardinal helper which is not on this hot path.
value = _resolve_count_value(count_token, count_kind)
if value is None:
return None
# CandidateInitial requires an anchor verb token recognized in its
# post-init whitelist (has/have/had/owns/owned/holds/held/contains/
# contained — matched by the recognizer's narrowness rule). We pick
# the literal verb token from the sentence so the round-trip ground
# check inside _initial_admissible succeeds. Falls back to 'has' when
# the verb cannot be located in the surface; that fallback only fires
# when the recognizer's match diverges from the sentence and is the
# under-admit path.
verb_in_sentence = _locate_possession_verb(sentence)
if verb_in_sentence is None:
return None
try:
quantity = Quantity(value=value, unit=counted_noun)
initial = InitialPossession(entity=subject_role, quantity=quantity)
except MathGraphError:
return None
try:
return CandidateInitial(
initial=initial,
source_span=sentence,
matched_anchor=verb_in_sentence,
matched_value_token=count_token,
matched_unit_token=counted_noun,
matched_entity_token=subject_role,
)
except ValueError:
return None
def _resolve_count_value(count_token: str, count_kind: str) -> int | None:
"""Map ``count_token`` to a numeric value.
Integer tokens parse with ``int``. Word-form tokens look up
``WORD_NUMBERS`` from the language pack; unknown words refuse.
Hyphenated compounds (``twenty-five``) defer to D.2.x — v1 returns
``None`` for them.
"""
if count_kind == "integer":
try:
return int(count_token)
except ValueError:
return None
if count_kind == "word":
# Local import to keep module import-time cheap and to avoid a
# circular import via the math_candidate_parser surface.
from generate.math_roundtrip import WORD_NUMBERS
token_lc = count_token.lower()
if token_lc in WORD_NUMBERS:
return int(WORD_NUMBERS[token_lc])
# Hyphenated compound: defer to D.2.x.
return None
return None
def _locate_possession_verb(sentence: str) -> str | None:
"""Return the first possession-anchor verb (lowercased) found in
``sentence`` whitespace-tokenized, or ``None`` when absent.
The verb is the surface token that ``CandidateInitial.__post_init__``
validates against its registered anchor whitelist. Returning the
LITERAL surface keeps the round-trip ground check in
``_initial_admissible`` honest.
"""
possession_verbs = ("has", "have", "had")
for raw in sentence.split():
tok = raw.strip(".,;:!?\"'()[]{}").lower()
if tok in possession_verbs:
return tok
return None
# ---------------------------------------------------------------------------
# Dispatch table — keep deterministic and explicit.
# Adding a category here is the SINGLE place a new D.2.x category
# registers its injector. No global state, no side effects.
# ---------------------------------------------------------------------------
_INJECTORS: Mapping[ShapeCategory, "type"] = {
ShapeCategory.DISCRETE_COUNT_STATEMENT: inject_discrete_count_statement, # type: ignore[dict-item]
# The five other recognizer categories route to the empty-tuple
# fallback (skip-only) until their D.2.x injector lands:
#
# ShapeCategory.DESCRIPTIVE_SETUP_NO_QUANTITY — by design (no quantity)
# ShapeCategory.RATE_WITH_CURRENCY — D.2.2 follow-up
# ShapeCategory.TEMPORAL_AGGREGATION — D.2.3 follow-up
# ShapeCategory.MULTIPLICATIVE_AGGREGATION — D.2.4 follow-up
# ShapeCategory.CURRENCY_AMOUNT — D.2.5 follow-up
}
__all__ = [
"inject_from_match",
"inject_discrete_count_statement",
]

212
generate/recognizer_match.py
View file

@ -380,9 +380,9 @@ def _has_currency_symbol(statement: str) -> bool:
def _match_discrete_count_statement(
statement: str, spec: Mapping[str, Any]
) -> tuple[tuple[Mapping[str, Any], ...], Literal["count"]] | None:
"""Detection-only match for "X has N Y" shape.
"""ADR-0163.D.2 — extraction match for "X has N Y" shape.
Conditions:
Detection conditions (same as round-2 detection-only matcher):
- statement carries ≥1 quantity marker (digit or number word)
- statement does NOT carry a currency symbol (else currency_amount)
- statement does NOT carry per-unit framing (else rate_with_currency)
@ -390,8 +390,35 @@ def _match_discrete_count_statement(
(else temporal_aggregation)
- spec's anchor_kind is "discrete_count"
Returns ``(empty parsed_anchors, "count")`` on a hit; real value
extraction is Phase D.2 follow-up.
Extraction (D.2 v1) populates a SINGLE anchor when ALL of the
following narrowness rules hold; otherwise returns
``(empty parsed_anchors, "count")`` (detection-only fallback, same
skip-only safety as round 2). Narrowness layers (refusal-preferring,
wrong=0 doctrine):
1. Statement matches the canonical possession form
``<ProperNoun> <poss-verb> <count> <counted_noun>...``.
Subject must be a single capitalized proper noun (no
conjunctions, no leading pronoun). Possession verb must come
from the v1 closed whitelist (has/have/had); broader verbs
(owns/holds/contains) defer to a coordinated CandidateInitial
change in a follow-up PR.
2. Statement carries exactly ONE numeric token (digit or word
numeral) — a second count indicates multi-anchor content the
v1 schema cannot honor; refuse extraction.
3. Statement contains no clause-splitting connectives (``but``,
``then``, ``however``, ``before``, ``after``, ``and``,
``or``) — these indicate trailing operations or enumerations
that would invalidate a single InitialPossession.
4. count_kind ∈ spec.observed_count_kinds.
5. counted_noun ∈ spec.observed_counted_nouns (case-insensitive,
matched against the closed lemma list from Phase B/C).
The matcher returns ``(populated parsed_anchors, "count")`` on
extraction success, ``(tuple(), "count")`` on detection-only
fallback (skip-only safe), or ``None`` on detection failure.
Phase D.2's per-category injector consumes the populated anchors;
the empty-tuple fallback continues the round-2 skip-only behavior.
"""
if spec.get("anchor_kind") != "discrete_count":
return None
@ -404,9 +431,186 @@ def _match_discrete_count_statement(
return None
if _has_temporal_quantifier(padded):
return None
anchor = _try_extract_discrete_count_anchor(statement, padded, spec)
if anchor is not None:
return ((anchor,), "count")
return (tuple(), "count")
# ---------------------------------------------------------------------------
# ADR-0163.D.2 — discrete_count_statement value extraction (v1).
# ---------------------------------------------------------------------------
# Closed possession-verb whitelist. These verbs assert a static
# possession state (no goal, no acquisition event, no transfer). Verbs
# like 'collected', 'wants', 'lost', 'bought', etc. are deliberately
# omitted — they encode operations, not initial state, and admitting
# them as InitialPossession would over-extract.
#
# v1 intentionally restricts the surface to has/have/had so the
# extracted matched_anchor token is always accepted by the downstream
# CandidateInitial post-init whitelist. Widening to owns/holds/contains
# requires a coordinated CandidateInitial change and lands in a follow-up
# PR after the framework's empirical lift is operator-reviewed.
_POSSESSION_VERBS: Final[frozenset[str]] = frozenset({
"has", "have", "had",
})
# Pronoun subjects refused at extraction (ambiguous referent). The
# extractor requires a concrete proper-noun subject the source span can
# ground.
_REFUSED_SUBJECT_TOKENS: Final[frozenset[str]] = frozenset({
"he", "she", "they", "it", "we", "you", "i",
"him", "her", "them", "us",
})
# Clause-splitting / enumeration markers. Their presence indicates a
# second clause that may carry operations or additional anchors, so
# v1 refuses extraction (skip-only fallback preserves wrong=0).
_CLAUSE_SPLIT_TOKENS: Final[tuple[str, ...]] = (
" but ", " then ", " however ", " before ", " after ",
" and ", " or ", " while ", " until ", " unless ",
", and ", ", but ", ", or ", ", then ",
)
# Hyphenated compound cardinal: 'twenty-five', 'ninety-nine'. These
# are word-form counts. The narrowness rule below classifies any
# non-digit token in the count slot as count_kind='word'.
_HYPHEN_CARDINAL_RE: Final[re.Pattern[str]] = re.compile(r"^[a-z]+-[a-z]+$")
def _extract_discrete_count_re_for(counted_nouns: list[str]) -> re.Pattern[str]:
"""Build the extraction regex for a closed counted-noun set.
The counted-noun alternation is constructed from the spec's
``observed_counted_nouns``; multi-word nouns (e.g., ``Pokemon cards``)
are honored verbatim. Longest-first to prevent the alternation
swallowing a prefix.
"""
# Sort longest-first so 'Pokemon cards' wins over 'cards'.
options = sorted({n for n in counted_nouns if n}, key=len, reverse=True)
noun_alt = "|".join(re.escape(n) for n in options)
return re.compile(
r"^\s*"
r"(?P<subject>(?-i:[A-Z][a-z]+))" # case-sensitive proper noun
r"\s+(?P<verb>[A-Za-z]+)" # any word; verified against whitelist
r"\s+(?P<count>\d+|[A-Za-z\-]+)" # integer or word/hyphenated cardinal
r"\s+(?P<noun>" + noun_alt + r")"
r"(?:\b.*)?$", # optional trailing content
flags=re.IGNORECASE,
)
_DIGIT_RUN_RE: Final[re.Pattern[str]] = re.compile(r"\d+(?:\.\d+)?")
def _count_quantity_tokens(statement: str, padded_lower: str) -> int:
"""Total numeric tokens (digit runs + number words) in *statement*.
Used for the "exactly one count" narrowness rule. Hyphenated
cardinals count as one token; a multi-digit integer (``400``) counts
as one token, not as multiple single-digit hits.
"""
digit_hits = len(_DIGIT_RUN_RE.findall(statement))
word_hits = 0
for raw in padded_lower.split():
tok = raw.strip(".,;:!?\"'()[]{}").lower()
if tok in _NUMBER_WORDS:
word_hits += 1
elif _HYPHEN_CARDINAL_RE.match(tok):
# Hyphenated cardinal only counts when at least one half is
# a known number word.
left, _, right = tok.partition("-")
if left in _NUMBER_WORDS or right in _NUMBER_WORDS:
word_hits += 1
return digit_hits + word_hits
def _try_extract_discrete_count_anchor(
statement: str,
padded_lower: str,
spec: Mapping[str, Any],
) -> Mapping[str, Any] | None:
"""Refusal-preferring single-anchor extraction (D.2 v1).
Returns ``None`` when any narrowness layer fails — the caller then
falls back to skip-only detection. The returned anchor is the
discrete_count_statement schema dict: ``kind``, ``subject_role``,
``count_token``, ``count_kind``, ``counted_noun``.
"""
raw_kinds = spec.get("observed_count_kinds") or ()
raw_nouns = spec.get("observed_counted_nouns") or ()
observed_kinds: list[str] = [str(k) for k in raw_kinds]
observed_nouns: list[str] = [str(n) for n in raw_nouns]
if not observed_kinds or not observed_nouns:
return None
# Narrowness #3 — clause-split / enumeration markers.
for token in _CLAUSE_SPLIT_TOKENS:
if token in padded_lower:
return None
# Narrowness #2 — exactly one numeric token.
if _count_quantity_tokens(statement, padded_lower) != 1:
return None
# Narrowness #1 + #5 — shape + counted-noun lemma.
extract_re = _extract_discrete_count_re_for(observed_nouns)
m = extract_re.match(statement.strip())
if m is None:
return None
subject = m.group("subject")
if subject.lower() in _REFUSED_SUBJECT_TOKENS:
return None
verb = m.group("verb").lower()
if verb not in _POSSESSION_VERBS:
return None
count_token = m.group("count")
if count_token.isdigit():
count_kind = "integer"
else:
# Word-form cardinal — must be a known number word (single or
# hyphenated compound). Anything else is unrecognized and the
# extractor refuses.
lc = count_token.lower()
if lc in _NUMBER_WORDS:
count_kind = "word"
elif _HYPHEN_CARDINAL_RE.match(lc):
left, _, right = lc.partition("-")
if left in _NUMBER_WORDS or right in _NUMBER_WORDS:
count_kind = "word"
else:
return None
else:
return None
# Narrowness #4 — count_kind in observed set.
if count_kind not in observed_kinds:
return None
counted_noun = m.group("noun")
# Canonicalize counted_noun to the spec's observed casing where
# available; fall back to literal surface.
canon = counted_noun
counted_noun_lc = counted_noun.lower()
for observed_n in observed_nouns:
if observed_n.lower() == counted_noun_lc:
canon = observed_n
break
return {
"kind": "discrete_count",
"subject_role": subject,
"count_token": count_token,
"count_kind": count_kind,
"counted_noun": canon,
}
def _match_multiplicative_aggregation(
statement: str, spec: Mapping[str, Any]
) -> tuple[tuple[Mapping[str, Any], ...], Literal["aggregate"]] | None:

446
tests/test_adr_0163_d2_discrete_count_injection.py Normal file
View file

@ -0,0 +1,446 @@
"""ADR-0163.D.2 — discrete_count_statement injection v1.
This test file is the single load-bearing artifact of D.2 v1. It
enforces the wrong=0 safety net by testing five categories:
a. EXTRACTION CORRECTNESS — matcher extracts correct anchors.
b. EXTRACTION REFUSAL — matcher refuses on ambiguous shapes.
c. INJECTION CORRECTNESS — injector builds CandidateInitial that
passes _initial_admissible.
d. NO-FALSE-LIFT INVARIANT — synthetic adversarial cases never
produce a wrong answer.
e. END-TO-END LIFT — discrete_count injection wires through the
candidate-graph and lifts a refusal to a correct answer when
the statement is unambiguous and groundable.
"""
from __future__ import annotations
from evals.refusal_taxonomy.shape_categories import ShapeCategory
from generate.math_candidate_graph import parse_and_solve
from generate.math_candidate_parser import CandidateInitial
from generate.math_problem_graph import InitialPossession, Quantity
from generate.recognizer_anchor_inject import (
inject_discrete_count_statement,
inject_from_match,
)
from generate.recognizer_match import (
RecognizerMatch,
_padded_lower,
_try_extract_discrete_count_anchor,
match,
)
from generate.recognizer_registry import load_ratified_registry
# Spec mirror — kept locally so the tests don't depend on registry
# load order. The values mirror the ratified Phase C round-2 spec for
# discrete_count_statement.
_SPEC = {
"anchor_kind": "discrete_count",
"shape_category": "discrete_count_statement",
"graph_intent": "count",
"anchor_count_min": 1,
"anchor_count_max": 5,
"outcome": "admissible",
"observed_count_kinds": ["integer", "word"],
"observed_counted_nouns": [
"Pokemon cards", "apples", "balloons", "books", "cats",
"chickens", "dogs", "followers", "goat", "horses", "kittens",
"marbles", "motorcycles", "nephews", "paintbrushes",
"paperclips", "parakeets", "pounds", "puppies", "seashells",
"stickers", "sunflowers", "swallows", "turtles", "typewriters",
],
}
def _try_extract(statement: str):
return _try_extract_discrete_count_anchor(
statement, _padded_lower(statement), _SPEC,
)
def _ratified_registry():
"""Live ratified registry; resolved once for end-to-end tests."""
return load_ratified_registry()
# ---------------------------------------------------------------------------
# (a) Extraction correctness — matcher extracts the right anchors.
# ---------------------------------------------------------------------------
class TestExtractionCorrectness:
def test_basic_integer_count(self) -> None:
a = _try_extract("Sam has 5 apples.")
assert a == {
"kind": "discrete_count",
"subject_role": "Sam",
"count_token": "5",
"count_kind": "integer",
"counted_noun": "apples",
}
def test_past_tense_had(self) -> None:
a = _try_extract("Nicole had 400 paperclips.")
assert a is not None
assert a["subject_role"] == "Nicole"
assert a["count_token"] == "400"
assert a["count_kind"] == "integer"
assert a["counted_noun"] == "paperclips"
def test_word_form_count(self) -> None:
a = _try_extract("Sam has twenty books.")
assert a is not None
assert a["count_token"] == "twenty"
assert a["count_kind"] == "word"
def test_hyphenated_word_form(self) -> None:
a = _try_extract("Sam has twenty-five books.")
assert a is not None
assert a["count_token"] == "twenty-five"
assert a["count_kind"] == "word"
def test_multi_word_counted_noun(self) -> None:
a = _try_extract("Sam has 5 Pokemon cards.")
assert a is not None
# Canonicalized to spec casing.
assert a["counted_noun"] == "Pokemon cards"
def test_trailing_modifier(self) -> None:
# Trailing prepositional phrase is allowed; the regex anchors
# on the noun and tolerates benign tail content (no
# clause-split markers).
a = _try_extract("Sam has 5 apples on the table.")
assert a is not None
assert a["count_token"] == "5"
assert a["counted_noun"] == "apples"
# ---------------------------------------------------------------------------
# (b) Extraction refusal — refuse on ambiguity, never over-admit.
# ---------------------------------------------------------------------------
class TestExtractionRefusal:
def test_multi_subject_refused(self) -> None:
assert _try_extract("Tom and Mary have 5 apples.") is None
def test_indefinite_quantifier_refused(self) -> None:
# 'some apples' — no concrete count. The detection-level
# check filters this through _has_any_quantity_marker
# already, but the extractor must independently refuse when
# the count token cannot be resolved.
assert _try_extract("Sam has some apples.") is None
def test_missing_counted_noun_refused(self) -> None:
assert _try_extract("Sam has 5.") is None
def test_pronoun_subject_refused(self) -> None:
assert _try_extract("He has 5 apples.") is None
def test_lowercase_subject_refused(self) -> None:
assert _try_extract("sam has 5 apples.") is None
def test_clause_split_refused(self) -> None:
# "but then" indicates a trailing operation; v1 refuses.
assert _try_extract(
"Yun had 20 paperclips initially, but then lost 12."
) is None
def test_enumeration_and_refused(self) -> None:
# Multi-anchor enumeration: " and " split refuses.
assert _try_extract(
"Malcolm has 240 followers on Instagram and 500 followers on Facebook."
) is None
def test_multi_count_refused(self) -> None:
# Two digit runs — v1 admits exactly one count.
assert _try_extract("He has 2 horses, 5 dogs.") is None
def test_unobserved_counted_noun_refused(self) -> None:
# 'widgets' is not in the spec's observed_counted_nouns.
assert _try_extract("Sam has 5 widgets.") is None
def test_non_possession_verb_refused(self) -> None:
# 'wants', 'collected', 'bought' — operation verbs, not state.
assert _try_extract("Michael wants 10 pounds.") is None
assert _try_extract("Nicole collected 400 paperclips.") is None
assert _try_extract("Sam bought 5 apples.") is None
def test_owns_outside_v1_whitelist(self) -> None:
# v1 restricts to has/have/had to align with CandidateInitial's
# post-init whitelist. Broader possession verbs (owns/holds/
# contains) defer to follow-up.
assert _try_extract("Sam owns 12 books.") is None
# ---------------------------------------------------------------------------
# (c) Injection correctness — built CandidateInitial passes the
# structural admissibility check.
# ---------------------------------------------------------------------------
def _make_match(parsed_anchors) -> RecognizerMatch:
"""Build a synthetic RecognizerMatch for injector unit tests."""
from generate.recognizer_registry import RatifiedRecognizer
rec = RatifiedRecognizer(
proposal_id="test-discrete-count",
shape_category=ShapeCategory.DISCRETE_COUNT_STATEMENT,
canonical_pattern=dict(_SPEC),
spec_digest="test-digest",
review_date="2026-05-27",
ratified_at_revision="test",
)
return RecognizerMatch(
recognizer=rec,
category=ShapeCategory.DISCRETE_COUNT_STATEMENT,
outcome="admissible",
graph_intent="count",
parsed_anchors=tuple(parsed_anchors),
)
class TestInjectionCorrectness:
def test_injects_candidate_initial(self) -> None:
m = _make_match([{
"kind": "discrete_count",
"subject_role": "Sam",
"count_token": "5",
"count_kind": "integer",
"counted_noun": "apples",
}])
out = inject_discrete_count_statement(m, "Sam has 5 apples.")
assert len(out) == 1
cand = out[0]
assert isinstance(cand, CandidateInitial)
assert cand.initial == InitialPossession(
entity="Sam",
quantity=Quantity(value=5, unit="apples"),
)
assert cand.matched_anchor == "has"
assert cand.matched_value_token == "5"
assert cand.matched_unit_token == "apples"
assert cand.matched_entity_token == "Sam"
assert cand.source_span == "Sam has 5 apples."
def test_injects_word_form(self) -> None:
m = _make_match([{
"kind": "discrete_count",
"subject_role": "Sam",
"count_token": "twenty",
"count_kind": "word",
"counted_noun": "books",
}])
out = inject_discrete_count_statement(m, "Sam has twenty books.")
assert len(out) == 1
assert out[0].initial.quantity.value == 20
def test_empty_parsed_anchors_returns_empty(self) -> None:
m = _make_match([])
out = inject_discrete_count_statement(m, "Sam has 5 apples.")
assert out == ()
def test_injector_passes_initial_admissible(self) -> None:
# The candidate-graph's _initial_admissible MUST accept the
# injected CandidateInitial. This is the structural-grounding
# safety net.
from generate.math_candidate_graph import _initial_admissible
m = _make_match([{
"kind": "discrete_count",
"subject_role": "Sam",
"count_token": "5",
"count_kind": "integer",
"counted_noun": "apples",
}])
out = inject_discrete_count_statement(m, "Sam has 5 apples.")
assert out
assert _initial_admissible(out[0]) is True
def test_dispatch_routes_to_per_category_injector(self) -> None:
m = _make_match([{
"kind": "discrete_count",
"subject_role": "Sam",
"count_token": "5",
"count_kind": "integer",
"counted_noun": "apples",
}])
out_dispatch = inject_from_match(m, "Sam has 5 apples.")
out_direct = inject_discrete_count_statement(m, "Sam has 5 apples.")
assert out_dispatch == out_direct
def test_dispatch_unsupported_category_returns_empty(self) -> None:
from generate.recognizer_registry import RatifiedRecognizer
rec = RatifiedRecognizer(
proposal_id="test-rate",
shape_category=ShapeCategory.RATE_WITH_CURRENCY,
canonical_pattern={},
spec_digest="test",
review_date="2026-05-27",
ratified_at_revision="test",
)
m = RecognizerMatch(
recognizer=rec,
category=ShapeCategory.RATE_WITH_CURRENCY,
outcome="admissible",
graph_intent="rate",
parsed_anchors=({"any": "thing"},),
)
assert inject_from_match(m, "Tina makes $18.00 an hour.") == ()
def test_injection_under_admits_on_unresolvable_verb(self) -> None:
# If the source sentence has no possession-anchor verb, the
# injector refuses (returns ()). This is the under-admit
# safety net for matcher/sentence disagreement.
m = _make_match([{
"kind": "discrete_count",
"subject_role": "Sam",
"count_token": "5",
"count_kind": "integer",
"counted_noun": "apples",
}])
out = inject_discrete_count_statement(m, "Sam collected 5 apples.")
assert out == ()
# ---------------------------------------------------------------------------
# (d) No-false-lift invariant — adversarial cases never produce a
# wrong answer. The case must either refuse or produce the
# entity-consistent correct answer; wrong=0 is non-negotiable.
# ---------------------------------------------------------------------------
class TestNoFalseLiftInvariant:
def test_clause_split_adversarial(self) -> None:
# "Yun had 20 paperclips initially, but then lost 12. How
# many paperclips does Yun have?" Wrong reading is 20;
# correct reading is 8. The matcher MUST refuse extraction
# so the case refuses.
r = parse_and_solve(
"Yun had 20 paperclips initially, but then lost 12. "
"How many paperclips does Yun have?"
)
# Under v1, this refuses; the answer must never be 20.0.
assert r.answer != 20
assert r.answer != 20.0
def test_enumeration_adversarial(self) -> None:
# "Malcolm has 240 followers on Instagram and 500 followers
# on Facebook. How many followers does Malcolm have?" Wrong
# reading injects only 240 (missing the 500); a wrong=0
# violation if admitted. The matcher MUST refuse.
r = parse_and_solve(
"Malcolm has 240 followers on Instagram and 500 followers on Facebook. "
"How many followers does Malcolm have?"
)
assert r.answer != 240
assert r.answer != 240.0
def test_branch_disagreement_safety_net(self) -> None:
# Construct a problem where the existing parser already
# handles the statement; the recognizer would also match but
# the injection path is never consulted because choices is
# non-empty. This proves injection is upstream-gated.
r = parse_and_solve(
"Sam has 5 apples. Sam buys 3 apples. "
"How many apples does Sam have?"
)
assert r.is_admitted
assert r.answer == 8
def test_existing_parser_unchanged_for_canonical_form(self) -> None:
# Canonical "X has N Y" is handled by the existing parser
# without ever reaching injection. Confirms no behavioral
# regression on the base case.
r = parse_and_solve("Sam has 5 apples. How many apples does Sam have?")
assert r.is_admitted
assert r.answer == 5
# ---------------------------------------------------------------------------
# (e) End-to-end lift — injection wires through and lifts a refusal
# to a correct answer when the statement is unambiguous and
# groundable.
# ---------------------------------------------------------------------------
class TestEndToEndLift:
def test_trailing_clause_lift(self) -> None:
# The existing _INITIAL_HAS_RE refuses statements with
# arbitrary trailing prepositional phrases (e.g., 'on the
# table top above the shelf'). The discrete_count matcher
# admits, the injector builds a CandidateInitial, and the
# solver answers correctly.
problem = (
"Sam has 5 apples on the table top above the shelf where books are. "
"How many apples does Sam have?"
)
r = parse_and_solve(problem)
assert r.is_admitted
assert r.answer == 5
def test_lift_uses_recognizer_path(self) -> None:
# Confirm the lift specifically comes through recognizer
# injection: the same sentence in isolation produces zero
# candidates from the existing parser.
from generate.math_candidate_graph import _filtered_statement_choices
s = "Sam has 5 apples on the table top above the shelf where books are."
assert _filtered_statement_choices(s) == []
# But the recognizer matches it.
m = match(s, _ratified_registry())
assert m is not None
assert m.category is ShapeCategory.DISCRETE_COUNT_STATEMENT
assert m.parsed_anchors # non-empty
def test_unobserved_noun_refuses_end_to_end(self) -> None:
# 'widgets' is not in the spec's observed_counted_nouns.
# The detection-only fallback is taken (skip-only), but the
# question still needs an entity ground — without state, the
# problem refuses.
r = parse_and_solve(
"Sam has 5 widgets blah blah blah blah blah. "
"How many widgets does Sam have?"
)
assert not r.is_admitted
assert r.answer is None
# ---------------------------------------------------------------------------
# Replay-gate sanity (safety net #4) — the existing replay gate is
# evaluated outside the test harness, but the injection MUST be
# deterministic so the gate's byte-equality comparison holds.
# ---------------------------------------------------------------------------
class TestDeterminism:
def test_extraction_is_deterministic(self) -> None:
s = "Sam has 5 apples."
a1 = _try_extract(s)
a2 = _try_extract(s)
assert a1 == a2
def test_injection_is_deterministic(self) -> None:
m = _make_match([{
"kind": "discrete_count",
"subject_role": "Sam",
"count_token": "5",
"count_kind": "integer",
"counted_noun": "apples",
}])
out1 = inject_discrete_count_statement(m, "Sam has 5 apples.")
out2 = inject_discrete_count_statement(m, "Sam has 5 apples.")
assert out1 == out2
def test_end_to_end_is_deterministic(self) -> None:
problem = (
"Sam has 5 apples on the table top above the shelf where books are. "
"How many apples does Sam have?"
)
r1 = parse_and_solve(problem)
r2 = parse_and_solve(problem)
assert r1.answer == r2.answer
assert r1.refusal_reason == r2.refusal_reason