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

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

evals/multi_register_program.py

@@ -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

tests/test_adr_0250_multi_register.py

@@ -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)