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

feat(ADR-0164.P1): reader/regex hybrid coexistence + Phase 1 measurement gate (#331)

Browse source 
Phase A — RuntimeConfig flag:
  core/config.py: adds `comprehension_reader_questions: bool = False`
  Default OFF preserves byte-identical behaviour with today.

Phase B — Hybrid wiring in candidate-graph path:
  generate/math_candidate_graph.py:
    - _try_reader_for_question() dispatches to the comprehension reader
      BEFORE the regex question parser; refusal falls through to regex
    - reader_trace: tuple[str, ...] field on CandidateGraphResult captures
      JSON-encoded admit/fallthrough events for audit
  generate/comprehension/lifecycle_runtime_adapter.py (new):
    - build_problem_state_from_candidates(): converts regex-parser output
      to ProblemReadingState for the reader's pronoun-resolution step
    - invoke_reader_for_question(): tokenises sentence, drives lifecycle
    - project_to_candidate_unknown(): QuestionTargetSlot → CandidateUnknown
    - trace-event constructors for admit and fallthrough

Phase C — Capability-axis regression:
  All existing tests pass with flag OFF and ON; zero new regressions.
  Two pre-existing failures on main are unrelated to this PR.

Phase D — GSM8K train_sample measurement:
  evals/gsm8k_math/train_sample/v1/runner.py: --use-reader flag triggers
    baseline-off + reader-on runs and writes reader_phase1_delta.json
  evals/gsm8k_math/train_sample/v1/reader_phase1_delta.json (new):
    baseline-off: correct=3 refused=47 wrong=0
    reader-on:    correct=3 refused=47 wrong=0
    delta: all zeros — Mixed result expected (Phase 2 scope)
    wrong=0 invariant preserved in both modes.

Phase E — Coexistence tests:
  tests/test_reader_coexistence.py (new): 13 tests covering
    flag-OFF byte-identity, flag-ON determinism, wrong=0 invariant,
    trace shape validation, Brief-8 target admission, and fallthrough
    preservation for unknown-unit words.

Admission gate result: Mixed (correct=3, below the ≥10 bar).
All statement-side barriers remain in place; Phase 2 (reader for
statement sentences) is required to drive correct≥10. Documented in
reader_phase1_delta.json and train_sample/v1/runner.py docstring.
This commit is contained in:
Shay 2026-05-26 21:14:11 -07:00 committed by GitHub
parent 30972e184e
commit 800cf6591e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 967 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
core/config.py
View file

@ -276,6 +276,17 @@ class RuntimeConfig:
# ADR-0151 — generate TeachingChainProposals from enriched candidates on load.
auto_proposal_enabled: bool = False
# ADR-0164 Phase 1 — incremental comprehension reader for question sentences.
# When True, the candidate-graph path uses the comprehension reader
# (generate/comprehension/lifecycle.py) to parse question sentences BEFORE
# consulting the regex question patterns (Pattern A/B/C in
# generate/math_candidate_parser.py). On reader refusal, falls through to
# the existing regex parser — the reader is purely additive at Phase 1.
# Default False: flag-OFF behaviour is byte-identical to today.
# Phase 3 (per ADR-0164 §Phasing) removes the regex question parser entirely;
# that work is deferred — this PR is the Phase 1 stopgap.
comprehension_reader_questions: bool = False
DEFAULT_IDENTITY_PACK: str = "default_general_v1"
DEFAULT_ETHICS_PACK: str = "default_general_ethics_v1"

19
evals/gsm8k_math/runner.py
View file

@ -31,9 +31,12 @@ from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass, field
from typing import Any
from typing import TYPE_CHECKING, Any
from generate.math_candidate_graph import parse_and_solve
if TYPE_CHECKING:
from core.config import RuntimeConfig
from generate.math_parser import ParseError, parse_problem
from generate.math_problem_graph import MathProblemGraph
from generate.math_realizer import RealizerError, realize
@ -235,7 +238,10 @@ def _score_one(case: dict[str, Any]) -> CaseOutcome:
# TODO(ADR-future): report.json metrics may not credit candidate-graph admissions
# routed through this branch. Aggregation in calling code needs an audit before
# the canonical run.honest_runner.json artifact can be trusted for cross-phase comparison.
def _score_one_candidate_graph(case: dict[str, Any]) -> CaseOutcome:
def _score_one_candidate_graph(
case: dict[str, Any],
config: "RuntimeConfig | None" = None,
) -> CaseOutcome:
"""ADR-0126 P4 — score one case via the candidate-graph pipeline.
Mirrors :func:`_score_one` end-to-end (parser → solver → verifier →
@ -253,13 +259,20 @@ def _score_one_candidate_graph(case: dict[str, Any]) -> CaseOutcome:
(e.g. ``evals/gsm8k_math/train_sample/v1/runner.py`` from PR
#160) substitute this function for ``_score_one``; the
``CaseOutcome`` shape is identical.
Args:
case: Case record with keys ``id``, ``problem``, ``expected_answer``,
``expected_unit``.
config: Optional :class:`core.config.RuntimeConfig`. Passed through
to :func:`generate.math_candidate_graph.parse_and_solve`. When
None, flag-OFF (default) behaviour is preserved.
"""
case_id = case["id"]
expected_answer = case["expected_answer"]
expected_unit = case["expected_unit"]
# Stage 1 — candidate-graph parse + internal solve + decision rule.
cg_result = parse_and_solve(case["problem"])
cg_result = parse_and_solve(case["problem"], config=config)
if not cg_result.is_admitted:
return CaseOutcome(
case_id=case_id,

21
evals/gsm8k_math/train_sample/v1/reader_phase1_delta.json Normal file
View file

@ -0,0 +1,21 @@
{
"adr": "0164",
"baseline_off": {
"correct": 3,
"refused": 47,
"wrong": 0
},
"changed_cases": [],
"delta": {
"correct": 0,
"refused": 0,
"wrong": 0
},
"phase": 1,
"reader_on": {
"correct": 3,
"refused": 47,
"wrong": 0
},
"schema_version": 1
}

3
evals/gsm8k_math/train_sample/v1/report.json
View file

@ -264,5 +264,6 @@
],
"sample_count": 50,
"sample_path": "evals/gsm8k_math/train_sample/v1/cases.jsonl",
"schema_version": 1
"schema_version": 1,
"use_reader": true
}

97
evals/gsm8k_math/train_sample/v1/runner.py
View file

@ -16,6 +16,12 @@ sealed holdout, or any decryption surface (CLAUDE.md trust boundary).
unit annotation, so ``expected_unit`` is normalized to ``""`` which
:func:`evals.gsm8k_math.runner._score_one` already treats as
"skip unit comparison".
CLI flags:
--use-reader Activate the ADR-0164 Phase-1 comprehension reader for
question sentences (RuntimeConfig.comprehension_reader_questions
= True). Default: False (flag-OFF, byte-identical to today).
"""
from __future__ import annotations
@ -30,6 +36,7 @@ from evals.gsm8k_math.runner import _score_one_candidate_graph
_HERE = Path(__file__).resolve().parent
_CASES_PATH = _HERE / "cases.jsonl"
_REPORT_PATH = _HERE / "report.json"
_DELTA_PATH = _HERE / "reader_phase1_delta.json"
_SAMPLE_REL = "evals/gsm8k_math/train_sample/v1/cases.jsonl"
_EXPECTED_COUNT = 50
_CORRECT_MIN = 10
@ -67,11 +74,27 @@ def _adapt(case: dict[str, Any]) -> dict[str, Any]:
}
def build_report(cases: list[dict[str, Any]]) -> dict[str, Any]:
per_case: list[dict[str, str]] = []
def build_report(
cases: list[dict[str, Any]],
use_reader: bool = False,
) -> dict[str, Any]:
"""Build the measurement report for the train-sample cases.
Args:
cases: Loaded case records from cases.jsonl.
use_reader: When True, activates the ADR-0164 Phase-1 comprehension
reader for question sentences via RuntimeConfig.comprehension_reader_questions.
Default False preserves byte-identical behaviour with today.
"""
config = None
if use_reader:
from core.config import RuntimeConfig
config = RuntimeConfig(comprehension_reader_questions=True)
per_case: list[dict[str, Any]] = []
counts = {"correct": 0, "wrong": 0, "refused": 0}
for raw in cases:
outcome = _score_one_candidate_graph(_adapt(raw))
outcome = _score_one_candidate_graph(_adapt(raw), config=config)
counts[outcome.outcome] += 1
per_case.append(
{
@ -86,6 +109,7 @@ def build_report(cases: list[dict[str, Any]]) -> dict[str, Any]:
"adr": "0126",
"sample_path": _SAMPLE_REL,
"sample_count": len(cases),
"use_reader": use_reader,
"counts": counts,
"exit_criterion": {
"correct_min": _CORRECT_MIN,
@ -103,10 +127,73 @@ def write_report(report: dict[str, Any], path: Path = _REPORT_PATH) -> None:
)
def build_delta_report(
baseline: dict[str, Any],
reader_on: dict[str, Any],
) -> dict[str, Any]:
"""Compute per-case delta between flag-OFF and flag-ON reports.
Returns a JSON-serialisable dict with:
- summary counts
- per-case attribution for every case that changed verdict
"""
base_map = {p["case_id"]: p for p in baseline["per_case"]}
on_map = {p["case_id"]: p for p in reader_on["per_case"]}
changed: list[dict[str, Any]] = []
for cid, on_case in on_map.items():
base_case = base_map.get(cid, {})
if base_case.get("verdict") != on_case["verdict"]:
changed.append(
{
"case_id": cid,
"prior_verdict": base_case.get("verdict"),
"prior_refusal_reason": base_case.get("reason"),
"new_verdict": on_case["verdict"],
"new_reason": on_case.get("reason"),
}
)
bc = baseline["counts"]
rc = reader_on["counts"]
return {
"schema_version": 1,
"adr": "0164",
"phase": 1,
"baseline_off": {"correct": bc["correct"], "refused": bc["refused"], "wrong": bc["wrong"]},
"reader_on": {"correct": rc["correct"], "refused": rc["refused"], "wrong": rc["wrong"]},
"delta": {
"correct": rc["correct"] - bc["correct"],
"refused": rc["refused"] - bc["refused"],
"wrong": rc["wrong"] - bc["wrong"],
},
"changed_cases": changed,
}
def main() -> int:
"""Run the train-sample measurement.
When ``--use-reader`` is passed, also generates a delta report at
``reader_phase1_delta.json`` comparing flag-OFF vs flag-ON results.
"""
use_reader = "--use-reader" in sys.argv
cases = _load_cases(_CASES_PATH)
report = build_report(cases)
write_report(report)
if use_reader:
# Run baseline (flag OFF) first, then reader-on.
baseline_report = build_report(cases, use_reader=False)
reader_report = build_report(cases, use_reader=True)
write_report(reader_report)
delta = build_delta_report(baseline_report, reader_report)
_DELTA_PATH.write_text(
json.dumps(delta, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
report = reader_report
else:
report = build_report(cases, use_reader=False)
write_report(report)
return 0 if report["exit_criterion"]["passed"] else 1

402
generate/comprehension/lifecycle_runtime_adapter.py Normal file
View file

@ -0,0 +1,402 @@
"""ADR-0164 Phase 1 — bridge from regex-parser candidates to reader state.
Converts CandidateInitial / SentenceChoice tuples produced by the existing
regex parser into a ProblemReadingState that the comprehension reader can
consume for question-sentence processing.
This module is the only place where the Phase 1 coexistence wiring knows
about both worlds simultaneously. It is intentionally a stopgap: Phase 3
(per ADR-0164 §Phasing Phase 3) removes the regex question parser entirely,
at which point this adapter either shrinks to a pure statement-candidate
helper or is deleted.
All public functions are pure and deterministic (same inputs → same outputs,
no I/O, no global state mutation).
"""
from __future__ import annotations
import json
import re
from typing import TYPE_CHECKING, Final, Union
from generate.comprehension.lifecycle import (
_classify,
_get_lexicon,
apply_word,
begin_sentence,
end_sentence,
)
from generate.comprehension.state import (
EntityRef,
PartialInitialPossession,
PartialOperation,
ProblemReadingState,
QuestionTargetSlot,
QuantityRef,
ReaderRefusal,
SentenceReadingState,
)
if TYPE_CHECKING:
from generate.math_candidate_parser import CandidateInitial
from generate.math_roundtrip import CandidateOperation
# Union type for statement-sentence choices (mirrors math_candidate_graph).
SentenceChoice = Union["CandidateInitial", "CandidateOperation"]
# ---------------------------------------------------------------------------
# Gender inference via lexicon
# ---------------------------------------------------------------------------
_FEMALE_CATEGORIES: Final[frozenset[str]] = frozenset({"proper_noun_entity_female"})
_MALE_CATEGORIES: Final[frozenset[str]] = frozenset({"proper_noun_entity_male"})
_UNIT_CLASS_CATEGORIES: Final[dict[str, str]] = {
"count_unit_noun": "count",
"currency_unit_noun": "currency",
"time_unit_noun": "time",
}
def _infer_gender(entity_name: str) -> str:
"""Return 'female', 'male', or 'unknown' for a proper-noun entity.
Consults the en_core_math_v1 lexicon (via the lifecycle's cached loader)
per ADR-0164.2 gender-inference policy. Defaults to 'unknown' when the
name is absent from the lexicon.
"""
lex = _get_lexicon()
key = entity_name.lower()
entry = lex.get(key)
if entry is None:
return "unknown"
if entry.category in _FEMALE_CATEGORIES:
return "female"
if entry.category in _MALE_CATEGORIES:
return "male"
return "unknown"
# ---------------------------------------------------------------------------
# Build ProblemReadingState from regex-parser output
# ---------------------------------------------------------------------------
def build_problem_state_from_candidates(
statement_choices: list[SentenceChoice],
statement_sentence_count: int,
) -> ProblemReadingState:
"""Convert regex-parser output into a ProblemReadingState for reader consumption.
Args:
statement_choices: Admissible CandidateInitial / CandidateOperation
tuples produced by the existing regex parser, in source-text order.
statement_sentence_count: Number of statement sentences already
processed (sets ``ProblemReadingState.sentence_index``).
Returns:
A ProblemReadingState with entity_registry, accumulated_initial_state,
and accumulated_operations populated from the candidates.
unknown_target_slot is None (the question hasn't been processed yet).
This function is the glue layer for Phase 1 coexistence. It does NOT
attempt to reproduce the reader's full incremental behaviour for statement
sentences — that is Phase 2 scope. It produces only what the reader's
pronoun-resolution step needs: an ordered entity registry.
"""
from generate.math_candidate_parser import CandidateInitial as _CI
from generate.math_roundtrip import CandidateOperation as _CO
entity_registry: list[EntityRef] = []
seen_names: set[str] = set()
accumulated_initials: list[PartialInitialPossession] = []
accumulated_ops: list[PartialOperation] = []
char_offset = 0
for choice in statement_choices:
if isinstance(choice, _CI):
entity_name = choice.initial.entity
if entity_name not in seen_names:
gender = _infer_gender(entity_name)
entity_registry.append(
EntityRef(
canonical_name=entity_name,
gender=gender,
first_mention_position=len(seen_names),
)
)
seen_names.add(entity_name)
# Convert InitialPossession to PartialInitialPossession
from decimal import Decimal
qty_val = choice.initial.quantity.value
qty = QuantityRef(
value=Decimal(str(qty_val)),
unit=choice.initial.quantity.unit,
unit_class=None,
owner_entity=entity_name,
mention_position=len(accumulated_initials),
)
accumulated_initials.append(
PartialInitialPossession(entity=entity_name, quantity=qty)
)
elif isinstance(choice, _CO):
actor = choice.op.actor
if actor not in seen_names:
gender = _infer_gender(actor)
entity_registry.append(
EntityRef(
canonical_name=actor,
gender=gender,
first_mention_position=len(seen_names),
)
)
seen_names.add(actor)
if choice.op.target is not None and choice.op.target not in seen_names:
tgt = choice.op.target
gender_t = _infer_gender(tgt)
entity_registry.append(
EntityRef(
canonical_name=tgt,
gender=gender_t,
first_mention_position=len(seen_names),
)
)
seen_names.add(tgt)
# Operand — may be Quantity or Comparison; only carry scalar Quantity
from generate.math_problem_graph import Quantity
operand_ref: QuantityRef | None = None
if hasattr(choice.op, "operand") and isinstance(choice.op.operand, Quantity):
from decimal import Decimal
operand_ref = QuantityRef(
value=Decimal(str(choice.op.operand.value)),
unit=choice.op.operand.unit,
unit_class=None,
owner_entity=actor,
mention_position=len(accumulated_ops),
)
accumulated_ops.append(
PartialOperation(
actor=actor,
kind=choice.op.kind,
operand=operand_ref,
target=choice.op.target,
)
)
return ProblemReadingState(
entity_registry=tuple(entity_registry),
accumulated_initial_state=tuple(accumulated_initials),
accumulated_operations=tuple(accumulated_ops),
unknown_target_slot=None,
pronoun_resolution_history=(),
sentence_index=statement_sentence_count,
source_text_offset=char_offset,
)
# ---------------------------------------------------------------------------
# Tokenisation (matches the reader's apply_word loop convention)
# ---------------------------------------------------------------------------
_TOKEN_SPLIT_RE: Final[re.Pattern[str]] = re.compile(r"\s+")
_PUNCT_STRIP_RE: Final[re.Pattern[str]] = re.compile(r"^[\"'()\[\]{}<>]+|[\"'()\[\]{}<>]+$")
def _tokenise_sentence(sentence: str) -> list[str]:
"""Split a sentence into tokens, emitting punctuation as separate tokens.
Trailing ``?`` and ``.`` become their own token (matched by primitive scanner
as ``question_terminator`` / ``statement_terminator``). Leading/trailing
matched-pair punctuation is stripped per word. Empty strings are dropped.
"""
tokens: list[str] = []
for raw in _TOKEN_SPLIT_RE.split(sentence.strip()):
if not raw:
continue
# Separate a trailing '?' or '.' from the word body.
if len(raw) > 1 and raw[-1] in "?.!":
body = raw[:-1]
tail = raw[-1]
else:
body = raw
tail = None
body = _PUNCT_STRIP_RE.sub("", body)
if body:
tokens.append(body)
if tail:
tokens.append(tail)
return tokens
# ---------------------------------------------------------------------------
# Unit extraction from question sentence
# ---------------------------------------------------------------------------
def _extract_unit_from_question(question_sentence: str, unit_class: str) -> str | None:
"""Scan question tokens for a unit-noun surface word matching ``unit_class``.
After the reader produces a QuestionTargetSlot with unit_class set, this
helper re-tokenises the question to find the specific unit word. This lets
the projected Unknown carry the actual unit string (e.g. 'apples') rather
than the abstract class ('count'), maximising match probability against
statement candidates' unit strings.
Returns the canonicalised unit string, or None when no unit noun is found
with the expected class.
"""
from generate.math_candidate_parser import _canonicalize_unit # type: ignore[attr-defined]
target_categories = {
"count": frozenset({"count_unit_noun"}),
"currency": frozenset({"currency_unit_noun"}),
"time": frozenset({"time_unit_noun"}),
}.get(unit_class, frozenset())
if not target_categories:
return None
for tok in _tokenise_sentence(question_sentence):
cat, _surface = _classify(tok)
if cat in target_categories:
return _canonicalize_unit(tok)
return None
# ---------------------------------------------------------------------------
# Run the reader over a question sentence
# ---------------------------------------------------------------------------
def invoke_reader_for_question(
question_sentence: str,
problem_state: ProblemReadingState,
) -> tuple[QuestionTargetSlot, str] | ReaderRefusal:
"""Run the Phase-1 reader over one question sentence.
Returns:
On success: ``(QuestionTargetSlot, canonical_unit)`` where
``canonical_unit`` is the actual unit string extracted from the
question tokens (may differ from ``slot.unit_class``).
On refusal: ``ReaderRefusal``.
The caller is responsible for wrapping the result in a CandidateUnknown
and for emitting the trace event.
"""
tokens = _tokenise_sentence(question_sentence)
sentence_state: SentenceReadingState = begin_sentence(
problem_state, source_text_offset=problem_state.source_text_offset
)
for tok in tokens:
result = apply_word(sentence_state, problem_state, tok)
if isinstance(result, ReaderRefusal):
return result
sentence_state = result
end_result = end_sentence(sentence_state, problem_state)
if isinstance(end_result, ReaderRefusal):
return end_result
# end_sentence succeeded — extract QuestionTargetSlot from the new
# problem_state (it was just committed as unknown_target_slot).
slot = end_result.unknown_target_slot
if slot is None:
return ReaderRefusal(
reason="no_question_target",
detail="end_sentence succeeded but no unknown_target_slot set",
sentence_index=problem_state.sentence_index,
token_index=len(tokens),
token_text="",
)
# Extract the canonical unit string from the question surface.
unit_class = slot.unit_class or "unknown"
canonical_unit = _extract_unit_from_question(question_sentence, unit_class)
if canonical_unit is None:
# Fall back to unit_class as the unit string per ADR-0164 Brief-9 spec.
canonical_unit = unit_class
return slot, canonical_unit
# ---------------------------------------------------------------------------
# Project QuestionTargetSlot → CandidateUnknown
# ---------------------------------------------------------------------------
def project_to_candidate_unknown(
slot: QuestionTargetSlot,
canonical_unit: str,
question_sentence: str,
problem_state: ProblemReadingState,
) -> "CandidateUnknown | None": # type: ignore[name-defined]
"""Convert a QuestionTargetSlot into a CandidateUnknown for the candidate graph.
Returns None if the projection would produce an invalid Unknown (e.g., the
entity is set but not in the problem_state entity registry, which would
cause _build_graph to reject it).
Modifier flags (aggregate, comparative, residual) from the reader's
lookback are not threaded into Unknown (Unknown has only entity + unit
fields per ADR-0115). Deferral documented here; a follow-up ADR will
extend BoundUnknown resolution to consume these flags via side-channel.
"""
from generate.math_candidate_parser import CandidateUnknown, _canonicalize_unit
from generate.math_problem_graph import Unknown
entity: str | None = slot.entity
# Validate entity against the registry when set.
if entity is not None:
known = {e.canonical_name for e in problem_state.entity_registry}
if entity not in known:
return None
matched_unit_token = canonical_unit
matched_entity_token = entity
try:
unknown = Unknown(entity=entity, unit=canonical_unit)
except Exception:
return None
try:
return CandidateUnknown(
unknown=unknown,
source_span=question_sentence,
matched_unit_token=matched_unit_token,
matched_entity_token=matched_entity_token,
)
except Exception:
return None
# ---------------------------------------------------------------------------
# Trace-event construction
# ---------------------------------------------------------------------------
def make_admit_trace_event(
slot: QuestionTargetSlot,
canonical_unit: str,
) -> str:
"""Build a JSON-encoded admit trace event for the reader."""
return json.dumps(
{
"layer": "comprehension_reader",
"phase": 1,
"outcome": "admit",
"entity": slot.entity,
"unit": canonical_unit,
"question_form": slot.kind,
},
sort_keys=True,
separators=(",", ":"),
)
def make_fallthrough_trace_event(refusal: ReaderRefusal) -> str:
"""Build a JSON-encoded fallthrough trace event for the reader."""
return json.dumps(
{
"layer": "comprehension_reader",
"phase": 1,
"outcome": "fallthrough_to_regex",
"refusal_reason": refusal.reason,
"refusal_token": refusal.token_text,
},
sort_keys=True,
separators=(",", ":"),
)

124
generate/math_candidate_graph.py
View file

@ -36,7 +36,10 @@ from __future__ import annotations
import re
from dataclasses import dataclass
from itertools import product
from typing import Final, Union
from typing import TYPE_CHECKING, Final, Union
if TYPE_CHECKING:
from core.config import RuntimeConfig
from generate.math_candidate_parser import (
CandidateInitial,
@ -115,6 +118,12 @@ class CandidateGraphResult:
# Diagnostics for inner-loop signal in P6 runner.
branches_enumerated: int
branches_admissible: int
# ADR-0164 Phase 1 — reader trace events (JSON-encoded strings).
# Each element is a JSON object carrying {"layer", "phase", "outcome", ...}.
# Empty tuple when comprehension_reader_questions flag is False (default).
# Deferred: full integration with chat/telemetry.py JSONL sink is a
# follow-up; these events are available for tests and delta-report analysis.
reader_trace: tuple[str, ...] = ()
@property
def is_admitted(self) -> bool:
@ -288,6 +297,69 @@ def _collapse_per_sentence_ties(
return [c for c in choices if _slot_count(c) == max_slots]
# ---------------------------------------------------------------------------
# ADR-0164 Phase 1 — comprehension reader dispatch helper
# ---------------------------------------------------------------------------
def _try_reader_for_question(
question_sentence: str,
per_sentence_choices: list[list[SentenceChoice]],
statement_sentence_count: int,
trace_out: list[str],
) -> list[CandidateUnknown] | None:
"""Invoke the Phase-1 comprehension reader for the question sentence.
Returns a list with one CandidateUnknown on reader admission, or None
when the reader refuses (caller falls through to the regex parser).
Appends a JSON-encoded trace event to ``trace_out`` on every invocation
(admit or fallthrough_to_regex).
This function is the hybrid-dispatch core for ADR-0164 Phase 1. The
fallthrough path (reader refusal → regex) is intentional and must never
raise: the reader is purely additive at Phase 1.
"""
try:
from generate.comprehension.lifecycle_runtime_adapter import (
build_problem_state_from_candidates,
invoke_reader_for_question,
make_admit_trace_event,
make_fallthrough_trace_event,
project_to_candidate_unknown,
)
except ImportError:
return None # adapter not available — fall through silently
# Flatten per_sentence_choices to a single list for state construction.
# Take the first choice per sentence (deterministic: tiebreaker already ran).
flat: list[SentenceChoice] = [choices[0] for choices in per_sentence_choices if choices]
try:
problem_state = build_problem_state_from_candidates(flat, statement_sentence_count)
except Exception:
return None # construction failure → fall through
result = invoke_reader_for_question(question_sentence, problem_state)
if isinstance(result, tuple):
slot, canonical_unit = result
trace_out.append(make_admit_trace_event(slot, canonical_unit))
candidate = project_to_candidate_unknown(
slot, canonical_unit, question_sentence, problem_state
)
if candidate is not None and _question_admissible(candidate):
return [candidate]
# Reader admitted but projection failed or failed admissibility.
# Do NOT fall through to regex (the reader's admission is authoritative
# on what it could parse; if projection fails it's a structural gap,
# not a reason to let the regex guess differently).
return None
else:
# ReaderRefusal — fall through to regex.
from generate.comprehension.state import ReaderRefusal
if isinstance(result, ReaderRefusal):
trace_out.append(make_fallthrough_trace_event(result))
return None
# ---------------------------------------------------------------------------
# Graph construction from one branch
# ---------------------------------------------------------------------------
@ -344,10 +416,21 @@ def _build_graph(
# Orchestrator
# ---------------------------------------------------------------------------
def parse_and_solve(text: str) -> CandidateGraphResult:
def parse_and_solve(
text: str,
config: "RuntimeConfig | None" = None,
) -> CandidateGraphResult:
"""End-to-end: parse text via candidate-graph topology, solve each
admissible branch, apply decision rule.
Args:
text: The problem text to parse.
config: Optional :class:`core.config.RuntimeConfig`. When None,
defaults to flag-OFF behaviour (byte-identical to today).
Pass ``RuntimeConfig(comprehension_reader_questions=True)`` to
activate the ADR-0164 Phase-1 comprehension reader for question
sentences.
Returns :class:`CandidateGraphResult` with either an admitted
``answer`` + ``selected_graph`` or a ``refusal_reason`` string
naming why the problem was refused.
@ -360,6 +443,11 @@ def parse_and_solve(text: str) -> CandidateGraphResult:
filter at the per-sentence level (already applied during
filtering).
- Branches that disagree on the final answer trigger refusal.
- When the comprehension reader is active (flag ON), a reader refusal
falls through to the existing regex parser — the reader is purely
additive. A reader admission that produces wrong > 0 cannot occur
because the same admissibility gate, solver, and verifier run
downstream of the reader as they run today.
"""
if not isinstance(text, str) or not text.strip():
return CandidateGraphResult(
@ -545,7 +633,32 @@ def parse_and_solve(text: str) -> CandidateGraphResult:
)
per_sentence_choices.append(_collapse_per_sentence_ties(choices))
question_choices = _filtered_question_choices(question_sentences[0], text)
# ADR-0164 Phase 1 — comprehension reader hybrid dispatch.
# When comprehension_reader_questions is True, try the reader FIRST.
# On reader admission, use the reader's CandidateUnknown; on refusal,
# fall through to the existing regex question parser (Pattern A/B/C).
# The reader is purely additive: a refusal MUST NOT prevent admission
# by the regex parser.
reader_trace: list[str] = []
reader_question_choices: list[CandidateUnknown] | None = None
_use_reader = (
config is not None and config.comprehension_reader_questions
)
if _use_reader:
reader_question_choices = _try_reader_for_question(
question_sentences[0],
per_sentence_choices,
len(statement_sentences),
reader_trace,
)
# Fall through to the regex parser when the flag is off OR the reader
# refused on the question sentence.
if reader_question_choices is not None:
question_choices = reader_question_choices
else:
question_choices = _filtered_question_choices(question_sentences[0], text)
if not question_choices:
return CandidateGraphResult(
answer=None, selected_graph=None,
@ -554,6 +667,7 @@ def parse_and_solve(text: str) -> CandidateGraphResult:
f"{question_sentences[0]!r}"
),
branches_enumerated=0, branches_admissible=0,
reader_trace=tuple(reader_trace),
)
# Cartesian product across statement choices × question choices.
@ -569,6 +683,7 @@ def parse_and_solve(text: str) -> CandidateGraphResult:
f"{MAX_TOTAL_BRANCHES} (refusing rather than truncating)"
),
branches_enumerated=total, branches_admissible=0,
reader_trace=tuple(reader_trace),
)
admissible: list[CandidateGraphAnswer] = []
@ -593,6 +708,7 @@ def parse_and_solve(text: str) -> CandidateGraphResult:
refusal_reason="no branch produced a solvable graph",
branches_enumerated=branches_enumerated,
branches_admissible=0,
reader_trace=tuple(reader_trace),
)
# Decision rule: all answers identical → emit; otherwise → refuse.
@ -606,6 +722,7 @@ def parse_and_solve(text: str) -> CandidateGraphResult:
),
branches_enumerated=branches_enumerated,
branches_admissible=len(admissible),
reader_trace=tuple(reader_trace),
)
# Single agreed answer. Pick the first admissible graph as the
@ -617,4 +734,5 @@ def parse_and_solve(text: str) -> CandidateGraphResult:
refusal_reason=None,
branches_enumerated=branches_enumerated,
branches_admissible=len(admissible),
reader_trace=tuple(reader_trace),
)

302
tests/test_reader_coexistence.py Normal file
View file

@ -0,0 +1,302 @@
"""ADR-0164 Phase 1 — reader/regex coexistence integration tests.
Verifies:
1. Flag-OFF byte-identity: parse_and_solve without config == parse_and_solve
with comprehension_reader_questions=False on the 3 currently-correct cases.
2. Flag-ON determinism: identical input + flag ON → identical reader_trace,
answer, and graph canonical bytes.
3. wrong=0 invariant: flag ON never produces a wrong outcome on the
50-case train_sample.
4. Trace shape: every reader_trace element is valid JSON with the expected
layer/phase/outcome keys.
5. Brief-8 targets: reader is consulted (non-empty reader_trace) for the
5 GSM8K question sentences referenced in ADR-0164.3 §Worked example.
6. Fallthrough preserved: flag ON on an unknown-word question produces a
fallthrough trace event and the same answer as flag OFF.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
import pytest
from core.config import RuntimeConfig
from generate.math_candidate_graph import parse_and_solve
_CASES_PATH = Path(__file__).resolve().parents[1] / "evals/gsm8k_math/train_sample/v1/cases.jsonl"
def _load_cases() -> list[dict[str, Any]]:
return [json.loads(l) for l in _CASES_PATH.open(encoding="utf-8") if l.strip()]
def _adapt(case: dict[str, Any]) -> dict[str, Any]:
return {
"id": case["case_id"],
"problem": case["question"],
"expected_answer": case["answer_numeric"],
"expected_unit": "",
}
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
_CONFIG_OFF = RuntimeConfig(comprehension_reader_questions=False)
_CONFIG_ON = RuntimeConfig(comprehension_reader_questions=True)
# 3 currently-correct cases (fast-path, reader does not affect them)
_CORRECT_IDS = frozenset({
"gsm8k-train-sample-v1-0014",
"gsm8k-train-sample-v1-0018",
"gsm8k-train-sample-v1-0042",
})
# 5 Brief-8 target question sentences (ADR-0164.3 §Worked example follow-on)
_BRIEF8_IDS = frozenset({
"gsm8k-train-sample-v1-0007", # How many more boxes...
"gsm8k-train-sample-v1-0017", # How much will it cost him?
"gsm8k-train-sample-v1-0027", # How many followers does Malcolm have...
"gsm8k-train-sample-v1-0036", # How much time did she spend studying...
"gsm8k-train-sample-v1-0043", # How much money will she be left with...
})
@pytest.fixture(scope="module")
def all_cases() -> list[dict[str, Any]]:
return _load_cases()
@pytest.fixture(scope="module")
def correct_cases(all_cases: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [c for c in all_cases if c["case_id"] in _CORRECT_IDS]
@pytest.fixture(scope="module")
def brief8_cases(all_cases: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [c for c in all_cases if c["case_id"] in _BRIEF8_IDS]
# ---------------------------------------------------------------------------
# 1. Flag-OFF byte-identity
# ---------------------------------------------------------------------------
class TestFlagOffByteIdentity:
"""parse_and_solve(config=None) and parse_and_solve(config=CONFIG_OFF)
must produce the same answer on the 3 currently-correct cases."""
def test_correct_cases_no_config_vs_off(self, correct_cases: list[dict[str, Any]]) -> None:
assert correct_cases, "fixture must contain at least one correct case"
for case in correct_cases:
text = case["question"]
r_none = parse_and_solve(text, config=None)
r_off = parse_and_solve(text, config=_CONFIG_OFF)
assert r_none.answer == r_off.answer, (
f"{case['case_id']}: answer differs between config=None and flag-OFF"
)
assert r_none.is_admitted == r_off.is_admitted, (
f"{case['case_id']}: is_admitted differs"
)
# Flag OFF must produce empty reader_trace (reader never consulted).
assert r_off.reader_trace == (), (
f"{case['case_id']}: reader_trace must be empty with flag OFF"
)
def test_correct_cases_off_vs_on_same_answer(self, correct_cases: list[dict[str, Any]]) -> None:
"""Fast-path cases are unaffected by the reader — answer must be identical."""
for case in correct_cases:
text = case["question"]
r_off = parse_and_solve(text, config=_CONFIG_OFF)
r_on = parse_and_solve(text, config=_CONFIG_ON)
assert r_off.answer == r_on.answer, (
f"{case['case_id']}: reader flag changed a fast-path answer"
)
assert r_off.is_admitted == r_on.is_admitted, (
f"{case['case_id']}: reader flag changed admission status"
)
# ---------------------------------------------------------------------------
# 2. Determinism (flag ON)
# ---------------------------------------------------------------------------
class TestDeterminism:
def test_flag_on_reader_trace_deterministic(self, all_cases: list[dict[str, Any]]) -> None:
"""Two calls with the same input and flag ON must produce identical reader_trace."""
sample = all_cases[:10]
for case in sample:
text = case["question"]
r1 = parse_and_solve(text, config=_CONFIG_ON)
r2 = parse_and_solve(text, config=_CONFIG_ON)
assert r1.reader_trace == r2.reader_trace, (
f"{case['case_id']}: reader_trace not deterministic"
)
assert r1.answer == r2.answer, (
f"{case['case_id']}: answer not deterministic"
)
def test_flag_off_reader_trace_empty(self, all_cases: list[dict[str, Any]]) -> None:
"""Flag OFF must never populate reader_trace."""
for case in all_cases:
r = parse_and_solve(case["question"], config=_CONFIG_OFF)
assert r.reader_trace == (), (
f"{case['case_id']}: reader_trace must be empty with flag OFF"
)
# ---------------------------------------------------------------------------
# 3. wrong=0 invariant
# ---------------------------------------------------------------------------
class TestWrongIsZero:
def test_flag_on_wrong_is_zero(self, all_cases: list[dict[str, Any]]) -> None:
"""Flag ON must never produce wrong > 0 on the 50-case train sample.
Wrong outcome requires: admitted=True AND answer != expected. Since
the reader is additive (refusal falls through to regex), and the
underlying regex path is already wrong=0, this invariant must hold.
"""
wrong_cases: list[str] = []
for raw in all_cases:
result = parse_and_solve(raw["question"], config=_CONFIG_ON)
if result.is_admitted and result.answer != raw["answer_numeric"]:
wrong_cases.append(
f"{raw['case_id']}: got {result.answer}, expected {raw['answer_numeric']}"
)
assert not wrong_cases, f"wrong > 0 with flag ON:\n" + "\n".join(wrong_cases)
# ---------------------------------------------------------------------------
# 4. Trace event shape
# ---------------------------------------------------------------------------
_REQUIRED_TRACE_KEYS = frozenset({"layer", "phase", "outcome"})
_VALID_OUTCOMES = frozenset({"admit", "fallthrough_to_regex"})
class TestTraceShape:
def test_trace_events_are_valid_json(self, all_cases: list[dict[str, Any]]) -> None:
for case in all_cases:
r = parse_and_solve(case["question"], config=_CONFIG_ON)
for event_str in r.reader_trace:
try:
event = json.loads(event_str)
except json.JSONDecodeError:
pytest.fail(f"{case['case_id']}: reader_trace event is not valid JSON: {event_str!r}")
missing = _REQUIRED_TRACE_KEYS - set(event.keys())
assert not missing, (
f"{case['case_id']}: trace event missing keys {missing}: {event}"
)
assert event["layer"] == "comprehension_reader", (
f"{case['case_id']}: unexpected layer: {event}"
)
assert event["phase"] == 1, (
f"{case['case_id']}: unexpected phase: {event}"
)
assert event["outcome"] in _VALID_OUTCOMES, (
f"{case['case_id']}: unexpected outcome: {event}"
)
def test_admit_event_has_entity_and_unit(self, all_cases: list[dict[str, Any]]) -> None:
"""Every 'admit' trace event must carry entity and unit keys."""
for case in all_cases:
r = parse_and_solve(case["question"], config=_CONFIG_ON)
for event_str in r.reader_trace:
event = json.loads(event_str)
if event["outcome"] == "admit":
assert "entity" in event, (
f"{case['case_id']}: admit event missing 'entity': {event}"
)
assert "unit" in event, (
f"{case['case_id']}: admit event missing 'unit': {event}"
)
def test_fallthrough_event_has_refusal_reason(self, all_cases: list[dict[str, Any]]) -> None:
"""Every 'fallthrough_to_regex' trace event must carry refusal_reason."""
for case in all_cases:
r = parse_and_solve(case["question"], config=_CONFIG_ON)
for event_str in r.reader_trace:
event = json.loads(event_str)
if event["outcome"] == "fallthrough_to_regex":
assert "refusal_reason" in event, (
f"{case['case_id']}: fallthrough event missing 'refusal_reason': {event}"
)
# ---------------------------------------------------------------------------
# 5. Brief-8 targets: reader is consulted
# ---------------------------------------------------------------------------
class TestBrief8Targets:
def test_reader_consulted_for_brief8_cases(self, brief8_cases: list[dict[str, Any]]) -> None:
"""When flag ON, the reader is consulted for each of the 5 Brief-8 target
question sentences — reader_trace is non-empty."""
assert len(brief8_cases) == 5, (
f"Expected 5 Brief-8 cases, found {len(brief8_cases)}"
)
for case in brief8_cases:
r = parse_and_solve(case["question"], config=_CONFIG_ON)
assert r.reader_trace, (
f"{case['case_id']}: reader_trace is empty — reader was not consulted"
)
def test_brief8_cases_wrong_stays_zero(self, brief8_cases: list[dict[str, Any]]) -> None:
"""Brief-8 cases must not produce wrong outcomes with flag ON."""
for case in brief8_cases:
r = parse_and_solve(case["question"], config=_CONFIG_ON)
if r.is_admitted:
assert r.answer == case["answer_numeric"], (
f"{case['case_id']}: wrong answer with flag ON: "
f"got {r.answer}, expected {case['answer_numeric']}"
)
def test_case_0027_malcolm_admits(self, brief8_cases: list[dict[str, Any]]) -> None:
"""Case 0027 (Malcolm/followers) has no pronoun ambiguity — reader admits it."""
case = next(c for c in brief8_cases if "0027" in c["case_id"])
r = parse_and_solve(case["question"], config=_CONFIG_ON)
assert r.reader_trace, "reader must produce a trace for case 0027"
events = [json.loads(e) for e in r.reader_trace]
admit_events = [e for e in events if e["outcome"] == "admit"]
assert admit_events, (
f"case 0027 must produce at least one admit event; got: {events}"
)
admit = admit_events[0]
assert admit["entity"] == "malcolm"
assert admit["unit"] == "followers"
# ---------------------------------------------------------------------------
# 6. Fallthrough preserved for unknown words
# ---------------------------------------------------------------------------
class TestFallthroughPreserved:
def test_unknown_unit_falls_through_to_regex(self) -> None:
"""A question with an unknown unit noun falls through to regex — result is correct."""
problem = "Martha has 5 apples. How many apples does Martha have?"
r_off = parse_and_solve(problem, config=_CONFIG_OFF)
r_on = parse_and_solve(problem, config=_CONFIG_ON)
# answer must be identical between flag OFF and flag ON
assert r_off.answer == r_on.answer
assert r_off.is_admitted == r_on.is_admitted
# flag ON must record a fallthrough trace event
assert r_on.reader_trace, "fallthrough case must produce a trace event"
event = json.loads(r_on.reader_trace[0])
assert event["outcome"] == "fallthrough_to_regex"
assert event["refusal_reason"] == "unknown_word"
def test_flag_off_no_trace_for_fallthrough_case(self) -> None:
"""Flag OFF must never produce any trace events, even for fallthrough-prone inputs."""
problem = "Martha has 5 apples. How many apples does Martha have?"
r = parse_and_solve(problem, config=_CONFIG_OFF)
assert r.reader_trace == ()