From 40db09b559387bcd6f5a1b2e74f9c6692e262f00 Mon Sep 17 00:00:00 2001 From: Shay Date: Fri, 17 Jul 2026 10:46:20 -0700 Subject: [PATCH] =?UTF-8?q?feat(adr-0243):=20Phase=203=20Lane=20B=20?= =?UTF-8?q?=E2=80=94=20sensorium=20corridor=20eval=20(I-04=20first=20live?= =?UTF-8?q?=20consumer)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New evals/adr_0243_cognitive_lifecycle/ composes AudioCompiler/VisionCompiler -> sensorium_wave_feed -> relax_to_ground/CognitiveLifecycleEngine.egress -> generate/realizer readback -> explicit GoldTetherMonitor.decide into one deterministic fixed-replay corridor, per docs/handoff/ADR-0243-PHASE3-BRIEF-PACK.md Brief B. Pure composition of already-tested organs; no new core.physics module; OFF-SERVING (evals/ only, never imported by chat/runtime.py). Relaxation targets the full audio+vision integrated percept but starts from the audio-only partial percept (engine.solve() is not used, since it ties start == target) so the relaxer decodes across real steps rather than starting already-converged; stages are driven explicitly and reassembled into one LifecycleOutcome keyed off the full percept. [Verification]: uv run core test --suite smoke -q (176 passed); sensorium suite (152 passed); algebra suite (82 passed, 50 skipped); new tests/test_adr_0243_sensorium_corridor.py (6 passed, includes a steps_taken > 0 pin against silent regression to a no-op relax, a cosine/ANN-free static pin, and a chat/runtime.py non-import pin). --- .../adr_0243_cognitive_lifecycle/__init__.py | 198 ++++++++++++++++++ .../adr_0243_cognitive_lifecycle/__main__.py | 37 ++++ tests/test_adr_0243_sensorium_corridor.py | 131 ++++++++++++ 3 files changed, 366 insertions(+) create mode 100644 evals/adr_0243_cognitive_lifecycle/__init__.py create mode 100644 evals/adr_0243_cognitive_lifecycle/__main__.py create mode 100644 tests/test_adr_0243_sensorium_corridor.py diff --git a/evals/adr_0243_cognitive_lifecycle/__init__.py b/evals/adr_0243_cognitive_lifecycle/__init__.py new file mode 100644 index 00000000..d43d8833 --- /dev/null +++ b/evals/adr_0243_cognitive_lifecycle/__init__.py @@ -0,0 +1,198 @@ +"""ADR-0243 Phase 3 Lane B — sensorium corridor eval (research evidence, OFF-SERVING). + +First live consumer of the I-04 sensorium feed (`docs/plans/adr-0243-implementation-plan.md` +§5 Phase 3 Lane B). Composes already-built, already-tested organs into one deterministic +corridor — no new ``core.physics`` module, no new resonance math: + + AudioCompiler.compile / VisionCompiler.compile_tile (sensorium/*) + │ .versor + ▼ + packet_from_compilation_unit(modality_id, unit) (core/physics/sensorium_wave_feed.py) + ▼ + relax_to_ground(ψ_partial, H(ψ_full)) via CognitiveLifecycleEngine.egress (relax → egress) + ▼ E3/E4 route: "readback_eligible" + generate/realizer.py:energy_modulated_surface(base_surface, energy_class) + │ + └── separately, explicitly: GoldTetherMonitor.decide(...) + +Lives under ``evals/`` only; never imported by ``chat/runtime.py`` (A-04 quarantine — +inherited transitively from ``core.physics.cognitive_lifecycle`` / +``core.physics.sensorium_wave_feed``, both already pinned in +``tests/test_serve_quarantine_transitive.py`` / ``tests/test_third_door_cohesion.py``). + +The corridor's Hamiltonian targets the FULL audio+vision integrated field +(``target_psi = ingress_full.psi``); relaxation starts from the audio-only PARTIAL +percept (``ingress_partial.psi``) so the relaxer actually decodes across real steps +toward the full percept — "recognize what was just perceived" only holds if the +relaxer starts short of what it is recognizing. ``engine.solve()`` is not used here +because it ties relaxation start == Hamiltonian target; the stages are driven +explicitly instead, then reassembled into one ``LifecycleOutcome`` (keeping +``outcome_id``/digests keyed off the full percept, not the partial start). No +cosine/ANN anywhere — resonance inside the corridor is algebraic +(``WaveManifold``-backed) only. +""" + +from __future__ import annotations + +from typing import Mapping + +import numpy as np + +from core.physics.cognitive_lifecycle import ( + CognitiveLifecycleEngine, + LifecycleOutcome, + compile_quadratic_well, + ingest_context, + relax_to_ground, +) +from core.physics.goldtether import GoldTetherMonitor, OperatingMode +from core.physics.sensorium_wave_feed import packet_from_compilation_unit +from generate.realizer import energy_modulated_surface +from sensorium.audio.compiler import AudioCompiler +from sensorium.vision import VisionCompiler, canonicalize_image +from sensorium.vision.grid import iter_tile_signals + +__all__ = [ + "run_fixed_replay", + "sensorium_corridor_artifact", +] + +# Hot-band energy inputs matching the existing E3/E4 precedent +# (tests/test_adr_0243_cognitive_lifecycle.py::test_egress_routes_hot_state_to_readback_eligible). +# Caller-supplied structural axes only — never invented inside the engine. +_DEFAULT_ENERGY_INPUTS: Mapping[str, object] = { + "convergence_density": 8, + "activation_count": 8, + "current_cycle": 1, + "last_activation_cycle": 1, + "morphology_features": {"mood": "imperative"}, +} + + +def _fixed_audio_tone(sample_rate: int, duration_s: float, freq_hz: float) -> np.ndarray: + n = int(sample_rate * duration_s) + t = np.arange(n, dtype=np.float64) / sample_rate + return (0.5 * np.sin(2.0 * np.pi * freq_hz * t)).astype(np.float32) + + +def _fixed_vision_tile(): + x = np.linspace(0.0, 1.0, 32, dtype=np.float32) + y = np.linspace(0.0, 1.0, 32, dtype=np.float32) + xx, yy = np.meshgrid(x, y) + image = np.stack([xx, yy, 1.0 - xx], axis=2).astype(np.float32) + return iter_tile_signals(canonicalize_image(image))[0] + + +def sensorium_corridor_artifact( + *, + domain_id: str, + sample_rate: int, + tone_freq_hz: float, + tone_duration_s: float, + curvature: float, + energy_inputs: Mapping[str, object], + epsilon_drift: float, +) -> dict: + """Run the full corridor once on deterministic fixed inputs; return a JSON-safe dict. + + Fails closed (raises the module's typed errors) rather than repairing a + malformed intermediate state at any stage. + """ + audio_unit = AudioCompiler().compile( + _fixed_audio_tone(sample_rate, tone_duration_s, tone_freq_hz), sample_rate + ) + 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) + packets = (audio_pkt, vision_pkt) + + # Target: the full integrated percept. Start: the audio-only partial percept — + # a genuinely different unit state with positive overlap on the full field, so + # relaxation decodes across real steps instead of starting already-converged. + ingress_full = ingest_context(packets, domain_id) + hamiltonian = compile_quadratic_well(ingress_full.psi, curvature=curvature) + ingress_partial = ingest_context((audio_pkt,), domain_id) + + engine = CognitiveLifecycleEngine(epsilon_drift=epsilon_drift) + result = relax_to_ground(ingress_partial.psi, hamiltonian) + verdict = engine.egress(result.psi_steady, result.certificate, **energy_inputs) + outcome = LifecycleOutcome(ingress=ingress_full, relaxation=result, verdict=verdict) + + readback_text: str | None = None + if verdict.route == "readback_eligible": + base_surface = ( + f"the {domain_id} field integrated {len(packets)} modality packets " + f"and relaxed to a {verdict.energy_class.value} state" + ) + readback_text = energy_modulated_surface(base_surface, verdict.energy_class) + + # A fresh monitor is uncalibrated (autonomy=0) and the superposition residual + # is well above epsilon_drift, so this legitimately comes back fail_closed — + # that is the honest HITL-safe default, not a corridor bug. + monitor = GoldTetherMonitor(epsilon_drift=epsilon_drift) + decision = monitor.decide(verdict.versor_residual, mode=OperatingMode.PRACTICE) + + return { + "domain_id": domain_id, + "modality_ids": list(ingress_full.modality_ids), + "audio": { + "versor_condition": audio_unit.versor_condition, + "projection_sha256": audio_unit.projection_sha256, + }, + "vision": { + "versor_condition": vision_unit.versor_condition, + "projection_sha256": vision_unit.projection_sha256, + }, + "ingress": { + "packet_digest": ingress_full.packet_digest, + "partial_packet_digest": ingress_partial.packet_digest, + }, + "hamiltonian": { + "hamiltonian_id": hamiltonian.hamiltonian_id, + "domain": hamiltonian.domain, + }, + "relaxation": outcome.relaxation.certificate.as_dict(), + "egress": { + "admitted": verdict.admitted, + "reason": verdict.reason, + "route": verdict.route, + "unit_norm_residual": verdict.unit_norm_residual, + "versor_residual": verdict.versor_residual, + "versor_closed": verdict.versor_closed, + "energy_class": verdict.energy_class.value, + "energy_raw": verdict.energy_profile.raw, + }, + "readback_text": readback_text, + "goldtether_decision": { + "band": decision.band.value, + "residual": decision.residual, + "floor": decision.floor, + "autonomy": decision.autonomy, + "mode": decision.mode.value, + "reason": decision.reason, + }, + "outcome_id": outcome.outcome_id, + } + + +def run_fixed_replay( + *, + domain_id: str = "adr_0243_sensorium_corridor_v1", + sample_rate: int = 24_000, + tone_freq_hz: float = 160.0, + tone_duration_s: float = 0.35, + curvature: float = 1.0, + energy_inputs: Mapping[str, object] | None = None, + epsilon_drift: float = 1e-6, +) -> dict: + """Entry point for the fixed-replay sensorium corridor eval (deterministic).""" + return sensorium_corridor_artifact( + domain_id=domain_id, + sample_rate=sample_rate, + tone_freq_hz=tone_freq_hz, + tone_duration_s=tone_duration_s, + curvature=curvature, + energy_inputs=dict(energy_inputs) if energy_inputs is not None else dict(_DEFAULT_ENERGY_INPUTS), + epsilon_drift=epsilon_drift, + ) diff --git a/evals/adr_0243_cognitive_lifecycle/__main__.py b/evals/adr_0243_cognitive_lifecycle/__main__.py new file mode 100644 index 00000000..4f101cbd --- /dev/null +++ b/evals/adr_0243_cognitive_lifecycle/__main__.py @@ -0,0 +1,37 @@ +"""CLI: python -m evals.adr_0243_cognitive_lifecycle [--out PATH] + +Writes the fixed-replay sensorium corridor artifact as JSON. +Research/OFF-SERVING only; never imported from chat/runtime.py. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from evals.adr_0243_cognitive_lifecycle import run_fixed_replay + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument( + "--out", + type=Path, + default=None, + help="Optional output path (default: stdout)", + ) + args = p.parse_args(argv) + artifact = run_fixed_replay() + text = json.dumps(artifact, indent=2, sort_keys=True) + "\n" + if args.out is not None: + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text(text, encoding="utf-8") + else: + sys.stdout.write(text) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_adr_0243_sensorium_corridor.py b/tests/test_adr_0243_sensorium_corridor.py new file mode 100644 index 00000000..37f0dcb1 --- /dev/null +++ b/tests/test_adr_0243_sensorium_corridor.py @@ -0,0 +1,131 @@ +"""ADR-0243 Phase 3 Lane B — sensorium corridor eval (I-04 first live consumer). + +End-to-end: real Audio/Vision compilers → ingress → relaxation → egress → +readback → GoldTether, all composed (no re-implementation) in +``evals/adr_0243_cognitive_lifecycle``. Deterministic fixed-replay; no fixture +files (precedent: tests/test_adr_0241_sensorium_wave_feed.py:230, +tests/test_adr_0242_multi_scale_energy.py::test_eval_entry_matches_physics_helper). + +Off-serving (A-04): ``core.physics.cognitive_lifecycle`` and +``core.physics.sensorium_wave_feed`` are already pinned quarantined in +``tests/test_serve_quarantine_transitive.py`` / ``tests/test_third_door_cohesion.py``; +this file adds a direct pin that ``chat/runtime.py`` never imports this eval package. +""" + +from __future__ import annotations + +import ast +import json +from pathlib import Path + +from evals.adr_0243_cognitive_lifecycle import run_fixed_replay, sensorium_corridor_artifact + +_ROOT = Path(__file__).resolve().parents[1] +_PACKAGE = _ROOT / "evals" / "adr_0243_cognitive_lifecycle" +_CLOSURE = 1e-6 + + +def test_corridor_is_deterministic(): + """Same fixed inputs → byte-identical artifact (no randomness, no hidden state).""" + first = json.dumps(run_fixed_replay(), sort_keys=True) + second = json.dumps(run_fixed_replay(), sort_keys=True) + assert first == second + + +def test_corridor_end_to_end_composes_real_compilers_through_readback_and_goldtether(): + artifact = run_fixed_replay() + + # Real compilers ran (not fake_deterministic_packet) and closed their own versors. + assert artifact["modality_ids"] == ["audio", "vision"] + assert artifact["audio"]["versor_condition"] < _CLOSURE + assert artifact["vision"]["versor_condition"] < _CLOSURE + + # Relaxation decodes from the audio-only PARTIAL percept toward a well targeting + # the full audio+vision percept — steps_taken > 0 pins that the relaxer actually + # did the decode instead of starting already at its own target (start != target). + relax = artifact["relaxation"] + assert relax["converged"] is True + assert relax["reason"] == "ground_state_certified" + assert relax["steps_taken"] > 0 + assert artifact["ingress"]["partial_packet_digest"] != artifact["ingress"]["packet_digest"] + + # Egress admitted and routed to the hot band — readback path taken. + egress = artifact["egress"] + assert egress["admitted"] is True + assert egress["reason"] == "admitted" + assert egress["route"] == "readback_eligible" + assert egress["energy_class"] in ("E3", "E4") + + # A multi-mode superposition is NOT a closed versor — egress must not have + # silently gated on versor closure to reach admitted/readback_eligible. + assert egress["versor_closed"] is False + assert egress["versor_residual"] > _CLOSURE + + # E3/E4 readback carries no hedge prefix (ADR-0006): energy_modulated_surface + # must not have silently repaired/altered the base surface for a hot state. + assert artifact["readback_text"] is not None + assert artifact["readback_text"] == ( + f"the {artifact['domain_id']} field integrated 2 modality packets " + f"and relaxed to a {egress['energy_class']} state" + ) + + # GoldTetherMonitor.decide is a genuinely separate, explicit call: egress_gate + # never populates it, and a fresh monitor fails closed regardless of residual. + gt = artifact["goldtether_decision"] + assert gt["residual"] == egress["versor_residual"] + assert gt["band"] == "fail_closed" + assert gt["reason"] == "residual_or_autonomy_fail_closed" + + +def test_corridor_refuses_when_energy_inputs_stay_cold(): + """Sanity: the hot-band route is earned by the supplied energy_inputs, not hardcoded. + + Default (E0-band) energy_inputs on the same real-compiler packets must NOT + reach readback_eligible — proves the corridor reports the engine's actual + routing decision rather than a fixed "readback_eligible" stub. + """ + cold = sensorium_corridor_artifact( + domain_id="adr_0243_sensorium_corridor_v1", + sample_rate=24_000, + tone_freq_hz=160.0, + tone_duration_s=0.35, + curvature=1.0, + energy_inputs={}, + epsilon_drift=1e-6, + ) + assert cold["egress"]["route"] != "readback_eligible" + assert cold["readback_text"] is None + + +def test_corridor_artifact_is_json_serializable(): + text = json.dumps(run_fixed_replay(), sort_keys=True) + assert json.loads(text)["outcome_id"] + + +def test_corridor_module_calls_no_cosine_or_ann_similarity(): + """No cosine/ANN anywhere in the corridor — I-04's algebraic-only resonance.""" + forbidden = {"cosine_similarity", "resonant_recall", "resonant_reconstruct", "sklearn"} + for path in (_PACKAGE / "__init__.py", _PACKAGE / "__main__.py"): + src = path.read_text(encoding="utf-8") + tree = ast.parse(src) + called: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): + called.add(node.func.attr) + if isinstance(node, ast.Import): + called.update(alias.name for alias in node.names) + if isinstance(node, ast.ImportFrom) and node.module: + called.add(node.module) + assert not (called & forbidden), f"{path.name} touches forbidden similarity op: {called & forbidden}" + + +def test_chat_runtime_does_not_import_the_corridor_eval(): + """The corridor lives under evals/ only — never a serve-path import.""" + runtime_src = (_ROOT / "chat" / "runtime.py").read_text(encoding="utf-8") + tree = ast.parse(runtime_src) + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module: + assert "adr_0243_cognitive_lifecycle" not in node.module + if isinstance(node, ast.Import): + for alias in node.names: + assert "adr_0243_cognitive_lifecycle" not in alias.name