core/generate/derivation/state/replay.py
Shay 9dcb00248c feat(derivation): route accumulation through ADR-0184 S2 semantic ledger
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).
2026-06-10 12:00:23 -07:00

49 lines
1.8 KiB
Python

"""ADR-0184 S2 — replay semantic ledgers into arithmetic proof objects.
Replay is the only bridge from scoped semantic state to arithmetic candidates.
It emits ``GroundedDerivation`` so the existing verifier/classifier/pool remain the
authoritative commit path.
"""
from __future__ import annotations
from generate.derivation.model import GroundedDerivation, Step
from generate.derivation.state.model import SemanticLedger, StateTransition
def replay_accumulation_ledger(ledger: SemanticLedger) -> GroundedDerivation | None:
"""Replay a SET_STATE + GAIN/LOSS ledger to ``GroundedDerivation``.
Returns ``None`` when the ledger is not the narrow accumulation shape S2 owns.
This keeps future transition kinds from accidentally being interpreted as a
linear arithmetic chain before they receive their own proof model.
"""
transitions = ledger.transitions
if len(transitions) < 2:
return None
start_transition = transitions[0]
if start_transition.op != "set":
return None
key = start_transition.key
start = start_transition.quantity.to_quantity()
steps: list[Step] = []
for transition in transitions[1:]:
if not _same_key(transition, start_transition):
return None
if transition.op not in {"gain", "loss"}:
return None
op = "add" if transition.op == "gain" else "subtract"
operand = transition.quantity.to_quantity(unit_override=key.unit)
steps.append(Step(op=op, operand=operand, cue=transition.cue))
if not steps:
return None
return GroundedDerivation(start=start, steps=tuple(steps))
def _same_key(a: StateTransition, b: StateTransition) -> bool:
"""True iff two transitions mutate the same scoped state key."""
return a.key == b.key