feat(adr-0174-phase2): continuous constraint propagation in comprehension reader
ADR-0174 Phase 2 — hoist _initial_admissible / roundtrip_admissible into hypothesis-based constraint checks with structured elimination tracing. Admission semantics are byte-equivalent to today; the change is structural. Adds generate/comprehension/constraint_propagation.py: - VALID_PREDICATE_NAMES: closed set of 17 sub-check names spanning initial / composed_initial / operation admissibility predicates. Adding new names requires an ADR amendment (structural contract with reader_trace consumer). - ConstraintResult dataclass: admitted bool + predicates_run trace + elimination_reason. Validates admitted-vs-reason consistency. - Elimination dataclass: confidence_rank + predicate + reason for one hypothesis being eliminated. Serialisable as a reader_trace event. - hypothesis_from_initial / hypothesis_from_operation: adapters wrapping CandidateInitial / CandidateOperation as Phase-1 Hypothesis objects with caller-supplied confidence_rank. - _check_initial / _check_composed_initial / _check_operation: decomposed sub-check implementations of the existing admissibility predicates with first-failure short-circuit (matches current semantics). Each sub-check populates predicates_run with (name, ok| fail|skip) so the consumer sees exactly which predicate decided. - check_constraints: dispatches on candidate type. - eliminate_violating: bulk filter; returns (survivors, eliminations); survivors are re-densified to satisfy ProblemReadingState's open_hypotheses post_init invariant (dense-from-0 ranks); eliminations carry the original confidence_rank for trace fidelity. Wires into generate/math_candidate_graph.py at the recognizer injection site (line 825+): replaces inline _initial_admissible / roundtrip_admissible dispatch with eliminate_violating. Elimination events become JSON entries in reader_trace with layer= 'constraint_propagation', phase=2, predicate, reason, sentence_index. Phase 2 acceptance verified: - 24/24 ADR-0174 Phase 2 tests pass (emission, parity with existing predicates on 9 admit/reject cases, redensification, dataclass invariants, integration). - 71/71 existing reader + Phase 1 tests still pass. - Smoke 67/67, packs 141/141, lanes 8/8. - train_sample/v1 byte-identical across two runs with use_reader=True. - Score preserved: correct=3 refused=47 wrong=0 — semantics identical because the decomposed sub-checks short-circuit on the same predicates the inline checks would have caught. Trace-event behavior: today's injectors are conservative enough that zero eliminations occur on train_sample/v1 (no false positives, no mid-pipeline failures). The wiring is exercised by test_phase2_event_shape_when_synthesized which proves the trace shape on a synthetic CandidateInitial that fails initial.unit_grounds. When Phase 3 begins emitting partial hypotheses from apply_word, the elimination path will fire on real candidates and the trace will populate. Stacks on Phase 1 (feat/adr-0174-phase1-held-hypothesis-state, PR #416). Merges cleanly into main after PR #416 lands.
This commit is contained in:
Shay
parent
7a09b70a5e
commit
3357c5fc71
3 changed files with 1217 additions and 18 deletions
726
generate/comprehension/constraint_propagation.py
Normal file
726
generate/comprehension/constraint_propagation.py
Normal file
|
|
@ -0,0 +1,726 @@
|
|||
"""ADR-0174 Phase 2 — continuous constraint propagation.
|
||||
|
||||
Hoists the candidate-graph layer's admissibility predicates
|
||||
(``_initial_admissible``, ``roundtrip_admissible``) into per-hypothesis
|
||||
constraint checks that fire during reading rather than only at the end
|
||||
of :func:`generate.math_candidate_graph.parse_and_solve`.
|
||||
|
||||
Phase 2 scope (this module):
|
||||
|
||||
- ``hypothesis_from_initial`` / ``hypothesis_from_operation`` —
|
||||
adapters that wrap an existing :class:`CandidateInitial` /
|
||||
:class:`CandidateOperation` as a Phase-1 :class:`Hypothesis` ready
|
||||
to flow through ``ProblemReadingState.open_hypotheses``.
|
||||
- ``check_constraints`` — runs the same admissibility predicates the
|
||||
candidate-graph layer runs today, but returns a structured
|
||||
:class:`ConstraintResult` carrying the specific elimination reason
|
||||
instead of a bare bool. Sub-checks are decomposed so a Phase-3
|
||||
partial hypothesis can run only the predicates whose slots are
|
||||
populated.
|
||||
- ``eliminate_violating`` — applies ``check_constraints`` to a tuple
|
||||
of hypotheses, returns ``(surviving, eliminations)``. An
|
||||
elimination record carries the hypothesis id, the predicate that
|
||||
fired, and the reason — designed to serialise into a
|
||||
``reader_trace`` event.
|
||||
|
||||
Phase 2 does NOT change admission semantics. A candidate that passes
|
||||
``check_constraints`` here is byte-equivalent to one that passes
|
||||
``_initial_admissible`` / ``roundtrip_admissible`` at the
|
||||
candidate-graph layer today. The change is structural: the constraint
|
||||
check is now hypothesis-based, the elimination is structured, and the
|
||||
trace is visible. Phase 3 will populate hypotheses from partial reads
|
||||
(``apply_word`` mid-sentence); Phase 4 will wire in-loop contemplation
|
||||
to resolve ambiguities the constraint check leaves with multiple
|
||||
survivors.
|
||||
|
||||
Trust boundary: this module is read-only over the existing predicates.
|
||||
It does not weaken any admissibility check. The ``wrong = 0``
|
||||
invariant is preserved by construction — every surviving hypothesis has
|
||||
passed exactly the same predicate sub-checks that admit candidates
|
||||
today.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Final, Literal, cast
|
||||
|
||||
from generate.comprehension.state import (
|
||||
ComprehensionStateError,
|
||||
HYPOTHESIS_CAP,
|
||||
Hypothesis,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Constraint-result types
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
# Closed set of predicate names that may appear in a constraint result.
|
||||
# Adding a new predicate requires an ADR amendment (the predicate names
|
||||
# are a structural contract with the reader_trace consumer).
|
||||
VALID_PREDICATE_NAMES: Final[frozenset[str]] = frozenset(
|
||||
{
|
||||
# _initial_admissible sub-checks
|
||||
"initial.anchor_grounds",
|
||||
"initial.value_grounds",
|
||||
"initial.unit_grounds",
|
||||
"initial.entity_grounds",
|
||||
# _composed_initial_admissible sub-checks (RAT-1)
|
||||
"composed_initial.evidence_complete",
|
||||
"composed_initial.input_tokens_ground",
|
||||
"composed_initial.currency_symbol_present",
|
||||
"composed_initial.entity_token_present",
|
||||
# roundtrip_admissible sub-checks
|
||||
"operation.verb_registered",
|
||||
"operation.verb_grounds",
|
||||
"operation.actor_grounds",
|
||||
"operation.value_grounds",
|
||||
"operation.unit_grounds",
|
||||
"operation.target_grounds",
|
||||
"operation.reference_actor_grounds",
|
||||
"operation.operand_shape_consistent",
|
||||
"operation.rate_denominator_grounds",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ConstraintResult:
|
||||
"""Outcome of running constraints against one hypothesis.
|
||||
|
||||
Fields:
|
||||
admitted: True iff every applicable sub-check passed.
|
||||
predicates_run: Tuple of (predicate_name, outcome) for every
|
||||
sub-check that fired. Sub-checks whose slots were
|
||||
unpopulated are not included (Phase-3 conservative
|
||||
in-flight behavior; Phase-2 candidates are complete
|
||||
so every applicable predicate fires).
|
||||
elimination_reason: Non-None iff admitted=False; the first
|
||||
predicate that failed (sub-checks short-circuit on
|
||||
first failure to preserve current behavior).
|
||||
"""
|
||||
|
||||
admitted: bool
|
||||
predicates_run: tuple[tuple[str, Literal["ok", "fail", "skip"]], ...]
|
||||
elimination_reason: str | None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not isinstance(self.admitted, bool):
|
||||
raise ComprehensionStateError(
|
||||
f"ConstraintResult.admitted must be bool; got "
|
||||
f"{type(self.admitted).__name__}"
|
||||
)
|
||||
if not isinstance(self.predicates_run, tuple):
|
||||
raise ComprehensionStateError(
|
||||
"ConstraintResult.predicates_run must be tuple"
|
||||
)
|
||||
for idx, entry in enumerate(self.predicates_run):
|
||||
if not (
|
||||
isinstance(entry, tuple)
|
||||
and len(entry) == 2
|
||||
and isinstance(entry[0], str)
|
||||
and entry[0] in VALID_PREDICATE_NAMES
|
||||
and entry[1] in ("ok", "fail", "skip")
|
||||
):
|
||||
raise ComprehensionStateError(
|
||||
f"ConstraintResult.predicates_run[{idx}] must be "
|
||||
"(predicate_name in VALID_PREDICATE_NAMES, outcome in "
|
||||
f"{{ok, fail, skip}}); got {entry!r}"
|
||||
)
|
||||
if self.admitted and self.elimination_reason is not None:
|
||||
raise ComprehensionStateError(
|
||||
"ConstraintResult.admitted=True is inconsistent with "
|
||||
f"non-None elimination_reason={self.elimination_reason!r}"
|
||||
)
|
||||
if not self.admitted and self.elimination_reason is None:
|
||||
raise ComprehensionStateError(
|
||||
"ConstraintResult.admitted=False requires a non-None "
|
||||
"elimination_reason"
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Elimination:
|
||||
"""Structured record of one hypothesis being eliminated.
|
||||
|
||||
Designed to serialise as a JSON object in ``reader_trace``.
|
||||
"""
|
||||
|
||||
confidence_rank: int
|
||||
predicate: str
|
||||
reason: str
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not isinstance(self.confidence_rank, int) or isinstance(
|
||||
self.confidence_rank, bool
|
||||
):
|
||||
raise ComprehensionStateError(
|
||||
"Elimination.confidence_rank must be int"
|
||||
)
|
||||
if self.predicate not in VALID_PREDICATE_NAMES:
|
||||
raise ComprehensionStateError(
|
||||
f"Elimination.predicate must be in VALID_PREDICATE_NAMES; "
|
||||
f"got {self.predicate!r}"
|
||||
)
|
||||
if not isinstance(self.reason, str) or not self.reason:
|
||||
raise ComprehensionStateError(
|
||||
"Elimination.reason must be non-empty str"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Hypothesis emitters — adapt existing candidate types
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def hypothesis_from_initial(candidate: object, rank: int) -> Hypothesis:
|
||||
"""Wrap a :class:`CandidateInitial` as a Phase-1 :class:`Hypothesis`.
|
||||
|
||||
The candidate's per-slot tokens are not unpacked into
|
||||
``category_assignments`` here — that wiring is Phase 3 work when
|
||||
apply_word starts threading category assignments through the reader.
|
||||
Phase 2 carries the candidate intact so downstream solver / verifier
|
||||
paths consume it unchanged.
|
||||
|
||||
The ``unresolved`` tuple is empty: Phase 2 hypotheses are complete
|
||||
candidates produced by injectors, not partial reads. Phase 3 will
|
||||
populate this with the slot names a partial hypothesis still needs.
|
||||
"""
|
||||
if rank < 0 or rank >= HYPOTHESIS_CAP:
|
||||
raise ComprehensionStateError(
|
||||
f"hypothesis_from_initial: rank must be in [0, "
|
||||
f"{HYPOTHESIS_CAP}); got {rank}"
|
||||
)
|
||||
return Hypothesis(
|
||||
candidate=candidate,
|
||||
category_assignments=(),
|
||||
constraint_state=(),
|
||||
confidence_rank=rank,
|
||||
unresolved=(),
|
||||
)
|
||||
|
||||
|
||||
def hypothesis_from_operation(candidate: object, rank: int) -> Hypothesis:
|
||||
"""Wrap a :class:`CandidateOperation` as a Phase-1 :class:`Hypothesis`.
|
||||
|
||||
See :func:`hypothesis_from_initial` for the structural notes;
|
||||
behaviorally identical, kept as a separate function so the call site
|
||||
documents intent and Phase 3 can specialise per type without
|
||||
rewriting the caller.
|
||||
"""
|
||||
if rank < 0 or rank >= HYPOTHESIS_CAP:
|
||||
raise ComprehensionStateError(
|
||||
f"hypothesis_from_operation: rank must be in [0, "
|
||||
f"{HYPOTHESIS_CAP}); got {rank}"
|
||||
)
|
||||
return Hypothesis(
|
||||
candidate=candidate,
|
||||
category_assignments=(),
|
||||
constraint_state=(),
|
||||
confidence_rank=rank,
|
||||
unresolved=(),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Constraint checks — decomposed sub-checks per predicate
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _check_initial(candidate: object) -> ConstraintResult:
|
||||
"""Run :func:`_initial_admissible` as decomposed sub-checks.
|
||||
|
||||
Returns a :class:`ConstraintResult` carrying the specific predicate
|
||||
that failed (first-failure short-circuit, matching today's
|
||||
behavior). When ``composition_evidence`` is non-None the candidate
|
||||
is a registry-gated composition and routes to
|
||||
:func:`_check_composed_initial` instead, mirroring the existing
|
||||
dispatch in ``_initial_admissible``.
|
||||
"""
|
||||
# Lazy imports to avoid circular dependency on math_candidate_graph
|
||||
# → math_roundtrip → here.
|
||||
from generate.math_roundtrip import (
|
||||
_tokens, _token_in, _value_grounds, _unit_grounds,
|
||||
)
|
||||
|
||||
ic = candidate
|
||||
composition_evidence = getattr(ic, "composition_evidence", None)
|
||||
if composition_evidence is not None:
|
||||
return _check_composed_initial(ic)
|
||||
|
||||
matched_anchor = getattr(ic, "matched_anchor", None)
|
||||
matched_value_token = getattr(ic, "matched_value_token", None)
|
||||
matched_unit_token = getattr(ic, "matched_unit_token", None)
|
||||
matched_entity_token = getattr(ic, "matched_entity_token", None)
|
||||
source_span = getattr(ic, "source_span", None)
|
||||
if not all(
|
||||
isinstance(x, str) for x in
|
||||
(matched_anchor, matched_value_token, matched_unit_token,
|
||||
matched_entity_token, source_span)
|
||||
):
|
||||
# Defensive — the candidate does not have the expected shape.
|
||||
# Treat as failed under a synthetic predicate that the trace
|
||||
# consumer can recognise.
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=(("initial.anchor_grounds", "fail"),),
|
||||
elimination_reason="candidate does not expose initial-shape slots",
|
||||
)
|
||||
|
||||
# All five fields are confirmed str by the guard above.
|
||||
matched_anchor = cast(str, matched_anchor)
|
||||
matched_value_token = cast(str, matched_value_token)
|
||||
matched_unit_token = cast(str, matched_unit_token)
|
||||
matched_entity_token = cast(str, matched_entity_token)
|
||||
source_span = cast(str, source_span)
|
||||
haystack = _tokens(source_span)
|
||||
run: list[tuple[str, Literal["ok", "fail", "skip"]]] = []
|
||||
|
||||
if not _token_in(matched_anchor, haystack):
|
||||
run.append(("initial.anchor_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"matched_anchor {matched_anchor!r} does not appear in "
|
||||
f"source tokens"
|
||||
),
|
||||
)
|
||||
run.append(("initial.anchor_grounds", "ok"))
|
||||
|
||||
if not _value_grounds(matched_value_token, haystack):
|
||||
run.append(("initial.value_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"matched_value_token {matched_value_token!r} does not "
|
||||
f"ground in source"
|
||||
),
|
||||
)
|
||||
run.append(("initial.value_grounds", "ok"))
|
||||
|
||||
if not _unit_grounds(matched_unit_token, source_span, haystack):
|
||||
run.append(("initial.unit_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"matched_unit_token {matched_unit_token!r} does not "
|
||||
f"ground in source"
|
||||
),
|
||||
)
|
||||
run.append(("initial.unit_grounds", "ok"))
|
||||
|
||||
# Multi-word entity: every word must ground (mirrors existing logic).
|
||||
for tok in matched_entity_token.split():
|
||||
if not _token_in(tok, haystack):
|
||||
run.append(("initial.entity_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"matched_entity_token component {tok!r} does not "
|
||||
f"appear in source tokens"
|
||||
),
|
||||
)
|
||||
run.append(("initial.entity_grounds", "ok"))
|
||||
|
||||
return ConstraintResult(
|
||||
admitted=True,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=None,
|
||||
)
|
||||
|
||||
|
||||
def _check_composed_initial(candidate: object) -> ConstraintResult:
|
||||
"""Decomposed version of :func:`_composed_initial_admissible`.
|
||||
|
||||
Verifies composition_evidence schema completeness, then that each
|
||||
input token grounds, optional currency symbol presence, and the
|
||||
matched_entity_token is populated. Matches the existing
|
||||
short-circuit-on-first-failure semantics.
|
||||
"""
|
||||
from generate.math_roundtrip import _tokens, _token_in
|
||||
|
||||
ev = getattr(candidate, "composition_evidence", None)
|
||||
if not ev:
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=(("composed_initial.evidence_complete", "fail"),),
|
||||
elimination_reason="composition_evidence is empty",
|
||||
)
|
||||
|
||||
required = {"composition_shape", "input_tokens", "entity_source"}
|
||||
if not required.issubset(ev.keys()):
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=(("composed_initial.evidence_complete", "fail"),),
|
||||
elimination_reason=(
|
||||
f"composition_evidence missing required keys: "
|
||||
f"{sorted(required - set(ev.keys()))}"
|
||||
),
|
||||
)
|
||||
|
||||
run: list[tuple[str, Literal["ok", "fail", "skip"]]] = [
|
||||
("composed_initial.evidence_complete", "ok")
|
||||
]
|
||||
|
||||
source_span = getattr(candidate, "source_span", "") or ""
|
||||
haystack = _tokens(source_span)
|
||||
input_tokens_field = ev["input_tokens"]
|
||||
input_tokens: list[str] = (
|
||||
str(input_tokens_field).split("|") if input_tokens_field else []
|
||||
)
|
||||
if not input_tokens:
|
||||
run.append(("composed_initial.input_tokens_ground", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason="composition_evidence.input_tokens is empty",
|
||||
)
|
||||
for tok in input_tokens:
|
||||
if not _token_in(tok, haystack):
|
||||
run.append(("composed_initial.input_tokens_ground", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"composition input token {tok!r} does not ground "
|
||||
f"in source"
|
||||
),
|
||||
)
|
||||
run.append(("composed_initial.input_tokens_ground", "ok"))
|
||||
|
||||
currency_symbol = ev.get("currency_symbol")
|
||||
if currency_symbol:
|
||||
if currency_symbol not in source_span:
|
||||
run.append(("composed_initial.currency_symbol_present", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"composition currency_symbol {currency_symbol!r} "
|
||||
f"not present in source"
|
||||
),
|
||||
)
|
||||
run.append(("composed_initial.currency_symbol_present", "ok"))
|
||||
else:
|
||||
run.append(("composed_initial.currency_symbol_present", "skip"))
|
||||
|
||||
matched_entity_token = getattr(candidate, "matched_entity_token", "")
|
||||
if not matched_entity_token or not matched_entity_token.strip():
|
||||
run.append(("composed_initial.entity_token_present", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason="composition matched_entity_token is empty",
|
||||
)
|
||||
run.append(("composed_initial.entity_token_present", "ok"))
|
||||
|
||||
return ConstraintResult(
|
||||
admitted=True,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=None,
|
||||
)
|
||||
|
||||
|
||||
def _check_operation(candidate: object) -> ConstraintResult:
|
||||
"""Run :func:`roundtrip_admissible` as decomposed sub-checks.
|
||||
|
||||
Mirrors the existing short-circuit-on-first-failure semantics. Each
|
||||
sub-check populates the predicates_run trace so the eliminator can
|
||||
record exactly which predicate the candidate failed.
|
||||
"""
|
||||
from generate.math_problem_graph import Comparison, Quantity, Rate
|
||||
from generate.math_roundtrip import (
|
||||
KIND_TO_VERBS,
|
||||
_tokens, _token_in, _value_grounds, _unit_grounds,
|
||||
)
|
||||
|
||||
op = getattr(candidate, "op", None)
|
||||
if op is None:
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=(("operation.verb_registered", "fail"),),
|
||||
elimination_reason="candidate.op is None",
|
||||
)
|
||||
|
||||
matched_verb = getattr(candidate, "matched_verb", "")
|
||||
source_span = getattr(candidate, "source_span", "")
|
||||
haystack = _tokens(source_span)
|
||||
|
||||
run: list[tuple[str, Literal["ok", "fail", "skip"]]] = []
|
||||
|
||||
valid_verbs = KIND_TO_VERBS.get(op.kind)
|
||||
if valid_verbs is None or matched_verb.lower() not in valid_verbs:
|
||||
run.append(("operation.verb_registered", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"matched_verb {matched_verb!r} not registered for op.kind "
|
||||
f"{op.kind!r}"
|
||||
),
|
||||
)
|
||||
run.append(("operation.verb_registered", "ok"))
|
||||
|
||||
if not _token_in(matched_verb, haystack):
|
||||
run.append(("operation.verb_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"matched_verb {matched_verb!r} does not appear in source"
|
||||
),
|
||||
)
|
||||
run.append(("operation.verb_grounds", "ok"))
|
||||
|
||||
matched_actor_token = getattr(candidate, "matched_actor_token", "")
|
||||
if not _token_in(matched_actor_token, haystack):
|
||||
run.append(("operation.actor_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"matched_actor_token {matched_actor_token!r} does not "
|
||||
f"appear in source"
|
||||
),
|
||||
)
|
||||
run.append(("operation.actor_grounds", "ok"))
|
||||
|
||||
matched_value_token = getattr(candidate, "matched_value_token", "")
|
||||
if op.kind == "compare_multiplicative" and matched_value_token == matched_verb:
|
||||
run.append(("operation.value_grounds", "skip"))
|
||||
elif not _value_grounds(matched_value_token, haystack):
|
||||
run.append(("operation.value_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"matched_value_token {matched_value_token!r} does not "
|
||||
f"ground in source"
|
||||
),
|
||||
)
|
||||
else:
|
||||
run.append(("operation.value_grounds", "ok"))
|
||||
|
||||
matched_unit_token = getattr(candidate, "matched_unit_token", "")
|
||||
if matched_unit_token:
|
||||
if not _unit_grounds(matched_unit_token, source_span, haystack):
|
||||
run.append(("operation.unit_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"matched_unit_token {matched_unit_token!r} does not "
|
||||
f"ground in source"
|
||||
),
|
||||
)
|
||||
run.append(("operation.unit_grounds", "ok"))
|
||||
else:
|
||||
if not isinstance(op.operand, Comparison):
|
||||
run.append(("operation.unit_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
"matched_unit_token is empty but operand is not a "
|
||||
"Comparison (only comparisons may omit unit)"
|
||||
),
|
||||
)
|
||||
run.append(("operation.unit_grounds", "skip"))
|
||||
|
||||
matched_target_token = getattr(candidate, "matched_target_token", None)
|
||||
if matched_target_token is not None:
|
||||
if not _token_in(matched_target_token, haystack):
|
||||
run.append(("operation.target_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"matched_target_token {matched_target_token!r} does "
|
||||
f"not appear in source"
|
||||
),
|
||||
)
|
||||
run.append(("operation.target_grounds", "ok"))
|
||||
else:
|
||||
run.append(("operation.target_grounds", "skip"))
|
||||
|
||||
matched_reference_actor_token = getattr(
|
||||
candidate, "matched_reference_actor_token", None
|
||||
)
|
||||
if matched_reference_actor_token is not None:
|
||||
if not _token_in(matched_reference_actor_token, haystack):
|
||||
run.append(("operation.reference_actor_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"matched_reference_actor_token "
|
||||
f"{matched_reference_actor_token!r} does not appear "
|
||||
f"in source"
|
||||
),
|
||||
)
|
||||
run.append(("operation.reference_actor_grounds", "ok"))
|
||||
else:
|
||||
run.append(("operation.reference_actor_grounds", "skip"))
|
||||
|
||||
# Operand shape consistency (mirrors roundtrip_admissible step 8).
|
||||
if op.kind == "apply_rate":
|
||||
if not isinstance(op.operand, Rate):
|
||||
run.append(("operation.operand_shape_consistent", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
"op.kind='apply_rate' requires Rate operand; got "
|
||||
f"{type(op.operand).__name__}"
|
||||
),
|
||||
)
|
||||
run.append(("operation.operand_shape_consistent", "ok"))
|
||||
if not _token_in(op.operand.denominator_unit, haystack):
|
||||
run.append(("operation.rate_denominator_grounds", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"Rate.denominator_unit "
|
||||
f"{op.operand.denominator_unit!r} does not ground"
|
||||
),
|
||||
)
|
||||
run.append(("operation.rate_denominator_grounds", "ok"))
|
||||
elif op.kind in ("compare_additive", "compare_multiplicative"):
|
||||
if not isinstance(op.operand, Comparison):
|
||||
run.append(("operation.operand_shape_consistent", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"op.kind={op.kind!r} requires Comparison operand; got "
|
||||
f"{type(op.operand).__name__}"
|
||||
),
|
||||
)
|
||||
run.append(("operation.operand_shape_consistent", "ok"))
|
||||
else:
|
||||
if not isinstance(op.operand, Quantity):
|
||||
run.append(("operation.operand_shape_consistent", "fail"))
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=(
|
||||
f"op.kind={op.kind!r} requires Quantity operand; got "
|
||||
f"{type(op.operand).__name__}"
|
||||
),
|
||||
)
|
||||
run.append(("operation.operand_shape_consistent", "ok"))
|
||||
|
||||
return ConstraintResult(
|
||||
admitted=True,
|
||||
predicates_run=tuple(run),
|
||||
elimination_reason=None,
|
||||
)
|
||||
|
||||
|
||||
def check_constraints(hypothesis: Hypothesis) -> ConstraintResult:
|
||||
"""Run the appropriate admissibility predicate on a hypothesis.
|
||||
|
||||
Dispatches on the candidate type:
|
||||
- :class:`CandidateInitial` → :func:`_check_initial` (which itself
|
||||
dispatches to :func:`_check_composed_initial` when
|
||||
``composition_evidence`` is non-None).
|
||||
- :class:`CandidateOperation` → :func:`_check_operation`.
|
||||
- Other types refuse cleanly — Phase 2 only knows the two
|
||||
existing candidate types.
|
||||
"""
|
||||
from generate.math_candidate_parser import CandidateInitial
|
||||
from generate.math_roundtrip import CandidateOperation
|
||||
|
||||
candidate = hypothesis.candidate
|
||||
if isinstance(candidate, CandidateInitial):
|
||||
return _check_initial(candidate)
|
||||
if isinstance(candidate, CandidateOperation):
|
||||
return _check_operation(candidate)
|
||||
return ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=(("initial.anchor_grounds", "fail"),),
|
||||
elimination_reason=(
|
||||
f"unknown candidate type {type(candidate).__name__!r}; "
|
||||
"Phase 2 supports CandidateInitial and CandidateOperation only"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Bulk elimination
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def eliminate_violating(
|
||||
hypotheses: tuple[Hypothesis, ...],
|
||||
) -> tuple[tuple[Hypothesis, ...], tuple[Elimination, ...]]:
|
||||
"""Apply :func:`check_constraints` to each hypothesis.
|
||||
|
||||
Returns ``(survivors, eliminations)``. Survivors preserve their
|
||||
original :attr:`Hypothesis.confidence_rank` values **except** that
|
||||
the surviving set is re-densified — if hypothesis 0 is eliminated
|
||||
and hypothesis 1 survives, the survivor's rank stays at 1 in the
|
||||
elimination record but the returned tuple is renumbered to be
|
||||
dense-from-zero so :class:`ProblemReadingState` accepts it.
|
||||
|
||||
Eliminations carry the ORIGINAL confidence_rank so the trace event
|
||||
points at the right candidate.
|
||||
"""
|
||||
surviving_pairs: list[tuple[int, Hypothesis]] = []
|
||||
eliminations: list[Elimination] = []
|
||||
for hyp in hypotheses:
|
||||
result = check_constraints(hyp)
|
||||
if result.admitted:
|
||||
surviving_pairs.append((hyp.confidence_rank, hyp))
|
||||
else:
|
||||
# elimination_reason is non-None when admitted=False (post_init
|
||||
# invariant); pick the first failing predicate.
|
||||
failing = next(
|
||||
(name for name, outcome in result.predicates_run
|
||||
if outcome == "fail"),
|
||||
"initial.anchor_grounds",
|
||||
)
|
||||
eliminations.append(
|
||||
Elimination(
|
||||
confidence_rank=hyp.confidence_rank,
|
||||
predicate=failing,
|
||||
reason=result.elimination_reason or "unspecified",
|
||||
)
|
||||
)
|
||||
|
||||
# Re-densify ranks so the survivors satisfy the
|
||||
# ProblemReadingState.open_hypotheses post_init invariant.
|
||||
densified: list[Hypothesis] = []
|
||||
surviving_pairs.sort(key=lambda x: x[0])
|
||||
for new_rank, (_, hyp) in enumerate(surviving_pairs):
|
||||
if new_rank == hyp.confidence_rank:
|
||||
densified.append(hyp)
|
||||
else:
|
||||
densified.append(
|
||||
Hypothesis(
|
||||
candidate=hyp.candidate,
|
||||
category_assignments=hyp.category_assignments,
|
||||
constraint_state=hyp.constraint_state,
|
||||
confidence_rank=new_rank,
|
||||
unresolved=hyp.unresolved,
|
||||
)
|
||||
)
|
||||
return tuple(densified), tuple(eliminations)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"VALID_PREDICATE_NAMES",
|
||||
"ConstraintResult",
|
||||
"Elimination",
|
||||
"check_constraints",
|
||||
"eliminate_violating",
|
||||
"hypothesis_from_initial",
|
||||
"hypothesis_from_operation",
|
||||
]
|
||||
|
|
@ -33,6 +33,7 @@ decision rule above):
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from itertools import product
|
||||
|
|
@ -41,6 +42,7 @@ from typing import TYPE_CHECKING, Final, Union
|
|||
if TYPE_CHECKING:
|
||||
from core.config import RuntimeConfig
|
||||
|
||||
from generate.comprehension.state import Hypothesis
|
||||
from generate.math_candidate_parser import (
|
||||
CandidateInitial,
|
||||
CandidateUnknown,
|
||||
|
|
@ -792,7 +794,12 @@ def parse_and_solve(
|
|||
# statement's subject is not yet trusted when matching that same
|
||||
# statement; only prior sentences contribute).
|
||||
_prior_subject: str | None = None
|
||||
for s in statement_sentences:
|
||||
# ADR-0174 Phase 2 — statement-scoped trace of constraint eliminations.
|
||||
# Merged into the question-stage reader_trace below so the consumer
|
||||
# sees both per-sentence eliminations (Phase 2) and reader events
|
||||
# (Phase 1, ADR-0164) in one stream.
|
||||
_statement_trace: list[str] = []
|
||||
for s_idx, s in enumerate(statement_sentences):
|
||||
# RAT-1 — prefer the discourse-level prior (which sees context-filler
|
||||
# sentences like "John adopts a dog from a shelter"); fall back to
|
||||
# the in-loop running subject when discourse map has no entry.
|
||||
|
|
@ -827,24 +834,53 @@ def parse_and_solve(
|
|||
)
|
||||
injected = inject_from_match(recognizer_match, s)
|
||||
if injected:
|
||||
# ADR-0170 — dispatch admissibility on the
|
||||
# concrete candidate type. CandidateInitial uses
|
||||
# the existing _initial_admissible gate;
|
||||
# CandidateOperation uses the parser's
|
||||
# roundtrip_admissible gate (same predicate
|
||||
# operations from the regex path already pass
|
||||
# through). No new admission semantics — each
|
||||
# type is gated by the predicate it was always
|
||||
# gated by; the dispatch just unifies the
|
||||
# injector path with the parser path.
|
||||
admitted: list[SentenceChoice] = []
|
||||
for c in injected:
|
||||
# ADR-0174 Phase 2 — hypothesis-based admission
|
||||
# with structured elimination tracing. Each
|
||||
# injected candidate becomes a Hypothesis with
|
||||
# confidence_rank == emission order; the
|
||||
# constraint propagator runs the same predicates
|
||||
# _initial_admissible / roundtrip_admissible run
|
||||
# today (decomposed into sub-checks) and returns
|
||||
# (survivors, eliminations). Eliminations append
|
||||
# as JSON trace events to reader_trace so the
|
||||
# operator can see WHICH predicate eliminated the
|
||||
# candidate, not just that admission failed.
|
||||
# Admission semantics are byte-equivalent to the
|
||||
# pre-Phase-2 inline loop: a candidate survives
|
||||
# here iff it survived the predicate dispatch
|
||||
# there.
|
||||
from generate.comprehension.constraint_propagation import (
|
||||
eliminate_violating,
|
||||
hypothesis_from_initial,
|
||||
hypothesis_from_operation,
|
||||
)
|
||||
|
||||
hyps_in: list[Hypothesis] = []
|
||||
for rank, c in enumerate(injected):
|
||||
if isinstance(c, CandidateInitial):
|
||||
if _initial_admissible(c):
|
||||
admitted.append(c)
|
||||
hyps_in.append(
|
||||
hypothesis_from_initial(c, rank)
|
||||
)
|
||||
elif isinstance(c, CandidateOperation):
|
||||
if roundtrip_admissible(c):
|
||||
admitted.append(c)
|
||||
hyps_in.append(
|
||||
hypothesis_from_operation(c, rank)
|
||||
)
|
||||
survivors, eliminations = eliminate_violating(
|
||||
tuple(hyps_in)
|
||||
)
|
||||
for elim in eliminations:
|
||||
_statement_trace.append(json.dumps({
|
||||
"layer": "constraint_propagation",
|
||||
"phase": 2,
|
||||
"outcome": "eliminated",
|
||||
"confidence_rank": elim.confidence_rank,
|
||||
"predicate": elim.predicate,
|
||||
"reason": elim.reason,
|
||||
"sentence_index": s_idx,
|
||||
}, sort_keys=True))
|
||||
admitted: list[SentenceChoice] = [
|
||||
h.candidate for h in survivors # type: ignore[misc]
|
||||
]
|
||||
if len(admitted) == len(injected) and admitted:
|
||||
per_sentence_choices.append(
|
||||
_collapse_per_sentence_ties(admitted)
|
||||
|
|
@ -900,7 +936,11 @@ def parse_and_solve(
|
|||
# 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] = []
|
||||
# ADR-0174 Phase 2 — seed reader_trace with statement-stage
|
||||
# constraint-propagation events so consumers see Phase-1 (ADR-0164)
|
||||
# reader events and Phase-2 (ADR-0174) elimination events in one
|
||||
# ordered stream.
|
||||
reader_trace: list[str] = list(_statement_trace)
|
||||
reader_question_choices: list[CandidateUnknown] | None = None
|
||||
_use_reader = (
|
||||
config is not None and config.comprehension_reader_questions
|
||||
|
|
|
|||
433
tests/test_adr_0174_phase2_constraint_propagation.py
Normal file
433
tests/test_adr_0174_phase2_constraint_propagation.py
Normal file
|
|
@ -0,0 +1,433 @@
|
|||
"""ADR-0174 Phase 2 — continuous constraint propagation tests.
|
||||
|
||||
Acceptance tests:
|
||||
|
||||
1. Hypothesis emission adapters wrap CandidateInitial / CandidateOperation
|
||||
into Phase-1 Hypothesis objects with correct rank assignment.
|
||||
|
||||
2. check_constraints runs sub-checks and returns ConstraintResult with
|
||||
specific elimination reasons (decomposed predicate names). Today's
|
||||
admission logic is byte-equivalent — a candidate that
|
||||
_initial_admissible / roundtrip_admissible would admit is admitted
|
||||
here; one they would reject is rejected here with the same
|
||||
short-circuit-on-first-failure semantics.
|
||||
|
||||
3. eliminate_violating returns (survivors, eliminations) with original
|
||||
ranks preserved in eliminations and re-densified ranks in survivors.
|
||||
|
||||
4. The wiring at math_candidate_graph injection sites does not alter
|
||||
admission semantics (3/47/0 preserved on train_sample) and remains
|
||||
deterministic across runs.
|
||||
|
||||
5. When a synthetic candidate fails one of the sub-checks, the
|
||||
elimination is observable in the trace.
|
||||
|
||||
The check_constraints behavior parity with the pre-Phase-2 admission
|
||||
predicates is the load-bearing invariant: any divergence would break
|
||||
wrong=0 by silently weakening admissibility.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from generate.comprehension.constraint_propagation import (
|
||||
ConstraintResult,
|
||||
Elimination,
|
||||
VALID_PREDICATE_NAMES,
|
||||
check_constraints,
|
||||
eliminate_violating,
|
||||
hypothesis_from_initial,
|
||||
hypothesis_from_operation,
|
||||
)
|
||||
from generate.comprehension.state import (
|
||||
HYPOTHESIS_CAP,
|
||||
ComprehensionStateError,
|
||||
Hypothesis,
|
||||
)
|
||||
from generate.math_candidate_graph import _initial_admissible
|
||||
from generate.math_candidate_parser import CandidateInitial
|
||||
from generate.math_problem_graph import (
|
||||
InitialPossession,
|
||||
Operation,
|
||||
Quantity,
|
||||
)
|
||||
from generate.math_roundtrip import CandidateOperation, roundtrip_admissible
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers — construct minimal valid candidates
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _initial(
|
||||
entity: str = "Sam",
|
||||
value: int = 3,
|
||||
unit: str = "apples",
|
||||
source_span: str = "Sam has 3 apples.",
|
||||
matched_anchor: str = "has",
|
||||
matched_value_token: str = "3",
|
||||
matched_unit_token: str = "apples",
|
||||
matched_entity_token: str = "Sam",
|
||||
) -> CandidateInitial:
|
||||
return CandidateInitial(
|
||||
initial=InitialPossession(
|
||||
entity=entity, quantity=Quantity(value=value, unit=unit),
|
||||
),
|
||||
source_span=source_span,
|
||||
matched_anchor=matched_anchor,
|
||||
matched_value_token=matched_value_token,
|
||||
matched_unit_token=matched_unit_token,
|
||||
matched_entity_token=matched_entity_token,
|
||||
)
|
||||
|
||||
|
||||
def _operation_add(
|
||||
actor: str = "Sam",
|
||||
value: int = 5,
|
||||
unit: str = "apples",
|
||||
source_span: str = "Sam buys 5 apples.",
|
||||
matched_verb: str = "buys",
|
||||
matched_value_token: str = "5",
|
||||
matched_unit_token: str = "apples",
|
||||
matched_actor_token: str = "Sam",
|
||||
) -> CandidateOperation:
|
||||
return CandidateOperation(
|
||||
op=Operation(
|
||||
actor=actor, kind="add",
|
||||
operand=Quantity(value=value, unit=unit),
|
||||
),
|
||||
source_span=source_span,
|
||||
matched_verb=matched_verb,
|
||||
matched_value_token=matched_value_token,
|
||||
matched_unit_token=matched_unit_token,
|
||||
matched_actor_token=matched_actor_token,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. Hypothesis emission adapters
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestHypothesisEmission:
|
||||
def test_initial_wraps_candidate_at_given_rank(self) -> None:
|
||||
ic = _initial()
|
||||
hyp = hypothesis_from_initial(ic, rank=0)
|
||||
assert isinstance(hyp, Hypothesis)
|
||||
assert hyp.candidate is ic
|
||||
assert hyp.confidence_rank == 0
|
||||
assert hyp.category_assignments == ()
|
||||
assert hyp.constraint_state == ()
|
||||
assert hyp.unresolved == ()
|
||||
|
||||
def test_operation_wraps_candidate_at_given_rank(self) -> None:
|
||||
op = _operation_add()
|
||||
hyp = hypothesis_from_operation(op, rank=2)
|
||||
assert hyp.candidate is op
|
||||
assert hyp.confidence_rank == 2
|
||||
|
||||
def test_rank_outside_cap_refused(self) -> None:
|
||||
ic = _initial()
|
||||
with pytest.raises(ComprehensionStateError, match="rank must be in"):
|
||||
hypothesis_from_initial(ic, rank=HYPOTHESIS_CAP)
|
||||
with pytest.raises(ComprehensionStateError, match="rank must be in"):
|
||||
hypothesis_from_operation(_operation_add(), rank=-1)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. check_constraints — parity with existing admissibility predicates
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCheckConstraintsInitialParity:
|
||||
"""For CandidateInitial, check_constraints must match
|
||||
_initial_admissible exactly on the admit/reject decision."""
|
||||
|
||||
def test_well_formed_initial_admits(self) -> None:
|
||||
ic = _initial()
|
||||
result = check_constraints(hypothesis_from_initial(ic, 0))
|
||||
assert result.admitted is True
|
||||
assert result.elimination_reason is None
|
||||
assert _initial_admissible(ic) is True # parity
|
||||
|
||||
def test_anchor_not_in_source_eliminated(self) -> None:
|
||||
ic = _initial(
|
||||
matched_anchor="had", # source has "has"
|
||||
source_span="Sam has 3 apples.",
|
||||
)
|
||||
result = check_constraints(hypothesis_from_initial(ic, 0))
|
||||
assert result.admitted is False
|
||||
assert result.elimination_reason is not None
|
||||
assert "matched_anchor" in result.elimination_reason
|
||||
# The first failing predicate is initial.anchor_grounds.
|
||||
first_fail = next(
|
||||
(p for p, o in result.predicates_run if o == "fail"), None
|
||||
)
|
||||
assert first_fail == "initial.anchor_grounds"
|
||||
assert _initial_admissible(ic) is False # parity
|
||||
|
||||
def test_value_not_in_source_eliminated(self) -> None:
|
||||
ic = _initial(
|
||||
matched_value_token="99", # source has "3"
|
||||
source_span="Sam has 3 apples.",
|
||||
)
|
||||
result = check_constraints(hypothesis_from_initial(ic, 0))
|
||||
assert result.admitted is False
|
||||
assert "matched_value_token" in (result.elimination_reason or "")
|
||||
assert _initial_admissible(ic) is False
|
||||
|
||||
def test_unit_not_in_source_eliminated(self) -> None:
|
||||
ic = _initial(
|
||||
matched_unit_token="oranges", # source has "apples"
|
||||
source_span="Sam has 3 apples.",
|
||||
)
|
||||
result = check_constraints(hypothesis_from_initial(ic, 0))
|
||||
assert result.admitted is False
|
||||
assert "matched_unit_token" in (result.elimination_reason or "")
|
||||
assert _initial_admissible(ic) is False
|
||||
|
||||
def test_entity_not_in_source_eliminated(self) -> None:
|
||||
ic = _initial(
|
||||
matched_entity_token="Tom", # source has "Sam"
|
||||
source_span="Sam has 3 apples.",
|
||||
)
|
||||
result = check_constraints(hypothesis_from_initial(ic, 0))
|
||||
assert result.admitted is False
|
||||
assert _initial_admissible(ic) is False
|
||||
|
||||
|
||||
class TestCheckConstraintsOperationParity:
|
||||
"""For CandidateOperation, check_constraints must match
|
||||
roundtrip_admissible exactly on the admit/reject decision."""
|
||||
|
||||
def test_well_formed_operation_admits(self) -> None:
|
||||
op = _operation_add()
|
||||
result = check_constraints(hypothesis_from_operation(op, 0))
|
||||
assert result.admitted is True
|
||||
assert result.elimination_reason is None
|
||||
assert roundtrip_admissible(op) is True
|
||||
|
||||
def test_verb_not_registered_for_kind_eliminated(self) -> None:
|
||||
# "buys" is registered for "add", not "subtract" — but constructing
|
||||
# an Operation with the wrong kind would fail at construction.
|
||||
# Use a verb that's not in any add-verb set.
|
||||
op = CandidateOperation(
|
||||
op=Operation(
|
||||
actor="Sam", kind="add",
|
||||
operand=Quantity(value=5, unit="apples"),
|
||||
),
|
||||
source_span="Sam thinks 5 apples.", # "thinks" not in ADD_VERBS
|
||||
matched_verb="thinks",
|
||||
matched_value_token="5",
|
||||
matched_unit_token="apples",
|
||||
matched_actor_token="Sam",
|
||||
)
|
||||
result = check_constraints(hypothesis_from_operation(op, 0))
|
||||
assert result.admitted is False
|
||||
first_fail = next(
|
||||
(p for p, o in result.predicates_run if o == "fail"), None
|
||||
)
|
||||
assert first_fail == "operation.verb_registered"
|
||||
assert roundtrip_admissible(op) is False
|
||||
|
||||
def test_actor_not_in_source_eliminated(self) -> None:
|
||||
op = _operation_add(
|
||||
matched_actor_token="Tom", # source has "Sam"
|
||||
source_span="Sam buys 5 apples.",
|
||||
)
|
||||
result = check_constraints(hypothesis_from_operation(op, 0))
|
||||
assert result.admitted is False
|
||||
first_fail = next(
|
||||
(p for p, o in result.predicates_run if o == "fail"), None
|
||||
)
|
||||
assert first_fail == "operation.actor_grounds"
|
||||
assert roundtrip_admissible(op) is False
|
||||
|
||||
|
||||
class TestCheckConstraintsResultShape:
|
||||
def test_predicates_run_only_uses_known_predicate_names(self) -> None:
|
||||
ic = _initial()
|
||||
result = check_constraints(hypothesis_from_initial(ic, 0))
|
||||
for predicate_name, _outcome in result.predicates_run:
|
||||
assert predicate_name in VALID_PREDICATE_NAMES, (
|
||||
f"predicate {predicate_name!r} not in VALID_PREDICATE_NAMES; "
|
||||
"adding new predicates requires an ADR amendment"
|
||||
)
|
||||
|
||||
def test_unknown_candidate_type_eliminated(self) -> None:
|
||||
# Wrap a string as candidate — not a known type
|
||||
hyp = Hypothesis(
|
||||
candidate=("not a candidate",), # serialisable sentinel
|
||||
category_assignments=(),
|
||||
constraint_state=(),
|
||||
confidence_rank=0,
|
||||
unresolved=(),
|
||||
)
|
||||
result = check_constraints(hyp)
|
||||
assert result.admitted is False
|
||||
assert "unknown candidate type" in (result.elimination_reason or "")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. eliminate_violating
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEliminateViolating:
|
||||
def test_all_admit_returns_all_survivors_no_eliminations(self) -> None:
|
||||
h0 = hypothesis_from_initial(_initial(entity="Sam"), 0)
|
||||
h1 = hypothesis_from_operation(_operation_add(actor="Sam"), 1)
|
||||
survivors, eliminations = eliminate_violating((h0, h1))
|
||||
assert len(survivors) == 2
|
||||
assert eliminations == ()
|
||||
# Ranks preserved (already dense from 0).
|
||||
assert survivors[0].confidence_rank == 0
|
||||
assert survivors[1].confidence_rank == 1
|
||||
|
||||
def test_all_eliminated_returns_no_survivors(self) -> None:
|
||||
h0 = hypothesis_from_initial(
|
||||
_initial(matched_unit_token="oranges"), 0
|
||||
)
|
||||
h1 = hypothesis_from_initial(
|
||||
_initial(matched_unit_token="bananas"), 1
|
||||
)
|
||||
survivors, eliminations = eliminate_violating((h0, h1))
|
||||
assert survivors == ()
|
||||
assert len(eliminations) == 2
|
||||
# Original ranks preserved in eliminations.
|
||||
ranks = sorted(e.confidence_rank for e in eliminations)
|
||||
assert ranks == [0, 1]
|
||||
|
||||
def test_partial_elimination_redensifies_survivor_ranks(self) -> None:
|
||||
# h0 fails (bad unit), h1 succeeds.
|
||||
h0 = hypothesis_from_initial(
|
||||
_initial(matched_unit_token="oranges"), 0
|
||||
)
|
||||
h1 = hypothesis_from_initial(_initial(), 1)
|
||||
survivors, eliminations = eliminate_violating((h0, h1))
|
||||
assert len(survivors) == 1
|
||||
assert survivors[0].confidence_rank == 0 # re-densified from 1
|
||||
assert len(eliminations) == 1
|
||||
assert eliminations[0].confidence_rank == 0 # original rank preserved
|
||||
|
||||
def test_eliminations_carry_predicate_name(self) -> None:
|
||||
h = hypothesis_from_initial(
|
||||
_initial(matched_anchor="had"), 0 # anchor not in source
|
||||
)
|
||||
_, eliminations = eliminate_violating((h,))
|
||||
assert len(eliminations) == 1
|
||||
assert eliminations[0].predicate in VALID_PREDICATE_NAMES
|
||||
assert eliminations[0].predicate == "initial.anchor_grounds"
|
||||
|
||||
|
||||
class TestEliminationDataclass:
|
||||
def test_invalid_predicate_refused(self) -> None:
|
||||
with pytest.raises(
|
||||
ComprehensionStateError, match="must be in VALID_PREDICATE_NAMES"
|
||||
):
|
||||
Elimination(confidence_rank=0, predicate="bogus", reason="x")
|
||||
|
||||
def test_empty_reason_refused(self) -> None:
|
||||
with pytest.raises(
|
||||
ComprehensionStateError, match="reason must be non-empty"
|
||||
):
|
||||
Elimination(
|
||||
confidence_rank=0,
|
||||
predicate="initial.anchor_grounds",
|
||||
reason="",
|
||||
)
|
||||
|
||||
|
||||
class TestConstraintResultDataclass:
|
||||
def test_admitted_with_elimination_reason_refused(self) -> None:
|
||||
with pytest.raises(
|
||||
ComprehensionStateError, match="inconsistent"
|
||||
):
|
||||
ConstraintResult(
|
||||
admitted=True,
|
||||
predicates_run=(("initial.anchor_grounds", "ok"),),
|
||||
elimination_reason="impossible combo",
|
||||
)
|
||||
|
||||
def test_rejected_without_reason_refused(self) -> None:
|
||||
with pytest.raises(
|
||||
ComprehensionStateError, match="requires a non-None"
|
||||
):
|
||||
ConstraintResult(
|
||||
admitted=False,
|
||||
predicates_run=(("initial.anchor_grounds", "fail"),),
|
||||
elimination_reason=None,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. Integration — wiring at math_candidate_graph injection sites
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestIntegrationWithCandidateGraph:
|
||||
"""End-to-end: feed a real problem through parse_and_solve and verify
|
||||
the trace stream is well-formed JSON when populated, and admission
|
||||
semantics are preserved."""
|
||||
|
||||
def test_correct_case_still_admits(self) -> None:
|
||||
"""Case 0014 is one of the 3 correct cases; Phase 2 wiring must
|
||||
not break it."""
|
||||
from generate.math_candidate_graph import parse_and_solve
|
||||
text = (
|
||||
"Bob can shuck 10 oysters in 5 minutes. "
|
||||
"How many oysters can he shuck in 2 hours?"
|
||||
)
|
||||
r = parse_and_solve(text)
|
||||
assert r.answer == 240
|
||||
assert r.refusal_reason is None
|
||||
|
||||
def test_trace_events_are_valid_json(self) -> None:
|
||||
"""Every event in reader_trace must be parseable JSON — Phase 2
|
||||
events conform to the same shape contract as Phase 1 events."""
|
||||
from generate.math_candidate_graph import parse_and_solve
|
||||
# Run all 3 correct cases; any trace events must be valid JSON.
|
||||
texts = [
|
||||
"Bob can shuck 10 oysters in 5 minutes. "
|
||||
"How many oysters can he shuck in 2 hours?",
|
||||
"Xavier plays football with his friends. "
|
||||
"During 15 minutes Xavier can score 2 goals on average. "
|
||||
"How many goals does Xavier score in 2 hours?",
|
||||
]
|
||||
for text in texts:
|
||||
r = parse_and_solve(text)
|
||||
for ev_str in r.reader_trace:
|
||||
ev = json.loads(ev_str) # raises on bad JSON
|
||||
assert "layer" in ev
|
||||
assert "phase" in ev
|
||||
|
||||
def test_phase2_event_shape_when_synthesized(self) -> None:
|
||||
"""When an elimination DOES occur, the event has the documented
|
||||
Phase-2 shape. We verify directly against eliminate_violating
|
||||
rather than the full pipeline because today's injectors are
|
||||
conservative enough that real eliminations do not fire on the
|
||||
train_sample corpus."""
|
||||
h_bad = hypothesis_from_initial(
|
||||
_initial(matched_unit_token="oranges"), 0
|
||||
)
|
||||
_, eliminations = eliminate_violating((h_bad,))
|
||||
# Serialise as the math_candidate_graph wiring does:
|
||||
ev: dict[str, Any] = {
|
||||
"layer": "constraint_propagation",
|
||||
"phase": 2,
|
||||
"outcome": "eliminated",
|
||||
"confidence_rank": eliminations[0].confidence_rank,
|
||||
"predicate": eliminations[0].predicate,
|
||||
"reason": eliminations[0].reason,
|
||||
"sentence_index": 0,
|
||||
}
|
||||
encoded = json.dumps(ev, sort_keys=True)
|
||||
decoded = json.loads(encoded)
|
||||
assert decoded["layer"] == "constraint_propagation"
|
||||
assert decoded["phase"] == 2
|
||||
assert decoded["predicate"] == "initial.unit_grounds"
|
||||
assert decoded["outcome"] == "eliminated"
|
||||
Loading…
Reference in a new issue