core/evals/generalized_lift_instrument.py
Shay b87cd46cf7 chore(adr-0250): ratify Accepted + fold in PR #74 review findings
Ratification: ADR-0250 Status Proposed->Accepted; §10 ruling record stamped
(Shay, citing the acceptance-evidence pack as-is; full-holdout 50/50 wrong=0
affirmed; next arc = seal dev-holdout-2, dev-1 pinned as regression floor).

Review findings (Shay, PR #74, non-blocking):
1. Instrument module docstring updated from 'Three deterministic domains' +
   'GSM8K NOT ingestible / no compiler' to the four-domain post-0250 reality
   (arithmetic-chain solves the full dev holdout; frontier surfaced in scope).
2. Restored per-case refusal reasons: _corridor_and_baseline now raises
   MultiRegisterError (not bare None); the domain loop records refusal.reason
   and distinguishes graph_parse_failed. 'Recorded, never silently dropped'
   stays literal (0 rows at refused=0, but the path is honest).
3. Tightened the Tier-1 try to wrap compile_turn_program only (execute outside),
   so a future typed execute raise surfaces as a Tier-1 failure instead of
   silently rerouting into Tier-2 and masking a regression; added the note that
   Tier-2 fails closed on non-convergence while Tier-1 records-only, gold
   comparison binding wrong=0 on both.

10/10 instrument pins green.

[Verification]: uv run python -m pytest tests/test_adr_0249_arithmetic_lift.py tests/test_generalized_lift_instrument.py -q
2026-07-18 15:07:29 -07:00

637 lines
24 KiB
Python

"""Generalized-lift instrument — corridor vs symbolic baseline (seam S4, OFF-SERVING).
Instrument-first doctrine (ADR-0190 lesson): this module DECIDES the
"meaningful, generalized lift without overfitting" question instead of
narrating it. Four deterministic domains, identical compiled problems for
both paths, independent gold, and an honest-NULL protocol: if the corridor
adds no delta, the report says so.
Domains (corridor's honest ingestible surface):
* ``propositional`` — entailment on an enumerated case family. Corridor:
:func:`propositional_entails` (exact ground-energy verdicts). Baseline: the
deductive flagship (ROBDD, ``generate.logic_equivalence`` — P ⊨ C iff
(P and C) ≡ P). Gold: independent brute-force truth tables computed here.
The wrong=0 guard binds the corridor on this domain.
* ``constrained-recognition`` — ambiguous two-mode ingress relaxed under the
problem well, then ARTICULATED via the seam-S1 readback; baseline is the
constraint-blind ingress argmax over the same vocabulary. The measured
delta isolates the relax+readback stages' contribution — an eigensolver
baseline WITH access to H would reach parity (disclosed in notes; this
domain measures loop integrity, not open-ended capability).
* ``multimodal-completion`` — the sensorium corridor pattern (audio partial →
full audio+vision percept), scored on whether articulation names BOTH
constituent percepts; baseline articulates the raw partial ingress.
* ``arithmetic-chain`` — real GSM8K dev-holdout problems routed to the
reader→Hamiltonian compiler: Tier-1 single-accumulator (ADR-0249) and Tier-2
multi-entity / transfer / certified summation (ADR-0250), each vs a symbolic
fold of the SAME compiled program. The full dev holdout solves wrong=0
(PARITY — the field matches arithmetic, it does not beat it).
Scope limitations are RECORDED, never silently dropped (no-silent-caps): the
reader→Hamiltonian compiler now solves the full dev holdout; the remaining
frontier (derived-operand transfers, non-affine kinds, >5-atom deduction) sits
at 0 cases on this holdout and is surfaced in the report's ``scope_limitations``.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from itertools import product
from pathlib import Path
from typing import Any, Sequence
import numpy as np
from algebra.rotor import make_rotor_from_angle
from core.physics.cognitive_lifecycle import (
CognitiveLifecycleEngine,
PropositionalProblem,
compile_quadratic_well,
egress_gate,
ingest_context,
propositional_entails,
relax_to_ground,
)
from core.physics.linguistic_readback import (
ReadbackRefusal,
articulate_outcome,
linguistic_readback,
)
from core.physics.sensorium_wave_feed import ModalityPacket
from core.physics.wave_manifold import WaveManifold
from generate.logic_equivalence import Verdict, check_equivalence
from generate.math_problem_graph import MathGraphError, graph_from_dict
from vocab.manifold import VocabManifold
from evals.turn_program import (
TurnProgramError,
compile_turn_program,
execute_turn_program,
)
from evals.multi_register_program import (
MultiRegisterError,
compile_multi_register_program,
execute_multi_register_program,
)
__all__ = [
"DomainOutcome",
"LiftInstrumentReport",
"run_generalized_lift_instrument",
"run_propositional_domain",
"run_recognition_domain",
"run_multimodal_domain",
"run_arithmetic_chain_domain",
]
# Hot-band energy axes (existing E3/E4 precedent; caller-supplied, never invented).
_HOT_ENERGY: dict[str, Any] = {
"convergence_density": 8,
"activation_count": 8,
"current_cycle": 1,
"last_activation_cycle": 1,
"morphology_features": {"mood": "imperative"},
}
_MIN_RESONANCE = 0.4
_LIFT, _PARITY, _DEFICIT = "LIFT", "PARITY", "DEFICIT"
@dataclass(frozen=True, slots=True)
class DomainOutcome:
"""One domain's corridor-vs-baseline scorecard (per-case rows disclosed)."""
domain_id: str
n_cases: int
corridor_correct: int
corridor_wrong: int
corridor_refused: int
baseline_correct: int
baseline_wrong: int
baseline_refused: int
notes: tuple[str, ...]
cases: tuple[dict[str, Any], ...]
@property
def delta_correct(self) -> int:
return self.corridor_correct - self.baseline_correct
@property
def verdict(self) -> str:
if self.delta_correct > 0:
return _LIFT
return _PARITY if self.delta_correct == 0 else _DEFICIT
def as_dict(self) -> dict[str, Any]:
return {
"domain_id": self.domain_id,
"n_cases": self.n_cases,
"corridor": {
"correct": self.corridor_correct,
"wrong": self.corridor_wrong,
"refused": self.corridor_refused,
},
"baseline": {
"correct": self.baseline_correct,
"wrong": self.baseline_wrong,
"refused": self.baseline_refused,
},
"delta_correct": self.delta_correct,
"verdict": self.verdict,
"notes": list(self.notes),
"cases": list(self.cases),
}
@dataclass(frozen=True, slots=True)
class LiftInstrumentReport:
outcomes: tuple[DomainOutcome, ...]
wrong_zero_guard_held: bool
honest_null: bool
scope_limitations: tuple[str, ...]
def as_dict(self) -> dict[str, Any]:
return {
"outcomes": [o.as_dict() for o in self.outcomes],
"wrong_zero_guard_held": self.wrong_zero_guard_held,
"honest_null": self.honest_null,
"scope_limitations": list(self.scope_limitations),
}
# --- Domain A: propositional entailment ------------------------------------------------
Literal = tuple[str, bool]
Clause = tuple[Literal, ...]
# (case_id, atoms, premise clauses (CNF), conclusion clause (single literal))
_PROP_CASES: tuple[tuple[str, tuple[str, ...], tuple[Clause, ...], Literal], ...] = (
("modus-ponens", ("a", "b"), ((("a", True),), (("a", False), ("b", True))), ("b", True)),
(
"chain-3",
("a", "b", "c"),
((("a", True),), (("a", False), ("b", True)), (("b", False), ("c", True))),
("c", True),
),
("disj-not-entailed", ("a", "b"), ((("a", True), ("b", True)),), ("a", True)),
("unsat-ex-falso", ("a", "b"), ((("a", True),), (("a", False),)), ("b", True)),
(
"neg-conclusion",
("a", "b"),
((("a", True),), (("a", False), ("b", False))),
("b", False),
),
("no-information", ("a", "b"), ((("a", True),),), ("b", True)),
(
"resolution",
("a", "b", "c"),
(
(("a", True), ("b", True)),
(("a", False), ("c", True)),
(("b", False), ("c", True)),
),
("c", True),
),
(
"contrapositive",
("a", "b"),
((("a", False), ("b", True)), (("b", False),)),
("a", False),
),
("premise-restates", ("a", "b"), ((("a", True),),), ("a", True)),
("wide-disj-not-entailed", ("a", "b", "c"), ((("a", True), ("b", True), ("c", True)),), ("c", True)),
)
def _lit_formula(lit: Literal) -> str:
atom, positive = lit
return atom if positive else f"(not {atom})"
def _clauses_formula(clauses: Sequence[Clause]) -> str:
return " and ".join("(" + " or ".join(_lit_formula(l) for l in clause) + ")" for clause in clauses)
def _truth_table_entailed(
atoms: Sequence[str], clauses: Sequence[Clause], conclusion: Literal
) -> bool:
"""Independent gold: every model of the premises satisfies the conclusion."""
for values in product((False, True), repeat=len(atoms)):
env = dict(zip(atoms, values))
if all(any(env[a] == pos for a, pos in clause) for clause in clauses):
atom, positive = conclusion
if env[atom] != positive:
return False
return True
def run_propositional_domain() -> DomainOutcome:
corridor_correct = corridor_wrong = corridor_refused = 0
baseline_correct = baseline_wrong = baseline_refused = 0
rows: list[dict[str, Any]] = []
for case_id, atoms, clauses, conclusion in _PROP_CASES:
gold = _truth_table_entailed(atoms, clauses, conclusion)
corridor_entailed = propositional_entails(
PropositionalProblem(atoms=atoms, clauses=clauses), (conclusion,)
).entailed
if corridor_entailed == gold:
corridor_correct += 1
else:
corridor_wrong += 1
premises_f = _clauses_formula(clauses)
conjunction_f = f"({premises_f}) and ({_lit_formula(conclusion)})"
robdd = check_equivalence(conjunction_f, premises_f)
if robdd.verdict is Verdict.REFUSED:
baseline_refused += 1
baseline_entailed: bool | None = None
else:
baseline_entailed = robdd.verdict is Verdict.EQUIVALENT
if baseline_entailed == gold:
baseline_correct += 1
else:
baseline_wrong += 1
rows.append(
{
"case_id": case_id,
"gold_entailed": gold,
"corridor_entailed": corridor_entailed,
"baseline_entailed": baseline_entailed,
}
)
return DomainOutcome(
domain_id="propositional",
n_cases=len(_PROP_CASES),
corridor_correct=corridor_correct,
corridor_wrong=corridor_wrong,
corridor_refused=corridor_refused,
baseline_correct=baseline_correct,
baseline_wrong=baseline_wrong,
baseline_refused=baseline_refused,
notes=(
"Baseline is the deductive flagship (ROBDD); PARITY here is the "
"expected honest outcome — both paths are exact on this regime.",
"wrong=0 guard binds the corridor on this domain.",
),
cases=tuple(rows),
)
# --- Domain B: constrained recognition + articulation ----------------------------------
_RECOGNITION_GRID: tuple[tuple[int, float], ...] = tuple(
(plane, angle) for plane in (6, 7, 8) for angle in (0.4, 0.8, 1.2)
)
def _grid_word(plane: int, angle: float) -> str:
return f"mode-p{plane}-a{int(round(angle * 10))}"
def _recognition_vocab() -> tuple[VocabManifold, tuple[np.ndarray, ...]]:
vocab = VocabManifold()
versors: list[np.ndarray] = []
for plane, angle in _RECOGNITION_GRID:
v = np.asarray(make_rotor_from_angle(angle, bivector_idx=plane), dtype=np.float64)
vocab.add(_grid_word(plane, angle), v)
versors.append(v)
return vocab, tuple(versors)
def _argmax_word(psi: np.ndarray, vocab: VocabManifold, manifold: WaveManifold) -> str:
best_score, best_idx = -np.inf, -1
for i in range(len(vocab)):
score = float(manifold.phase_correlation(psi, np.asarray(vocab.get_versor_at(i), dtype=np.float64))) / 2.0
if score > best_score:
best_score, best_idx = score, i
return vocab.get_word_at(best_idx)
def run_recognition_domain() -> DomainOutcome:
vocab, versors = _recognition_vocab()
manifold = WaveManifold()
engine = CognitiveLifecycleEngine()
n = len(_RECOGNITION_GRID)
corridor_correct = corridor_wrong = corridor_refused = 0
baseline_correct = baseline_wrong = 0
rows: list[dict[str, Any]] = []
for i, (plane, angle) in enumerate(_RECOGNITION_GRID):
target_word = _grid_word(plane, angle)
target = versors[i]
distractor = versors[(i + 1) % n]
packets = (
ModalityPacket(modality_id="mix:target", coefficients=0.45 * target),
ModalityPacket(modality_id="mix:distractor", coefficients=0.55 * distractor),
)
domain_id = f"recognition:{target_word}"
baseline_word = _argmax_word(
ingest_context(packets, domain_id).psi, vocab, manifold
)
if baseline_word == target_word:
baseline_correct += 1
else:
baseline_wrong += 1
row: dict[str, Any] = {
"case_id": target_word,
"baseline_word": baseline_word,
}
try:
outcome = engine.solve(
packets,
domain_id,
compile_quadratic_well(target),
energy_inputs=_HOT_ENERGY,
)
readback, roundtrip = articulate_outcome(
outcome, vocab, min_resonance=_MIN_RESONANCE, max_tokens=1
)
corridor_word = readback.tokens[0].word
row["corridor_word"] = corridor_word
row["roundtrip_agreement"] = roundtrip.agreement
if corridor_word == target_word:
corridor_correct += 1
else:
corridor_wrong += 1
except ReadbackRefusal as exc:
corridor_refused += 1
row["corridor_word"] = None
row["corridor_refusal"] = exc.reason
rows.append(row)
return DomainOutcome(
domain_id="constrained-recognition",
n_cases=n,
corridor_correct=corridor_correct,
corridor_wrong=corridor_wrong,
corridor_refused=corridor_refused,
baseline_correct=baseline_correct,
baseline_wrong=baseline_wrong,
baseline_refused=0,
notes=(
"Baseline is the constraint-blind ingress argmax over the same "
"vocabulary; the delta isolates the relax+readback stages.",
"An eigensolver baseline WITH access to H would reach parity — "
"this domain measures loop integrity, not open-ended capability.",
),
cases=tuple(rows),
)
# --- Domain C: multimodal completion ---------------------------------------------------
def run_multimodal_domain() -> DomainOutcome:
from evals.adr_0243_cognitive_lifecycle import _fixed_audio_tone, _fixed_vision_tile
from core.physics.sensorium_wave_feed import packet_from_compilation_unit
from sensorium.audio.compiler import AudioCompiler
from sensorium.vision import VisionCompiler
audio_unit = AudioCompiler().compile(_fixed_audio_tone(24_000, 0.25, 440.0), 24_000)
vision_unit = VisionCompiler().compile_tile(_fixed_vision_tile())
audio_pkt = packet_from_compilation_unit("audio", audio_unit)
vision_pkt = packet_from_compilation_unit("vision", vision_unit)
vocab = VocabManifold()
vocab.add("audio-tone", np.asarray(audio_pkt.coefficients, dtype=np.float64))
vocab.add("vision-tile", np.asarray(vision_pkt.coefficients, dtype=np.float64))
manifold = WaveManifold()
full = ingest_context((audio_pkt, vision_pkt), "multimodal-completion")
partial = ingest_context((audio_pkt,), "multimodal-completion")
expected_words = {"audio-tone", "vision-tile"}
def _resonant_words(psi: np.ndarray) -> set[str]:
found = set()
for i in range(len(vocab)):
score = (
float(
manifold.phase_correlation(
psi, np.asarray(vocab.get_versor_at(i), dtype=np.float64)
)
)
/ 2.0
)
if score >= _MIN_RESONANCE:
found.add(vocab.get_word_at(i))
return found
baseline_words = _resonant_words(partial.psi)
baseline_correct = int(baseline_words == expected_words)
corridor_correct = corridor_wrong = corridor_refused = 0
row: dict[str, Any] = {
"case_id": "audio-partial-to-full",
"baseline_words": sorted(baseline_words),
}
result = relax_to_ground(partial.psi, compile_quadratic_well(full.psi))
verdict = egress_gate(result.psi_steady, result.certificate, **_HOT_ENERGY)
try:
readback = linguistic_readback(
result.psi_steady,
result.certificate,
verdict,
vocab,
min_resonance=_MIN_RESONANCE,
max_tokens=2,
)
corridor_words = set(readback.words)
row["corridor_words"] = sorted(corridor_words)
if corridor_words == expected_words:
corridor_correct = 1
else:
corridor_wrong = 1
except ReadbackRefusal as exc:
corridor_refused = 1
row["corridor_refusal"] = exc.reason
return DomainOutcome(
domain_id="multimodal-completion",
n_cases=1,
corridor_correct=corridor_correct,
corridor_wrong=corridor_wrong,
corridor_refused=corridor_refused,
baseline_correct=baseline_correct,
baseline_wrong=1 - baseline_correct,
baseline_refused=0,
notes=(
"Correct = articulation names BOTH constituent percepts; baseline "
"articulates the raw audio-only partial ingress.",
),
cases=(row,),
)
# --- Composed instrument ---------------------------------------------------------------
# --- Domain D: multi-step arithmetic on the real GSM8K dev holdout (ADR-0249) ----------
# The sealed GSM8K dev holdout — real problems, never the templated cases.
_DEV_HOLDOUT = Path(__file__).parent / "gsm8k_math" / "dev" / "cases.jsonl"
def _load_dev_cases() -> tuple[dict[str, Any], ...]:
if not _DEV_HOLDOUT.exists():
return ()
with _DEV_HOLDOUT.open() as handle:
return tuple(json.loads(line) for line in handle if line.strip())
def _symbolic_fold(seed: float, steps) -> float:
"""The baseline: solve the SAME compiled turn program by plain arithmetic.
Corridor and baseline consume the identical `TurnProgram`; the corridor
relaxes each step, the baseline folds it numerically. The honest question
is whether field relaxation matches arithmetic on the same problem.
"""
answer = float(seed)
for step in steps:
answer = step.scale * answer + step.offset
return answer
def _symbolic_multi_register(program) -> float:
"""Baseline for Tier-2: solve the SAME multi-register program arithmetically.
Mirror of the corridor executor with Python arithmetic instead of
relaxation — per-register affine folds, transfers move `operand` between
registers, a `None` answer entity sums all registers.
"""
registers = {entity: seed for entity, seed in program.seeds}
for turn in program.turns:
if turn.kind == "transfer":
registers[turn.actor] -= turn.operand
registers[turn.target] += turn.operand
else:
registers[turn.actor] = turn.scale * registers[turn.actor] + turn.offset
if program.answer_entity is None:
return float(sum(registers.values()))
return registers[program.answer_entity]
def _corridor_and_baseline(graph):
"""Route a graph to Tier-1 (single-accumulator) or Tier-2 (multi-register),
returning ``(corridor_answer, baseline_answer)``. Both paths consume the
identical compiled program; the corridor relaxes, the baseline folds
arithmetically. Raises ``MultiRegisterError`` when neither tier ingests.
Only the Tier-1 COMPILE is wrapped for the fallthrough: a future typed raise
from ``execute_turn_program`` must surface as a Tier-1 execution failure, not
be silently rerouted into Tier-2 (which would mask a regression). The tiers
differ on non-convergence — Tier-2 fails closed (raises), Tier-1 records the
iterate — but the gold comparison binds wrong=0 on both regardless."""
try:
program = compile_turn_program(graph)
except TurnProgramError:
program = None
if program is not None:
return execute_turn_program(program).answer, _symbolic_fold(program.seed, program.steps)
mr_program = compile_multi_register_program(graph) # raises MultiRegisterError if un-ingestible
return execute_multi_register_program(mr_program).answer, _symbolic_multi_register(mr_program)
def run_arithmetic_chain_domain() -> DomainOutcome:
"""Corridor executor vs symbolic fold on real GSM8K problems (Tier-1 + Tier-2).
Each `ground_truth_graph` is routed to the single-accumulator turn program or
the multi-register program; both corridor and baseline consume that same
compiled program. Anything neither tier ingests (derived-operand transfers,
>5-atom shapes) is refused by BOTH and recorded, never silently dropped. The
expected honest verdict is PARITY with wrong=0: the field matches arithmetic,
it does not beat it; the result is real-holdout coverage.
"""
corridor_correct = corridor_wrong = corridor_refused = 0
baseline_correct = baseline_wrong = baseline_refused = 0
rows: list[dict[str, Any]] = []
cases = _load_dev_cases()
for case in cases:
case_id = case.get("id", "")
gold_raw = case.get("expected_answer")
graph_dict = case.get("ground_truth_graph")
try:
graph = graph_from_dict(graph_dict) if graph_dict else None
except (MathGraphError, KeyError, TypeError, ValueError):
graph = None
if graph is None: # reader/graph-parse failure — distinct from a compiler refusal
corridor_refused += 1
baseline_refused += 1
rows.append({"case_id": case_id, "ingested": False, "refusal": "graph_parse_failed"})
continue
try:
corridor_answer, baseline_answer = _corridor_and_baseline(graph)
except MultiRegisterError as refusal: # neither tier ingests — record the reason
corridor_refused += 1
baseline_refused += 1
rows.append({"case_id": case_id, "ingested": False, "refusal": refusal.reason})
continue
gold = None if gold_raw is None else float(gold_raw)
corridor_ok = gold is not None and abs(corridor_answer - gold) < 1e-4
baseline_ok = gold is not None and abs(baseline_answer - gold) < 1e-4
corridor_correct += int(corridor_ok)
corridor_wrong += int(not corridor_ok)
baseline_correct += int(baseline_ok)
baseline_wrong += int(not baseline_ok)
rows.append(
{
"case_id": case_id,
"ingested": True,
"gold": gold,
"corridor_answer": corridor_answer,
"baseline_answer": baseline_answer,
"corridor_ok": corridor_ok,
}
)
ingested = corridor_correct + corridor_wrong
return DomainOutcome(
domain_id="arithmetic-chain",
n_cases=len(cases),
corridor_correct=corridor_correct,
corridor_wrong=corridor_wrong,
corridor_refused=corridor_refused,
baseline_correct=baseline_correct,
baseline_wrong=baseline_wrong,
baseline_refused=baseline_refused,
notes=(
f"Real GSM8K dev holdout: {ingested}/{len(cases)} problems ingestible "
f"(Tier-1 single-accumulator + Tier-2 multi-entity/transfer/summation) and "
f"solved wrong=0. Derived-operand transfers / >5-atom shapes remain the frontier.",
"Baseline is a symbolic fold of the SAME compiled program; PARITY "
"with wrong=0 is the honest outcome — the field matches arithmetic, it "
"does not beat it. The result is real-holdout coverage, not a lift.",
),
cases=tuple(rows),
)
def run_generalized_lift_instrument() -> LiftInstrumentReport:
outcomes = (
run_propositional_domain(),
run_recognition_domain(),
run_multimodal_domain(),
run_arithmetic_chain_domain(),
)
propositional = outcomes[0]
arithmetic = outcomes[3]
return LiftInstrumentReport(
outcomes=outcomes,
# Both exact-regime domains (deductive flagship + arithmetic) preserve wrong=0.
wrong_zero_guard_held=(propositional.corridor_wrong == 0 and arithmetic.corridor_wrong == 0),
honest_null=all(o.delta_correct <= 0 for o in outcomes),
scope_limitations=(
"The reader→Hamiltonian compiler now solves the FULL real GSM8K dev "
"holdout wrong=0 (Tier-1 single-accumulator, ADR-0249; Tier-2 "
"multi-entity + constant-operand transfers + certified summation, "
"ADR-0250) — see the arithmetic-chain domain. Remaining general "
"frontier: derived-operand transfers ('half of X') and non-affine "
"kinds (rate/comparison/fraction/partition), both 0 on this holdout; "
"and >5-atom deduction. Refused, never silently dropped.",
),
)