core-labs/core
core-labs/core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

feat(adr-0250): T2a multi-register executor (multi-entity arithmetic)

Browse source 
Multi-entity arithmetic as a product of independent conformal lines: one
register (null point) per entity; per-register affine ops reuse Tier-1 transport;
transfer = coupled T-k/T+k on disjoint registers (exact by P1 per-line exactness).

Transactional atomicity: immutable register mapping, prepare->validate->commit —
both candidate states relaxed into locals, validated (both converged AND
conservation), then a NEW mapping only on full success. Failure anywhere = new
mapping never built, original untouched (no partial mutation, no rollback); abort
raises typed MultiRegisterError and aborts the whole program (fail-closed).

Conservation pin = hard-reject, RELATIVE (|after-before| <= rtol*max(1,|before|)):
decoded-quantity error scales with magnitude (found via gma-050 chaining to 222),
so absolute tol is wrong; relative still catches real violations (off by whole k).

Content-addressed MultiRegisterRecord chain carries entity + certificate id +
step, never a decoded value; GENESIS-linked; verify_multi_register_chain mirrors
verify_replay_chain. Tier-2a fail-closed taxonomy: total-unknown (needs summation
= 2b), derived operand, non-affine kind, non-positive scale, unit mismatch,
unknown endpoints. Off-serving (A-04).

REAL GSM8K single-entity refused holdout: 18/18 solved wrong=0 (26/50 -> 44/50).
11/11 pins green.

[Verification]: uv run python -m pytest tests/test_adr_0250_multi_register.py -q
This commit is contained in:
Shay 2026-07-18 14:13:55 -07:00
parent 97b2f8b7d0
commit 87f87c9b21
2 changed files with 501 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

308
evals/multi_register_program.py Normal file
View file

@ -0,0 +1,308 @@
"""evals.multi_register_program — Tier-2a multi-entity arithmetic (ADR-0250).
Multi-entity arithmetic as a **product of independent conformal lines**: each
entity is its own register (its own null point / its own relaxation). Per-register
ops (add/subtract/multiply/divide by a constant) reuse the Tier-1 transport; a
transfer is a **coupled pair of translators** — `T₋ₖ` on the actor, `T₊ₖ` on the
target — acting on disjoint registers, so exactness across the independent null
points is inherited from Tier-1 (ADR-0249 P1).
**Transactional atomicity.** Registers are an immutable snapshot. A transfer
runs prepare → validate → commit: both candidate states are relaxed into locals,
then validated (both converged AND the conservation pin), then — only on full
success — a *new* register mapping is produced. A failure anywhere means the new
mapping is never built, so the original stands untouched: no partial-mutation
window, no rollback. Any abort raises a typed `MultiRegisterError` and aborts the
whole program (fail-closed — a partial answer is never emitted, preserving
wrong=0).
**Conservation pin (hard-reject, relative).** A transfer must satisfy
`Σ decode(after) ≡ Σ decode(before)` within a *relative* tolerance
(`|Δ| ≤ rtol·max(1,|before|)`) — decoded-quantity error scales with magnitude, so
an absolute tolerance is wrong for large chains; a genuine non-conserving transfer
is off by a whole `k` and is still caught. Failure is an algebraic failure, not a
scope miss: fail closed.
Tier-2a scope: single-accumulator-per-entity affine + constant-operand transfers,
positive scale, matching units on add/subtract. A total-like unknown
(`unknown.entity is None`) is refused here — it needs the certified summation turn
(2b). Derived operands (rate/comparison/fraction/partition) are refused. Records
are content-addressed and GENESIS-linked (the Ring-2 chain-integrity pattern),
carrying the entity + certificate id + step, never a decoded value. Off-serving
(A-04); 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, Quantity
__all__ = [
"MultiRegisterError",
"RegisterTurn",
"MultiRegisterProgram",
"MultiRegisterRecord",
"MultiRegisterOutcome",
"compile_multi_register_program",
"execute_multi_register_program",
"verify_multi_register_chain",
]
_AFFINE = frozenset({"add", "subtract", "multiply", "divide"})
_CONSERVATION_RTOL = 1e-6
class MultiRegisterError(ValueError):
"""Typed, fail-closed refusal for multi-register compilation/execution."""
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:
return hashlib.sha256(
json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
).hexdigest()
@dataclass(frozen=True, slots=True)
class RegisterTurn:
"""One program step: a per-register affine op or a coupled transfer."""
kind: str
actor: str
scale: float
offset: float
operand: float
target: str | None = None
@dataclass(frozen=True, slots=True)
class MultiRegisterProgram:
seeds: tuple[tuple[str, float], ...] # (entity, seed) in graph order
turns: tuple[RegisterTurn, ...]
answer_entity: str
answer_unit: str
@dataclass(frozen=True, slots=True)
class MultiRegisterRecord:
"""Content-addressed, GENESIS-linked record of one certified register turn."""
sequence_index: int
entity: str
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),
"entity": self.entity,
"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 MultiRegisterOutcome:
answer: float
answer_unit: str
records: tuple[MultiRegisterRecord, ...]
certified: bool
def _quantity(op) -> Quantity:
if not isinstance(op.operand, Quantity):
raise MultiRegisterError(
"operand_not_constant_quantity", kind=op.kind, operand_type=type(op.operand).__name__
)
return op.operand
def compile_multi_register_program(graph: MathProblemGraph) -> MultiRegisterProgram:
"""Compile a multi-entity affine `MathProblemGraph` into a register program.
Fail-closed outside Tier-2a: total-like unknown (needs the summation turn),
derived operands, non-affine/non-transfer kinds, non-positive scale, unit
mismatch, unknown transfer endpoints.
"""
if not graph.initial_state:
raise MultiRegisterError("no_initial_state")
seeds: list[tuple[str, float]] = []
units: dict[str, str] = {}
for possession in graph.initial_state:
entity = possession.entity
if entity in units:
raise MultiRegisterError("duplicate_initial_entity", entity=entity)
units[entity] = possession.quantity.unit
seeds.append((entity, float(possession.quantity.value)))
answer_entity = graph.unknown.entity
if answer_entity is None:
raise MultiRegisterError("total_unknown_requires_summation")
if answer_entity not in units:
raise MultiRegisterError("unknown_entity_not_a_register", entity=answer_entity)
turns: list[RegisterTurn] = []
for op in graph.operations:
if op.actor not in units:
raise MultiRegisterError("actor_not_a_register", actor=op.actor)
if op.kind == "transfer":
if op.target not in units:
raise MultiRegisterError("transfer_target_not_a_register", target=op.target)
operand = _quantity(op)
value = float(operand.value)
if not math.isfinite(value):
raise MultiRegisterError("operand_not_finite", kind="transfer")
turns.append(
RegisterTurn("transfer", op.actor, 1.0, -value, value, target=op.target)
)
continue
if op.kind not in _AFFINE:
raise MultiRegisterError("operation_kind_out_of_scope", kind=op.kind)
if op.target is not None:
raise MultiRegisterError("affine_op_has_target", kind=op.kind, target=op.target)
operand = _quantity(op)
value = float(operand.value)
if not math.isfinite(value):
raise MultiRegisterError("operand_not_finite", kind=op.kind)
if op.kind in ("add", "subtract"):
if operand.unit != units[op.actor]:
raise MultiRegisterError(
"unit_mismatch", entity=op.actor, expected=units[op.actor], got=operand.unit
)
offset = value if op.kind == "add" else -value
turns.append(RegisterTurn(op.kind, op.actor, 1.0, offset, value))
elif op.kind == "multiply":
if value <= 0.0:
raise MultiRegisterError("non_positive_scale", kind="multiply", value=value)
turns.append(RegisterTurn("multiply", op.actor, value, 0.0, value))
else: # divide
if value <= 0.0:
raise MultiRegisterError("non_positive_scale", kind="divide", value=value)
turns.append(RegisterTurn("divide", op.actor, 1.0 / value, 0.0, value))
return MultiRegisterProgram(
seeds=tuple(seeds),
turns=tuple(turns),
answer_entity=answer_entity,
answer_unit=graph.unknown.unit,
)
def _unit(vector: np.ndarray) -> np.ndarray:
return (vector / np.linalg.norm(vector)).astype(np.float64)
def _relax_to(state: np.ndarray, target: np.ndarray):
result = relax_to_ground(state, compile_quadratic_well(_unit(target)), require_converged=False)
return result.psi_steady, result.certificate
def execute_multi_register_program(program: MultiRegisterProgram) -> MultiRegisterOutcome:
"""Thread an immutable register mapping through the turns; decode once.
Fail-closed: a non-converged relaxation or a conservation violation raises
and aborts the whole program — no partial-mutation, no partial answer.
"""
registers = {entity: _unit(embed_quantity(seed)) for entity, seed in program.seeds}
records: list[MultiRegisterRecord] = []
prev = GENESIS_DIGEST
index = 0
for turn in program.turns:
if turn.kind == "transfer":
target_entity = turn.target
if target_entity is None: # invariant: compile sets a target for transfers
raise MultiRegisterError("transfer_missing_target", actor=turn.actor)
# PREPARE — candidate states into locals; registers untouched.
actor_target = translate_quantity(registers[turn.actor], -turn.operand)
target_target = translate_quantity(registers[target_entity], +turn.operand)
actor_new, actor_cert = _relax_to(registers[turn.actor], actor_target)
target_new, target_cert = _relax_to(registers[target_entity], target_target)
# VALIDATE — both certified, then conservation (relative, hard-reject).
if not (actor_cert.converged and target_cert.converged):
raise MultiRegisterError(
"transfer_nonconverged", actor=turn.actor, target=target_entity
)
before = decode_quantity(registers[turn.actor]) + decode_quantity(registers[target_entity])
after = decode_quantity(actor_new) + decode_quantity(target_new)
if abs(after - before) > _CONSERVATION_RTOL * max(1.0, abs(before)):
raise MultiRegisterError(
"conservation_violation", before=before, after=after
)
# COMMIT — new mapping + two coupled records, atomically.
registers = {**registers, turn.actor: actor_new, target_entity: target_new}
for entity, cert, signed in (
(turn.actor, actor_cert, -turn.operand),
(target_entity, target_cert, +turn.operand),
):
record = MultiRegisterRecord(
index, entity, cert.certificate_id, cert.converged, ("transfer", 1.0, signed), prev
)
prev = record.record_digest()
records.append(record)
index += 1
else:
target = _unit(
translate_quantity(
dilate_quantity(registers[turn.actor], -math.log(turn.scale)), turn.offset
)
)
new_state, cert = _relax_to(registers[turn.actor], target)
if not cert.converged:
raise MultiRegisterError("turn_nonconverged", entity=turn.actor, kind=turn.kind)
registers = {**registers, turn.actor: new_state}
record = MultiRegisterRecord(
index, turn.actor, cert.certificate_id, cert.converged,
(turn.kind, turn.scale, turn.offset), prev,
)
prev = record.record_digest()
records.append(record)
index += 1
answer = decode_quantity(registers[program.answer_entity])
chain = tuple(records)
return MultiRegisterOutcome(
answer=answer,
answer_unit=program.answer_unit,
records=chain,
certified=verify_multi_register_chain(chain),
)
def verify_multi_register_chain(
records: "list[MultiRegisterRecord] | tuple[MultiRegisterRecord, ...]",
) -> bool:
"""Contiguous indices from 0 and every ``prev_record_digest`` = predecessor's
recomputed digest (genesis for the first). Mirrors ``verify_turn_chain``."""
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

193
tests/test_adr_0250_multi_register.py Normal file
View file

@ -0,0 +1,193 @@
"""ADR-0250 T2a — multi-entity arithmetic (multi-register executor) pins.
Multi-entity problems solved as a product of independent conformal lines, with
coupled-translator transfers, a relative conservation pin (hard-reject), and
prepare→validate→commit atomicity. Measured on the real GSM8K single-entity
refused holdout at wrong=0.
"""
from __future__ import annotations
import dataclasses
import json
from pathlib import Path
import pytest
from generate.math_problem_graph import (
InitialPossession,
MathProblemGraph,
Operation,
Quantity,
Rate,
Unknown,
)
from evals.turn_program import TurnProgramError, compile_turn_program
from evals.multi_register_program import (
MultiRegisterError,
compile_multi_register_program,
execute_multi_register_program,
verify_multi_register_chain,
)
_DEV = Path(__file__).resolve().parents[1] / "evals" / "gsm8k_math" / "dev" / "cases.jsonl"
def _solve(graph: MathProblemGraph):
return execute_multi_register_program(compile_multi_register_program(graph))
# --- Coupled-translator transfers between registers --------------------------
def _ruth_sara(unknown_entity: str) -> MathProblemGraph:
return MathProblemGraph(
entities=("Ruth", "Sara"),
initial_state=(
InitialPossession("Ruth", Quantity(36, "cards")),
InitialPossession("Sara", Quantity(19, "cards")),
),
operations=(Operation("Ruth", "transfer", Quantity(5, "cards"), target="Sara"),),
unknown=Unknown(unknown_entity, "cards"),
)
def test_transfer_updates_both_registers() -> None:
assert abs(_solve(_ruth_sara("Ruth")).answer - 31.0) < 1e-4 # 36 - 5
assert abs(_solve(_ruth_sara("Sara")).answer - 24.0) < 1e-4 # 19 + 5
def test_multi_step_large_magnitude_relative_conservation() -> None:
# Gwen: 37 *3 *2 -> 222, then two transfers to Leo, then -26 -> 26.
# Large intermediate (222) exercises the RELATIVE conservation pin.
graph = MathProblemGraph(
entities=("Gwen", "Leo"),
initial_state=(
InitialPossession("Gwen", Quantity(37, "cards")),
InitialPossession("Leo", Quantity(90, "cards")),
),
operations=(
Operation("Gwen", "multiply", Quantity(3, "factor")),
Operation("Gwen", "multiply", Quantity(2, "factor")),
Operation("Gwen", "transfer", Quantity(100, "cards"), target="Leo"),
Operation("Gwen", "transfer", Quantity(70, "cards"), target="Leo"),
Operation("Gwen", "subtract", Quantity(26, "cards")),
),
unknown=Unknown("Gwen", "cards"),
)
outcome = _solve(graph)
assert abs(outcome.answer - 26.0) < 1e-4
assert outcome.certified is True
def test_transfer_conserves_total() -> None:
ruth = _solve(_ruth_sara("Ruth")).answer
sara = _solve(_ruth_sara("Sara")).answer
assert abs((ruth + sara) - (36 + 19)) < 1e-4 # nothing created or destroyed
# --- The real GSM8K single-entity refused holdout, wrong=0 -------------------
@pytest.fixture(scope="module")
def real_single_entity_stats():
cases = [json.loads(line) for line in _DEV.read_text().splitlines() if line.strip()]
from generate.math_problem_graph import graph_from_dict
solved = wrong = 0
for case in cases:
graph = graph_from_dict(case["ground_truth_graph"])
try:
compile_turn_program(graph) # already Tier-1
continue
except TurnProgramError:
pass
if graph.unknown.entity is None:
continue # total-like -> summation turn (2b), not T2a
outcome = _solve(graph)
if abs(outcome.answer - float(case["expected_answer"])) < 1e-4:
solved += 1
else:
wrong += 1
return solved, wrong
def test_real_single_entity_wrong_zero(real_single_entity_stats) -> None:
solved, wrong = real_single_entity_stats
assert wrong == 0 # the load-bearing claim
assert solved == 18 # coverage tracker: dev holdout single-entity subset
# --- Content-addressed chain: carries the entity, tamper-evident, no answer --
def test_chain_verifies_carries_entity_and_detects_tamper() -> None:
outcome = _solve(_ruth_sara("Ruth"))
assert verify_multi_register_chain(outcome.records) is True
assert {r.entity for r in outcome.records} == {"Ruth", "Sara"} # both legs recorded
tampered = list(outcome.records)
tampered[0] = dataclasses.replace(tampered[0], certificate_id="forged")
assert verify_multi_register_chain(tampered) is False
def test_records_carry_no_decoded_answer() -> None:
outcome = _solve(_ruth_sara("Ruth"))
fields = {f.name for f in dataclasses.fields(outcome.records[0])}
assert "answer" not in fields and "decoded" not in fields
def test_deterministic() -> None:
a = [r.record_digest() for r in _solve(_ruth_sara("Ruth")).records]
b = [r.record_digest() for r in _solve(_ruth_sara("Ruth")).records]
assert a == b
# --- Fail-closed refusals (Tier-2a scope) -----------------------------------
def test_refuses_total_unknown_pending_summation() -> None:
# "how many altogether" needs the certified summation turn (2b), not T2a.
graph = MathProblemGraph(
entities=("Ann", "Bob"),
initial_state=(
InitialPossession("Ann", Quantity(5, "apples")),
InitialPossession("Bob", Quantity(3, "apples")),
),
operations=(),
unknown=Unknown(None, "apples"),
)
with pytest.raises(MultiRegisterError):
compile_multi_register_program(graph)
def test_refuses_derived_operand() -> None:
graph = MathProblemGraph(
entities=("Ann",),
initial_state=(InitialPossession("Ann", Quantity(5, "apples")),),
operations=(Operation("Ann", "apply_rate", Rate(2.0, "dollars", "apple")),),
unknown=Unknown("Ann", "dollars"),
)
with pytest.raises(MultiRegisterError):
compile_multi_register_program(graph)
def test_refuses_non_positive_scale() -> None:
graph = MathProblemGraph(
entities=("Ann",),
initial_state=(InitialPossession("Ann", Quantity(5, "apples")),),
operations=(Operation("Ann", "multiply", Quantity(0, "factor")),),
unknown=Unknown("Ann", "apples"),
)
with pytest.raises(MultiRegisterError):
compile_multi_register_program(graph)
def test_refuses_unit_mismatch() -> None:
graph = MathProblemGraph(
entities=("Ann",),
initial_state=(InitialPossession("Ann", Quantity(5, "apples")),),
operations=(Operation("Ann", "add", Quantity(3, "oranges")),),
unknown=Unknown("Ann", "apples"),
)
with pytest.raises(MultiRegisterError):
compile_multi_register_program(graph)