Merge pull request 'feat(adr-0249): P4 turn-program compiler + chained-relaxation executor' (#69) from feat/adr-0249-p4-turn-program into main
Reviewed-on: #69
This commit is contained in:
commit
805752db72
2 changed files with 404 additions and 0 deletions
243
evals/turn_program.py
Normal file
243
evals/turn_program.py
Normal file
|
|
@ -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
|
||||
161
tests/test_adr_0249_turn_program.py
Normal file
161
tests/test_adr_0249_turn_program.py
Normal file
|
|
@ -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)
|
||||
Loading…
Reference in a new issue