diff --git a/generate/comprehension/constraint_propagation.py b/generate/comprehension/constraint_propagation.py new file mode 100644 index 00000000..4e2a88b1 --- /dev/null +++ b/generate/comprehension/constraint_propagation.py @@ -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", +] diff --git a/generate/math_candidate_graph.py b/generate/math_candidate_graph.py index b25f31ad..9448c0d6 100644 --- a/generate/math_candidate_graph.py +++ b/generate/math_candidate_graph.py @@ -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 diff --git a/tests/test_adr_0174_phase2_constraint_propagation.py b/tests/test_adr_0174_phase2_constraint_propagation.py new file mode 100644 index 00000000..b091c383 --- /dev/null +++ b/tests/test_adr_0174_phase2_constraint_propagation.py @@ -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"