"""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. Three 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 v1'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. Scope limitations are RECORDED, never silently dropped (no-silent-caps): GSM8K and other natural-language arithmetic are NOT ingestible by corridor v1 — there is no reader→Hamiltonian compiler beyond the ≤5-atom propositional and quadratic-well domains. That compiler is the real composition frontier, and this instrument is the harness waiting for it. """ from __future__ import annotations from dataclasses import dataclass from itertools import product 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 vocab.manifold import VocabManifold __all__ = [ "DomainOutcome", "LiftInstrumentReport", "run_generalized_lift_instrument", "run_propositional_domain", "run_recognition_domain", "run_multimodal_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 --------------------------------------------------------------- def run_generalized_lift_instrument() -> LiftInstrumentReport: outcomes = ( run_propositional_domain(), run_recognition_domain(), run_multimodal_domain(), ) propositional = outcomes[0] return LiftInstrumentReport( outcomes=outcomes, wrong_zero_guard_held=(propositional.corridor_wrong == 0), honest_null=all(o.delta_correct <= 0 for o in outcomes), scope_limitations=( "GSM8K / natural-language arithmetic is NOT ingestible by corridor " "v1: no reader-to-Hamiltonian compiler exists beyond the <=5-atom " "propositional and quadratic-well domains. Recorded, not silently " "dropped; that compiler is the composition frontier this " "instrument is waiting to measure.", ), )