ADR-0184 S1 (state/bind.py, state/change.py) landed the reusable referent and
change-cue helpers. S2 adds the first explicit state-transition substrate and
routes both accumulation readings through it:
- state/model.py frozen SET/GAIN/LOSS transition model over one entity/unit
StateKey, with structural validation (closed op set; unit is a
str with "" = unitless, mirroring the extractor's Quantity.unit
contract; entity optional, mirroring leading_subject_token).
- state/ledger.py build_accumulation_ledger() — the proven single-referent
gain/loss reading, expressed as a SemanticLedger.
- state/replay.py replay_accumulation_ledger() — the only bridge to
GroundedDerivation; refuses any non-accumulation ledger shape
(non-SET start, cross-key mutation).
accumulate.py's _build_accumulation / _build_accumulation_anchor_skip now build a
ledger and replay it instead of constructing the arithmetic chain inline; the
public compose_accumulation()/accumulation_candidates() surfaces are unchanged.
Behavior-equivalent: a byte-for-byte differential over 937 real GSM8K problems
(accumulation_candidates, compose_accumulation, pooled_candidates, resolve_pooled)
shows 0 differences vs main. Semantic worlds never commit directly — replay emits
GroundedDerivation and verify.py / pool.py remain the sole authority. Sealed: no
serving/runtime/chat import; no CLAIMS or eval-lane-SHA movement.
Refs ADR-0184 section 7 (S2).
83 lines
3 KiB
Python
83 lines
3 KiB
Python
"""ADR-0184 S2 — semantic-ledger construction for accumulation.
|
|
|
|
This module owns only the semantic transition construction. It deliberately does
|
|
not verify or commit answers; callers replay the ledger into ``GroundedDerivation``
|
|
and then use the existing verifier/pool.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from generate.derivation.extract import extract_quantities
|
|
from generate.derivation.state.bind import continues_anchor_referent, leading_subject_token
|
|
from generate.derivation.state.change import classify_change_polarity, select_change_cue
|
|
from generate.derivation.state.model import (
|
|
SemanticLedger,
|
|
SemanticQuantity,
|
|
StateKey,
|
|
StateTransition,
|
|
)
|
|
|
|
|
|
def build_accumulation_ledger(
|
|
quantity_clauses: list[str], *, drop_isolated_foreign: bool
|
|
) -> SemanticLedger | None:
|
|
"""Build the single-referent gain/loss accumulation ledger.
|
|
|
|
``quantity_clauses`` must be the already-filtered sequence of clauses/sub-clauses
|
|
that contain quantities. The behavior mirrors the pre-S2 accumulation composer:
|
|
|
|
* first clause must have exactly one quantity and becomes ``set``;
|
|
* later clauses must continue the anchor referent;
|
|
* each later clause must reduce to exactly one change quantity;
|
|
* polarity must be an unambiguous licensed gain/loss cue;
|
|
* change quantities inherit the anchor unit during replay.
|
|
"""
|
|
|
|
if len(quantity_clauses) < 2:
|
|
return None
|
|
|
|
anchor_clause, *change_clauses = quantity_clauses
|
|
anchor_quantities = extract_quantities(anchor_clause)
|
|
if len(anchor_quantities) != 1:
|
|
return None
|
|
|
|
anchor_quantity = anchor_quantities[0]
|
|
anchor_subject = leading_subject_token(anchor_clause)
|
|
key = StateKey(entity=anchor_subject, unit=anchor_quantity.unit)
|
|
transitions: list[StateTransition] = [
|
|
StateTransition(
|
|
key=key,
|
|
op="set",
|
|
quantity=SemanticQuantity.from_quantity(anchor_quantity, clause_index=0),
|
|
cue="set",
|
|
clause_index=0,
|
|
)
|
|
]
|
|
|
|
for idx, clause in enumerate(change_clauses, start=1):
|
|
if not continues_anchor_referent(clause, anchor_subject):
|
|
return None
|
|
change_quantities = list(extract_quantities(clause))
|
|
if drop_isolated_foreign and len(change_quantities) > 1:
|
|
change_quantities = [
|
|
q for q in change_quantities if not (q.unit and q.unit != anchor_quantity.unit)
|
|
]
|
|
if len(change_quantities) != 1:
|
|
return None
|
|
polarity = classify_change_polarity(clause)
|
|
if polarity is None:
|
|
return None
|
|
change = change_quantities[0]
|
|
transitions.append(
|
|
StateTransition(
|
|
key=key,
|
|
op="gain" if polarity > 0 else "loss",
|
|
quantity=SemanticQuantity.from_quantity(change, clause_index=idx),
|
|
cue=select_change_cue(clause, polarity),
|
|
clause_index=idx,
|
|
)
|
|
)
|
|
|
|
if len(transitions) <= 1:
|
|
return None
|
|
return SemanticLedger(transitions=tuple(transitions))
|