feat(adr-0243): Phase 3 Lane B — sensorium corridor eval (I-04 first live consumer)
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).
This commit is contained in:
parent
d5db97f0e2
commit
40db09b559
3 changed files with 366 additions and 0 deletions
198
evals/adr_0243_cognitive_lifecycle/__init__.py
Normal file
198
evals/adr_0243_cognitive_lifecycle/__init__.py
Normal file
|
|
@ -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,
|
||||
)
|
||||
37
evals/adr_0243_cognitive_lifecycle/__main__.py
Normal file
37
evals/adr_0243_cognitive_lifecycle/__main__.py
Normal file
|
|
@ -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())
|
||||
131
tests/test_adr_0243_sensorium_corridor.py
Normal file
131
tests/test_adr_0243_sensorium_corridor.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue