65405f1128
* feat(derivation): Gate A2a unit partition injection Add typed unit_partition primitive with PartitionChunk/result_unit contract, recognizer-injector bridge, DCS yield guard, and pronoun lookback support. Closes unit_partition recognized_no_injection on live train_sample (0002 partition stmt reclassifies); wrong=0 preserved. * test(gsm8k): harden unit partition confusers * test(gsm8k): add unit partition pronoun safety regressions * chore(gsm8k): fix unit partition exemplar file ending * chore(derivation): type unit partition solution step operand
750 lines
29 KiB
Python
750 lines
29 KiB
Python
"""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",
|
|
"operation.partition_result_unit_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, PartitionChunk, 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"))
|
|
elif op.kind == "unit_partition":
|
|
if not isinstance(op.operand, PartitionChunk):
|
|
run.append(("operation.operand_shape_consistent", "fail"))
|
|
return ConstraintResult(
|
|
admitted=False,
|
|
predicates_run=tuple(run),
|
|
elimination_reason=(
|
|
"op.kind='unit_partition' requires PartitionChunk "
|
|
f"operand; got {type(op.operand).__name__}"
|
|
),
|
|
)
|
|
run.append(("operation.operand_shape_consistent", "ok"))
|
|
if not _token_in(op.operand.result_unit, haystack):
|
|
run.append(("operation.partition_result_unit_grounds", "fail"))
|
|
return ConstraintResult(
|
|
admitted=False,
|
|
predicates_run=tuple(run),
|
|
elimination_reason=(
|
|
f"PartitionChunk.result_unit "
|
|
f"{op.operand.result_unit!r} does not ground"
|
|
),
|
|
)
|
|
run.append(("operation.partition_result_unit_grounds", "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",
|
|
]
|