core/generate/binding_graph/adapter.py
Shay 3b30eb248a
feat(binding-graph): Phase 4 question-target binding (ADR-0135) (#179)
Refines BoundUnknown from "the symbol whose value the solver determines"
to "the symbol at a specific temporal/state index with a specific
question-form". Two new required fields on BoundUnknown — state_index
(initial/terminal/Operation(operation_index)) and question_form
(count/rate/total/difference/ratio/identity) — populated by the new
pure-function resolver in generate/binding_graph/question_target.py.

The adapter (ADR-0133) now delegates Unknown -> BoundUnknown construction
to bound_unknown_from_math_problem_graph. No runtime wiring, no solver
invocation. Phase 5 (bounded-grammar / B3 integration) remains deferred.

Refusal-first via the new QuestionTargetError (sibling of AdapterError /
AdmissibilityError). Closed reason vocab: not_a_math_problem_graph,
unknown_entity_not_in_entities, apply_rate_unit_mismatch,
unmappable_question_form. Closed precedence rule on question_form
documented in ADR-0135 (compare_multiplicative > compare_additive >
apply_rate{numerator|denominator unit-match} > count); ambiguity refuses.

SemanticSymbolicBindingGraph.__post_init__ gains a cross-collection
guard: Operation(operation_index) must satisfy operation_index <
len(equations). canonical_string emission widened to include
state=... form=... tokens (hash differs from Phase 3 main by design —
not a regression; byte-equal across runs preserved).

Parents: ADR-0132 / ADR-0133 / ADR-0134.

Tests: +70 new (45 unit in test_binding_graph_question_target.py +
25 integration in test_binding_graph_adapter_question_target.py); 5
Phase 1+3 BoundUnknown fixtures migrated. Total binding-graph lane
295/1 pass (1 pre-existing test_symbol_binding_uses_slots failure on
Python 3.14, unrelated to Phase 4 — exists on origin/main). Pyright
clean on new and modified files. No edits to algebra/, chat/, core/,
or runtime hot path. Field invariant untouched.
2026-05-23 11:24:49 -07:00

455 lines
17 KiB
Python

"""ADR-0133 / ADR-0134 — Adapter: ``MathProblemGraph`` → ``SemanticSymbolicBindingGraph``.
Phase 2 of the binding-graph layer (ADR-0133) introduced the pure
translation. Phase 3 (ADR-0134) wires unit-aware admissibility: every
emitted :class:`BoundEquation` now carries either
``admissibility_status='admitted'`` + populated ``unit_proof``, or
``admissibility_status='refused'`` + populated ``refusal_reason``. The
adapter contract / input type / output type are unchanged.
Mapping discipline (locked at top of module — see constants):
- each entity → one ``SymbolBinding`` with ``semantic_role='entity'``,
- each ``InitialPossession`` → one ``SymbolBinding``
(``semantic_role='quantity'``) + one ``BoundFact``,
- each ``Operation`` → one fresh result ``SymbolBinding`` plus one
``BoundEquation`` whose ``operation_kind`` is a verbatim passthrough
of the source op kind (closed vocab is shared by design),
- the ``Unknown`` → one synthesized ``SymbolBinding``
(``semantic_role='unknown'``) + one ``BoundUnknown``.
Phases 4+ deferred:
- question-target binding refinement (Phase 4),
- bounded-grammar / B3 integration (Phase 5).
Phase 3 also synthesizes a small number of *literal* operand symbols so
``check_admissibility`` can verify multiplicative-class equations from
``BoundEquation`` + ``symbols`` alone (no solver, no operand parsing). The
naming convention is locked here and consumed by
:mod:`generate.binding_graph.admissibility`:
- ``divide`` operand → ``op<NNN>__divisor`` literal SymbolBinding;
- ``multiply`` operand → ``op<NNN>__multiplicand`` literal SymbolBinding;
- ``apply_rate`` operand → ``op<NNN>__rate`` SymbolBinding with
``semantic_role='rate'`` and ``unit='<num>_per_<denom>'``.
"""
from __future__ import annotations
import re
from typing import Final
from generate.math_problem_graph import (
Comparison,
MathProblemGraph,
Operation,
Quantity,
Rate,
)
from .admissibility import AdmissibilityError, check_admissibility
from .model import (
BindingGraphError,
BoundEquation,
BoundFact,
SemanticSymbolicBindingGraph,
SourceSpanLink,
SymbolBinding,
)
from .question_target import bound_unknown_from_math_problem_graph
# ---------------------------------------------------------------------------
# Constants — locked mapping discipline (read these before editing logic).
# ---------------------------------------------------------------------------
#: ``source_id`` stamped onto every synthesized ``SourceSpanLink``.
#: ``MathProblemGraph`` carries no native source-span information, so
#: every span the adapter emits is synthetic and shares this id.
SYNTHETIC_SOURCE_ID: Final[str] = "math_problem_graph"
#: ``introduced_by`` stamped onto every ``SymbolBinding`` the adapter
#: emits. Replaying the adapter therefore yields byte-equal symbols.
INTRODUCED_BY: Final[str] = "bind_math_problem_graph"
#: Sentinel ``unit_proof`` stored on every refused equation. The matching
#: ``refusal_reason`` carries the typed failure reason; this sentinel exists
#: only because :attr:`BoundEquation.unit_proof` is non-optional in the
#: Phase-1 data model (see ADR-0132).
REFUSED_UNIT_PROOF: Final[str] = "unverified"
_SLUG_NON_ALNUM = re.compile(r"[^a-z0-9]+")
# ---------------------------------------------------------------------------
# Public error
# ---------------------------------------------------------------------------
class AdapterError(ValueError):
"""Raised on malformed input to :func:`bind_math_problem_graph`.
Sibling of :class:`generate.binding_graph.BindingGraphError` —
refusal-first by design. The adapter never silently coerces an
unrecognized input type.
"""
# ---------------------------------------------------------------------------
# Symbol-id helpers (pure)
# ---------------------------------------------------------------------------
def _slug(text: str) -> str:
"""ASCII-lowercase slug; non-alnum runs collapse to ``_``."""
return _SLUG_NON_ALNUM.sub("_", text.strip().lower()).strip("_")
def _safe_identifier(text: str, *, prefix: str) -> str:
"""Return ``f"{prefix}_{slug}"``, defaulting to ``prefix + '_x'`` on
empty slug. Guarantees a valid Python identifier."""
s = _slug(text)
if s == "":
s = "x"
return f"{prefix}_{s}"
def _entity_symbol_id(entity: str) -> str:
return _safe_identifier(entity, prefix="entity")
def _quantity_symbol_id(entity: str, unit: str, tick: int) -> str:
return f"q_{_slug(entity) or 'x'}_{_slug(unit) or 'x'}_t{tick}"
def _op_result_symbol_id(idx: int) -> str:
return f"op_{idx:03d}_result"
def _span(text: str) -> SourceSpanLink:
"""Synthesize a deterministic ``SourceSpanLink`` for ``text``.
``MathProblemGraph`` carries no native span information; Phase 2
therefore stamps every binding with a synthetic span anchored to
the rendered surface text. The span is byte-stable per input.
"""
if not isinstance(text, str) or text == "":
# Defensive: every caller passes non-empty text. Refuse rather
# than silently substitute.
raise AdapterError("synthetic span text must be a non-empty str")
return SourceSpanLink(
source_id=SYNTHETIC_SOURCE_ID, start=0, end=len(text), text=text
)
# ---------------------------------------------------------------------------
# RHS canonicalization (deterministic, string-only — no Polynomial coupling)
# ---------------------------------------------------------------------------
def _format_quantity(q: Quantity) -> str:
return f"{q.value} {q.unit}"
def _format_rate(r: Rate) -> str:
return f"{r.value} {r.numerator_unit}/{r.denominator_unit}"
def _format_comparison(c: Comparison) -> str:
if c.delta is not None:
return (
f"{c.direction}({c.reference_actor}, "
f"delta={_format_quantity(c.delta)})"
)
return f"{c.direction}({c.reference_actor}, factor={c.factor})"
def _format_operand(operand: Quantity | Rate | Comparison) -> str:
if isinstance(operand, Quantity):
return _format_quantity(operand)
if isinstance(operand, Rate):
return _format_rate(operand)
return _format_comparison(operand)
def _format_rhs(op: Operation) -> str:
head = f"{op.kind}({op.actor}"
if op.target is not None:
head += f"->{op.target}"
return head + f", {_format_operand(op.operand)})"
def _operand_unit_hint(operand: Quantity | Rate | Comparison) -> str | None:
"""The most relevant unit a dependency lookup should key on.
Used only for wiring deterministic ``BoundEquation.dependencies``
against pre-existing t0 symbols. Unit-aware *admissibility* is
Phase 3.
"""
if isinstance(operand, Quantity):
return operand.unit
if isinstance(operand, Rate):
return operand.denominator_unit
if operand.delta is not None:
return operand.delta.unit
return None
# ---------------------------------------------------------------------------
# Public adapter
# ---------------------------------------------------------------------------
def bind_math_problem_graph(
g: object,
) -> SemanticSymbolicBindingGraph:
"""Translate a ``MathProblemGraph`` into a ``SemanticSymbolicBindingGraph``.
Pure function. Deterministic: ``bind_math_problem_graph(g) ==
bind_math_problem_graph(g)`` byte-for-byte, and two graphs that
compare equal (``g1 == g2``) produce two binding graphs that compare
equal. Input is never mutated (cannot be — ``MathProblemGraph`` is
frozen — but the contract is asserted by tests).
Raises :class:`AdapterError` if ``g`` is not a
:class:`MathProblemGraph`. Every well-formed ``MathProblemGraph``
produces a well-formed ``SemanticSymbolicBindingGraph`` — the
adapter is total on the input type's image.
"""
if not isinstance(g, MathProblemGraph):
raise AdapterError(
"bind_math_problem_graph requires a MathProblemGraph; "
f"got {type(g).__name__}"
)
symbols: list[SymbolBinding] = []
facts: list[BoundFact] = []
equations: list[BoundEquation] = []
seen_ids: set[str] = set()
def _add(sym: SymbolBinding) -> None:
if sym.symbol_id in seen_ids:
# Idempotent — same symbol re-emitted from a different
# construction path collapses cleanly.
return
seen_ids.add(sym.symbol_id)
symbols.append(sym)
# ---- Entities (order of introduction) ---------------------------------
for entity in g.entities:
_add(
SymbolBinding(
symbol_id=_entity_symbol_id(entity),
name=entity,
semantic_role="entity",
source_span=_span(entity),
introduced_by=INTRODUCED_BY,
entity=entity,
)
)
# ---- Initial state → t0 quantity symbols + grounded facts -------------
t0_index: dict[tuple[str, str], str] = {}
for poss in g.initial_state:
sid = _quantity_symbol_id(poss.entity, poss.quantity.unit, 0)
span_text = f"{poss.entity}|{poss.quantity.unit}|t0"
_add(
SymbolBinding(
symbol_id=sid,
name=f"{poss.entity}.{poss.quantity.unit}@t0",
semantic_role="quantity",
source_span=_span(span_text),
introduced_by=INTRODUCED_BY,
entity=poss.entity,
unit=poss.quantity.unit,
)
)
facts.append(
BoundFact(
symbol_id=sid,
value=str(poss.quantity.value),
source_span=_span(span_text),
unit=poss.quantity.unit,
)
)
t0_index[(poss.entity, poss.quantity.unit)] = sid
# ---- Operations → fresh result symbol + bound equation ----------------
for idx, op in enumerate(g.operations):
result_sid = _op_result_symbol_id(idx)
op_span_text = f"op{idx:03d}|{op.kind}|{op.actor}"
_add(
SymbolBinding(
symbol_id=result_sid,
name=f"op{idx}.{op.kind}.{op.actor}",
semantic_role="quantity",
source_span=_span(op_span_text),
introduced_by=INTRODUCED_BY,
entity=op.actor,
)
)
deps: set[str] = set()
unit_hint = _operand_unit_hint(op.operand)
if unit_hint is not None:
actor_sid = t0_index.get((op.actor, unit_hint))
if actor_sid is not None:
deps.add(actor_sid)
if op.target is not None:
target_sid = t0_index.get((op.target, unit_hint))
if target_sid is not None:
deps.add(target_sid)
if isinstance(op.operand, Comparison):
ref_sid = t0_index.get(
(op.operand.reference_actor, unit_hint)
)
if ref_sid is not None:
deps.add(ref_sid)
# ---- Phase 3: synth literal operand symbols where needed ----------
# The verifier (admissibility.check_admissibility) re-derives the
# proof from dep symbols. multiply/divide/apply_rate need explicit
# operand-unit symbols because the operand quantity is literal in
# the source op (not a pre-existing t0 symbol). add/subtract/
# transfer/compare_additive already share their unit with an actor
# t0 dep. compare_multiplicative is dimensionless.
if op.kind in ("multiply", "divide") and isinstance(op.operand, Quantity):
# For multiply/divide the actor's existing t0 quantity acts as
# the dividend / first factor. Its unit need not match the
# operand's, so wire it here even when ``unit_hint`` doesn't
# yield a match above. Sorted by unit-key for determinism.
for (entity, _unit_key), sid in sorted(t0_index.items()):
if entity == op.actor:
deps.add(sid)
break
suffix = "__divisor" if op.kind == "divide" else "__multiplicand"
lit_sid = f"op_{idx:03d}{suffix}"
lit_span_text = f"op{idx:03d}|literal|{op.operand.unit}"
_add(
SymbolBinding(
symbol_id=lit_sid,
name=f"op{idx}.literal.{op.operand.unit}",
semantic_role="quantity",
source_span=_span(lit_span_text),
introduced_by=INTRODUCED_BY,
unit=op.operand.unit,
)
)
facts.append(
BoundFact(
symbol_id=lit_sid,
value=str(op.operand.value),
source_span=_span(lit_span_text),
unit=op.operand.unit,
)
)
deps.add(lit_sid)
elif op.kind == "apply_rate" and isinstance(op.operand, Rate):
rate_sid = f"op_{idx:03d}__rate"
composite_unit = (
f"{op.operand.numerator_unit}_per_{op.operand.denominator_unit}"
)
rate_span_text = f"op{idx:03d}|rate|{composite_unit}"
_add(
SymbolBinding(
symbol_id=rate_sid,
name=f"op{idx}.rate.{composite_unit}",
semantic_role="rate",
source_span=_span(rate_span_text),
introduced_by=INTRODUCED_BY,
unit=composite_unit,
)
)
facts.append(
BoundFact(
symbol_id=rate_sid,
value=str(op.operand.value),
source_span=_span(rate_span_text),
unit=composite_unit,
)
)
deps.add(rate_sid)
# ---- Phase 3: build the equation, then check admissibility --------
# ``check_admissibility`` operates on the equation + the symbol map
# we are *currently* building, so materialize the snapshot here.
symbols_snapshot: dict[str, SymbolBinding] = {
s.symbol_id: s for s in symbols
}
# The equation is constructed twice (pre-check shell with placeholder,
# then a final form) so we can hand a real ``BoundEquation`` to the
# verifier without leaking the placeholder into the binding graph.
proof_token = REFUSED_UNIT_PROOF
status = "refused"
refusal: str | None = None
try:
shell = BoundEquation(
lhs_symbol_id=result_sid,
rhs_canonical=_format_rhs(op),
dependencies=frozenset(deps),
operation_kind=op.kind,
unit_proof=REFUSED_UNIT_PROOF,
admissibility_status="refused",
source_span=_span(op_span_text),
refusal_reason="pre_check",
)
proof = check_admissibility(shell, symbols=symbols_snapshot)
except AdmissibilityError as exc:
refusal = exc.reason
else:
proof_token = proof.to_canonical_string()
status = "admitted"
refusal = None
equations.append(
BoundEquation(
lhs_symbol_id=result_sid,
rhs_canonical=_format_rhs(op),
dependencies=frozenset(deps),
operation_kind=op.kind,
unit_proof=proof_token,
admissibility_status=status,
source_span=_span(op_span_text),
refusal_reason=refusal,
)
)
# ---- Unknown → synthesized unknown symbol + BoundUnknown --------------
# ADR-0135: ``bound_unknown_from_math_problem_graph`` now owns the
# refined ``BoundUnknown`` (symbol_id / question_span / state_index /
# question_form / expected_unit). The adapter only retains responsibility
# for the matching ``SymbolBinding`` (``semantic_role='unknown'``).
unk = g.unknown
unk_text = (
f"{unk.entity}|{unk.unit}" if unk.entity is not None else f"total|{unk.unit}"
)
bound_unknown = bound_unknown_from_math_problem_graph(g)
_add(
SymbolBinding(
symbol_id=bound_unknown.symbol_id,
name=unk_text,
semantic_role="unknown",
source_span=_span(unk_text),
introduced_by=INTRODUCED_BY,
entity=unk.entity,
unit=unk.unit,
)
)
unknowns = (bound_unknown,)
try:
return SemanticSymbolicBindingGraph(
symbols=tuple(symbols),
facts=tuple(facts),
equations=tuple(equations),
unknowns=unknowns,
)
except BindingGraphError as exc:
# Cross-collection invariants of ``SemanticSymbolicBindingGraph``
# are stricter than ``MathProblemGraph``'s own checks. Surface
# any such failure as an ``AdapterError`` so callers see a single
# refusal type at the adapter boundary.
raise AdapterError(
f"adapter produced an invalid binding graph: {exc}"
) from exc