diff --git a/evals/turn_program.py b/evals/turn_program.py new file mode 100644 index 00000000..67079c87 --- /dev/null +++ b/evals/turn_program.py @@ -0,0 +1,243 @@ +"""evals.turn_program — multi-step arithmetic as a certified turn program (ADR-0249 P4). + +The composition frontier: a multi-step arithmetic problem is compiled from a +``MathProblemGraph`` into an ordered *turn program* — one affine relation-well +per step — and executed as a chain of certified relaxation turns. The +accumulator flows turn-to-turn as a field STATE and is decoded only once, at the +end (anti-hollow, spike §2/§4.1): the substrate performs every step, and the +composition depth lives in the certified chain, not in matrix size. + +Ring-2 relationship (spike §4.6 follow-up, verified in-tree): the residual +protocol (``core.ports.residual_protocol.run_residual_protocol``) is +*zero-bound / non-mutating* — its stage-5 recertification refuses if the witness +moved — so it certifies the admissibility of a *fixed* state, not a state +*transition*. Arithmetic turns mutate. So the per-turn certificate is the +relaxation's own ``RelaxationCertificate``, and the tamper-evident *sequence* is +recorded here with the Ring-2 chain-integrity *pattern* (content-addressed, +``GENESIS_DIGEST``-linked, ``verify_turn_chain`` mirroring ``verify_replay_chain``) +rather than by forcing mutating turns through the zero-bound protocol. + +Tier-1 scope (ruling #1): a single-accumulator chain over add / subtract / +multiply / divide with constant ``Quantity`` operands and positive scale. +Everything else (multi-entity, transfer, rate, comparison, fraction, partition, +non-positive scale, unit mismatch) is refused and recorded, never silently +dropped. Off-serving (A-04): bridges the generate-side graph to the corridor; +never imported by chat/runtime.py. +""" +from __future__ import annotations + +import hashlib +import json +import math +from dataclasses import dataclass + +import numpy as np + +from core.physics.cognitive_lifecycle import compile_quadratic_well, relax_to_ground +from core.physics.quantity_kernel import ( + decode_quantity, + dilate_quantity, + embed_quantity, + translate_quantity, +) +from core.ports.residual_protocol import GENESIS_DIGEST +from generate.math_problem_graph import MathProblemGraph, Operation, Quantity + +__all__ = [ + "TurnProgramError", + "AffineStep", + "TurnProgram", + "TurnRecord", + "TurnProgramOutcome", + "compile_turn_program", + "execute_turn_program", + "verify_turn_chain", +] + +_AFFINE_KINDS = frozenset({"add", "subtract", "multiply", "divide"}) + + +class TurnProgramError(ValueError): + """Typed, fail-closed refusal for turn-program compilation.""" + + def __init__(self, reason: str, **disclosure: object) -> None: + self.reason = reason + self.disclosure = disclosure + detail = ", ".join(f"{k}={v!r}" for k, v in disclosure.items()) + super().__init__(f"{reason}({detail})" if detail else reason) + + +def _digest(payload: dict) -> str: + """Full SHA-256 over canonical JSON (ADR-0245 §2.3; no ``default=str``).""" + return hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + + +@dataclass(frozen=True, slots=True) +class AffineStep: + """One relation `output = scale·input + offset`, with source provenance.""" + + scale: float + offset: float + kind: str + operand: float + + +@dataclass(frozen=True, slots=True) +class TurnProgram: + """A compiled single-accumulator arithmetic program (no answer inside).""" + + seed: float + steps: tuple[AffineStep, ...] + answer_unit: str + + +@dataclass(frozen=True, slots=True) +class TurnRecord: + """One content-addressed, GENESIS-linked record of a certified turn. + + Carries the turn's ``RelaxationCertificate`` id and the step provenance — + never a decoded value. Any tamper changes ``record_digest`` and breaks the + chain from that point (``verify_turn_chain``). + """ + + sequence_index: int + certificate_id: str + converged: bool + step: tuple[str, float, float] # (kind, scale, offset) + prev_record_digest: str + + def _payload(self) -> dict: + return { + "sequence_index": int(self.sequence_index), + "certificate_id": self.certificate_id, + "converged": bool(self.converged), + "step": [self.step[0], repr(float(self.step[1])), repr(float(self.step[2]))], + "prev_record_digest": self.prev_record_digest, + } + + def record_digest(self) -> str: + return _digest(self._payload()) + + +@dataclass(frozen=True, slots=True) +class TurnProgramOutcome: + """Decoded answer + the certified, tamper-evident turn chain.""" + + answer: float + answer_unit: str + records: tuple[TurnRecord, ...] + certified: bool + + +def _quantity_operand(op: Operation) -> Quantity: + if not isinstance(op.operand, Quantity): + raise TurnProgramError( + "operand_not_constant_quantity", kind=op.kind, operand_type=type(op.operand).__name__ + ) + return op.operand + + +def compile_turn_program(graph: MathProblemGraph) -> TurnProgram: + """Compile a single-accumulator affine ``MathProblemGraph`` into a turn program. + + Fail-closed on anything outside the Tier-1 envelope; the refused shapes are + the recorded composition frontier, not silent drops. + """ + if len(graph.initial_state) != 1: + raise TurnProgramError( + "not_single_accumulator", possessions=len(graph.initial_state) + ) + seed_possession = graph.initial_state[0] + accumulator = seed_possession.entity + current_unit = seed_possession.quantity.unit + seed = float(seed_possession.quantity.value) + + steps: list[AffineStep] = [] + for op in graph.operations: + if op.actor != accumulator: + raise TurnProgramError("multi_actor_operation", actor=op.actor) + if op.target is not None: + raise TurnProgramError("multi_entity_operation", kind=op.kind, target=op.target) + if op.kind not in _AFFINE_KINDS: + raise TurnProgramError("operation_kind_out_of_affine_scope", kind=op.kind) + operand = _quantity_operand(op) + value = float(operand.value) + if not math.isfinite(value): + raise TurnProgramError("operand_not_finite", kind=op.kind) + if op.kind == "add": + if operand.unit != current_unit: + raise TurnProgramError("unit_mismatch", expected=current_unit, got=operand.unit) + steps.append(AffineStep(scale=1.0, offset=value, kind="add", operand=value)) + elif op.kind == "subtract": + if operand.unit != current_unit: + raise TurnProgramError("unit_mismatch", expected=current_unit, got=operand.unit) + steps.append(AffineStep(scale=1.0, offset=-value, kind="subtract", operand=value)) + elif op.kind == "multiply": + if value <= 0.0: + raise TurnProgramError("non_positive_scale", kind="multiply", value=value) + steps.append(AffineStep(scale=value, offset=0.0, kind="multiply", operand=value)) + else: # divide + if value <= 0.0: + raise TurnProgramError("non_positive_scale", kind="divide", value=value) + steps.append(AffineStep(scale=1.0 / value, offset=0.0, kind="divide", operand=value)) + + return TurnProgram(seed=seed, steps=tuple(steps), answer_unit=graph.unknown.unit) + + +def _unit(vector: np.ndarray) -> np.ndarray: + return (vector / np.linalg.norm(vector)).astype(np.float64) + + +def execute_turn_program(program: TurnProgram) -> TurnProgramOutcome: + """Chain the turns: relax step by step, decode only the final state. + + The accumulator state flows without being decoded mid-chain (anti-hollow). + Each turn contributes its real ``RelaxationCertificate`` to a content-addressed + chain; ``certified`` holds iff every turn converged and the chain is intact. + """ + psi = _unit(embed_quantity(program.seed)) + records: list[TurnRecord] = [] + prev = GENESIS_DIGEST + all_converged = True + + for index, step in enumerate(program.steps): + # Transport the STATE (not a decoded value) by the step's structure. + target = _unit(translate_quantity(dilate_quantity(psi, -math.log(step.scale)), step.offset)) + hamiltonian = compile_quadratic_well(target) + result = relax_to_ground(psi, hamiltonian, require_converged=False) + psi = result.psi_steady + certificate = result.certificate + all_converged = all_converged and bool(certificate.converged) + record = TurnRecord( + sequence_index=index, + certificate_id=certificate.certificate_id, + converged=bool(certificate.converged), + step=(step.kind, step.scale, step.offset), + prev_record_digest=prev, + ) + prev = record.record_digest() + records.append(record) + + answer = decode_quantity(psi) + chain = tuple(records) + return TurnProgramOutcome( + answer=answer, + answer_unit=program.answer_unit, + records=chain, + certified=all_converged and verify_turn_chain(chain), + ) + + +def verify_turn_chain(records: "list[TurnRecord] | tuple[TurnRecord, ...]") -> bool: + """True iff indices are contiguous from 0 and every ``prev_record_digest`` + equals the recomputed digest of its predecessor (genesis for the first). + Mirrors ``residual_protocol.verify_replay_chain`` for the turn ledger.""" + for i, record in enumerate(records): + if record.sequence_index != i: + return False + expected_prev = GENESIS_DIGEST if i == 0 else records[i - 1].record_digest() + if record.prev_record_digest != expected_prev: + return False + return True diff --git a/tests/test_adr_0249_turn_program.py b/tests/test_adr_0249_turn_program.py new file mode 100644 index 00000000..4fda4d65 --- /dev/null +++ b/tests/test_adr_0249_turn_program.py @@ -0,0 +1,161 @@ +"""ADR-0249 P4 — turn-program compiler + chained-relaxation executor pins. + +Multi-step arithmetic is solved as a *turn program*: an ordered sequence of +affine relation-wells, each solved in one certified relaxation turn, with the +accumulator flowing turn-to-turn as a field STATE that is never decoded until +the end (anti-hollow, spike §2/§4.1). The sequence is recorded as a +content-addressed, GENESIS-linked turn chain (the Ring-2 chain-integrity +pattern, applied to the mutating arithmetic turns the zero-bound residual +protocol cannot itself carry). +""" +from __future__ import annotations + +import dataclasses + +import pytest + +from generate.math_problem_graph import ( + InitialPossession, + MathProblemGraph, + Operation, + Quantity, + Unknown, +) + +from evals.turn_program import ( + TurnProgramError, + compile_turn_program, + execute_turn_program, + verify_turn_chain, +) + + +def _graph(seed, steps, unit="apples", entity="tom"): + """Single-accumulator graph: seed then (kind, operand_value) steps.""" + ops = tuple( + Operation(entity, kind, Quantity(val, unit if kind in ("add", "subtract") else "factor")) + for kind, val in steps + ) + return MathProblemGraph( + entities=(entity,), + initial_state=(InitialPossession(entity, Quantity(seed, unit)),), + operations=ops, + unknown=Unknown(entity, unit), + ) + + +def _solve(seed, steps, unit="apples"): + program = compile_turn_program(_graph(seed, steps, unit)) + return execute_turn_program(program) + + +# --- Multi-step arithmetic by chained certified relaxation ------------------- + + +@pytest.mark.parametrize( + ("seed", "steps", "gold"), + [ + (5.0, [("add", 3.0), ("multiply", 2.0), ("subtract", 4.0)], 12.0), # ((5+3)*2)-4 + (10.0, [("divide", 2.0), ("add", 7.0), ("multiply", 3.0)], 36.0), # (10/2+7)*3 + (4.0, [("multiply", 3.0), ("add", 5.0)], 17.0), # 4*3+5 + (100.0, [("subtract", 40.0), ("divide", 4.0)], 15.0), # (100-40)/4 + (7.0, [("add", 0.0)], 7.0), # identity step + ], +) +def test_multi_step_arithmetic(seed, steps, gold) -> None: + outcome = _solve(seed, steps) + assert abs(outcome.answer - gold) < 1e-4 + assert outcome.certified is True + assert len(outcome.records) == len(steps) + + +def test_answer_unit_from_graph() -> None: + assert _solve(5.0, [("add", 3.0)], unit="dollars").answer_unit == "dollars" + + +# --- Certified turn chain: content-addressed, GENESIS-linked, tamper-evident -- + + +def test_turn_chain_verifies() -> None: + outcome = _solve(5.0, [("add", 3.0), ("multiply", 2.0)]) + assert verify_turn_chain(outcome.records) is True + assert tuple(r.sequence_index for r in outcome.records) == (0, 1) + + +def test_turn_chain_detects_tamper() -> None: + # Tampering a non-terminal record breaks its successor's prev-link (the + # standard hash-chain property; the terminal record's self-integrity comes + # from deterministic re-execution, not a downstream pointer). + outcome = _solve(5.0, [("add", 3.0), ("multiply", 2.0), ("subtract", 4.0)]) + tampered = list(outcome.records) + tampered[0] = dataclasses.replace(tampered[0], certificate_id="forged") + assert verify_turn_chain(tampered) is False + + +def test_records_are_deterministic() -> None: + a = _solve(5.0, [("add", 3.0), ("multiply", 2.0)]) + b = _solve(5.0, [("add", 3.0), ("multiply", 2.0)]) + assert [r.record_digest() for r in a.records] == [r.record_digest() for r in b.records] + + +def test_records_carry_no_decoded_answer() -> None: + # Anti-hollow: a turn record exposes the constraint provenance (certificate + # + step), never a decoded intermediate value. + outcome = _solve(5.0, [("add", 3.0), ("multiply", 2.0)]) + record_fields = {f.name for f in dataclasses.fields(outcome.records[0])} + assert "answer" not in record_fields + assert "decoded" not in record_fields + + +# --- Compilation is a pure, refusal-first mapping ---------------------------- + + +def test_compile_is_pure_no_answer() -> None: + program = compile_turn_program(_graph(5.0, [("add", 3.0), ("multiply", 2.0)])) + assert program.seed == 5.0 + assert tuple(s.kind for s in program.steps) == ("add", "multiply") + assert not hasattr(program, "answer") + + +# --- Fail-closed refusals (Tier-1 affine single-accumulator scope) ----------- + + +def test_refuses_multi_entity_graph() -> None: + graph = MathProblemGraph( + entities=("tom", "sue"), + initial_state=( + InitialPossession("tom", Quantity(5, "apples")), + InitialPossession("sue", Quantity(2, "apples")), + ), + operations=(), + unknown=Unknown("tom", "apples"), + ) + with pytest.raises(TurnProgramError): + compile_turn_program(graph) + + +def test_refuses_transfer_operation() -> None: + graph = MathProblemGraph( + entities=("tom", "sue"), + initial_state=(InitialPossession("tom", Quantity(5, "apples")),), + operations=(Operation("tom", "transfer", Quantity(2, "apples"), target="sue"),), + unknown=Unknown("tom", "apples"), + ) + with pytest.raises(TurnProgramError): + compile_turn_program(graph) + + +def test_refuses_non_positive_scale() -> None: + with pytest.raises(TurnProgramError): + compile_turn_program(_graph(5.0, [("multiply", 0.0)])) + + +def test_refuses_add_unit_mismatch() -> None: + graph = MathProblemGraph( + entities=("tom",), + initial_state=(InitialPossession("tom", Quantity(5, "apples")),), + operations=(Operation("tom", "add", Quantity(3, "oranges")),), + unknown=Unknown("tom", "apples"), + ) + with pytest.raises(TurnProgramError): + compile_turn_program(graph)