diff --git a/pyproject.toml b/pyproject.toml index 6b4fe1d1..1ce0b7b5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,13 @@ dev = [ "pytest-asyncio>=0.23", "hypothesis>=6.100", ] +# ADR-0181 PR-6 — audio teacher / shadow lanes. Strictly optional: the +# deterministic compiler is the substrate; these models only label or align and +# are admitted as typed hints (sensorium.audio.teachers). Never runtime-required. +audio-whisper = ["openai-whisper>=20231117"] +audio-nemo = ["nemo_toolkit[asr]>=1.23"] +audio-clap = ["laion-clap>=1.1.6"] +audio-encodec = ["encodec>=0.1.1"] [project.scripts] core = "core.cli:main" diff --git a/sensorium/audio/__init__.py b/sensorium/audio/__init__.py index 5b6f8348..ca47ff8e 100644 --- a/sensorium/audio/__init__.py +++ b/sensorium/audio/__init__.py @@ -7,8 +7,9 @@ replayable. Each compiled chunk is one AudioCompilationUnit — the Delta-CRDT delta the audio adapter writes into its thread-local arena (ADR-0181 §2.1). PR-2 ships the deterministic substrate; pack artifacts + the AudioProjectionHead -adapter land in PR-3; evals in PR-4; the Delta-CRDT arena/merge wiring (PR-5) -lives in ``sensorium.audio.arena``. +adapter land in PR-3; evals in PR-4; CRDT arena/merge wiring in PR-5 +(`sensorium.audio.arena`); teacher/shadow lanes in PR-6 +(`sensorium.audio.teachers` — typed hints only, never substrate). """ from sensorium.audio.arena import ( @@ -26,6 +27,17 @@ from sensorium.audio.operators import ( OperatorSpec, build_elliptic_rotor, ) +from sensorium.audio.teachers import ( + AudioTeacher, + StubTranscriptTeacher, + TeacherHint, + TeacherLaneSpec, + TeacherUnavailable, + attach_teacher_hints, + is_lane_available, + KNOWN_TEACHER_LANES, + load_teacher, +) from sensorium.audio.trace import audio_evidence_trace from sensorium.audio.types import ( AudioCompilationUnit, @@ -50,6 +62,15 @@ __all__ = [ "DEFAULT_OPERATOR_REGISTRY", "build_elliptic_rotor", "audio_evidence_trace", + "AudioTeacher", + "StubTranscriptTeacher", + "TeacherHint", + "TeacherLaneSpec", + "TeacherUnavailable", + "attach_teacher_hints", + "is_lane_available", + "KNOWN_TEACHER_LANES", + "load_teacher", "AudioCompilationUnit", "AudioIR", "AudioSignal", diff --git a/sensorium/audio/parser.py b/sensorium/audio/parser.py index f2853921..aba1eca1 100644 --- a/sensorium/audio/parser.py +++ b/sensorium/audio/parser.py @@ -97,14 +97,9 @@ def parse(tokens: tuple[AudioToken, ...], n_hops: int) -> AudioIR: AuditoryEvent("nonspeech.noise", start, end, (("noise_q", dur),), ()) ) - ir_payload = { - "speech": [_ev(e) for e in speech_spans], - "pause": [_ev(e) for e in pause_spans], - "prosody": [_ev(e) for e in prosody_arcs], - "turn": [_ev(e) for e in turn_events], - "non_speech": [_ev(e) for e in non_speech_events], - "content_anchor": [], - } + ir_payload = _ir_payload( + speech_spans, pause_spans, prosody_arcs, turn_events, non_speech_events, () + ) return AudioIR( speech_spans=tuple(speech_spans), pause_spans=tuple(pause_spans), @@ -124,3 +119,29 @@ def _ev(e: AuditoryEvent) -> dict: "attrs": [list(a) for a in e.attrs], "evidence_ids": list(e.evidence_ids), } + + +def _ir_payload(speech, pause, prosody, turn, non_speech, content_anchor) -> dict: + """Canonical JSON-serialisable IR image — the single source of truth for + ``ir_sha256`` so a hint-augmented IR (PR-6) hashes by the same rule.""" + return { + "speech": [_ev(e) for e in speech], + "pause": [_ev(e) for e in pause], + "prosody": [_ev(e) for e in prosody], + "turn": [_ev(e) for e in turn], + "non_speech": [_ev(e) for e in non_speech], + "content_anchor": [_ev(e) for e in content_anchor], + } + + +def ir_sha256_of(ir: AudioIR) -> str: + """Recompute ``ir_sha256`` from an AudioIR's events. Byte-identical to what + ``parse`` stored for an un-augmented IR (regression-guarded in tests); the + teacher-hint admission path (`sensorium.audio.teachers`) uses it to re-hash + an IR after appending content anchors.""" + return sha256_json( + _ir_payload( + ir.speech_spans, ir.pause_spans, ir.prosody_arcs, + ir.turn_events, ir.non_speech_events, ir.content_anchors, + ) + ) diff --git a/sensorium/audio/teachers.py b/sensorium/audio/teachers.py new file mode 100644 index 00000000..53c29264 --- /dev/null +++ b/sensorium/audio/teachers.py @@ -0,0 +1,253 @@ +""" +sensorium/audio/teachers.py — teacher / shadow lanes (ADR-0181 PR-6). + +Teachers **label or align**; they never define the substrate and never fold +embeddings into the main versor path (eval-plan §4, verbatim): + + Use teachers to label or align. + Never let teachers define the substrate. + Never fold teacher embeddings directly into the main versor path. + Only admit teacher outputs through typed, versioned, checksumed hints. + +A teacher emits typed, versioned, checksummed `TeacherHint`s. The only admission +path is `attach_teacher_hints`, which appends them to the IR's `content_anchors` +as `content.*` events. `content.*` is **not** an operator-registry key, so +`compile_events` skips them (compiler.py: "Events whose type has no operator are +skipped"): a hint contributes IR evidence and a different `ir_sha256`, but the +**versor and `projection_sha256` are byte-identical** with or without it. That is +the structural guarantee that teachers are shadow, proven failably in +`tests/test_audio_teachers.py`. + +Real model lanes (Whisper / NeMo / CLAP / EnCodec) are declared and gated behind +optional extras; their adapters are import-guarded and degrade gracefully when +the extra is absent. The deterministic `StubTranscriptTeacher` is the reference +implementation of the contract and needs no model weights. +""" + +from __future__ import annotations + +import importlib.util +from dataclasses import dataclass, replace +from typing import Callable, Literal, Protocol, Sequence, runtime_checkable + +from sensorium.audio.checksum import sha256_json +from sensorium.audio.parser import ir_sha256_of +from sensorium.audio.types import AudioIR, AudioSignal, AuditoryEvent + +HintType = Literal["transcript", "alignment", "sound_event", "lang_id", "reconstruction"] + +# The IR namespace teacher hints live under. Must never collide with an operator +# key (speech.* / pause.* / prosody.* / turn.* / nonspeech.*), or a hint would +# lower to a rotor and become substrate. +CONTENT_ANCHOR_PREFIX = "content" + + +@dataclass(frozen=True, slots=True) +class TeacherHint: + """A typed, versioned, checksummed teacher annotation (eval-plan §4). + + `payload` is typed/quantized ints or short strings — never a raw embedding. + `hint_sha256` checksums the canonical payload so the hint is content-addressed + and tamper-evident. + """ + + lane_id: str + lane_version: str + hint_type: HintType + start_hop: int + end_hop: int + payload: tuple[tuple[str, int | str], ...] + confidence_q: int # 0..255 + hint_sha256: str + + @classmethod + def make( + cls, + *, + lane_id: str, + lane_version: str, + hint_type: HintType, + start_hop: int, + end_hop: int, + payload: tuple[tuple[str, int | str], ...] = (), + confidence_q: int = 0, + ) -> TeacherHint: + digest = sha256_json( + { + "lane_id": lane_id, + "lane_version": lane_version, + "hint_type": hint_type, + "start_hop": start_hop, + "end_hop": end_hop, + "payload": [list(p) for p in payload], + "confidence_q": confidence_q, + } + ) + return cls( + lane_id=lane_id, + lane_version=lane_version, + hint_type=hint_type, + start_hop=start_hop, + end_hop=end_hop, + payload=payload, + confidence_q=confidence_q, + hint_sha256=digest, + ) + + @property + def evidence_id(self) -> str: + return f"{self.lane_id}:{self.lane_version}:{self.hint_sha256}" + + def to_anchor(self) -> AuditoryEvent: + """Lower to a `content.*` IR anchor — evidence only, never a rotor.""" + attrs: tuple[tuple[str, int | str], ...] = ( + ("lane_id", self.lane_id), + ("lane_version", self.lane_version), + ("hint_type", self.hint_type), + ("confidence_q", self.confidence_q), + *self.payload, + ) + return AuditoryEvent( + event_type=f"{CONTENT_ANCHOR_PREFIX}.{self.hint_type}", + start_hop=self.start_hop, + end_hop=self.end_hop, + attrs=attrs, + evidence_ids=(self.evidence_id,), + ) + + +@runtime_checkable +class AudioTeacher(Protocol): + """A shadow-lane annotator. `annotate` must be pure on the signal — no + cross-modal or global state — mirroring the compiler's determinism so a hint + never depends on ingest order.""" + + lane_id: str + lane_version: str + + def annotate(self, signal: AudioSignal) -> tuple[TeacherHint, ...]: ... + + +def attach_teacher_hints(ir: AudioIR, hints: Sequence[TeacherHint]) -> AudioIR: + """Admit teacher hints into an IR as `content_anchors` (the ONLY admission + path). Returns a NEW AudioIR with `ir_sha256` recomputed; the input is + unchanged (immutable). The versor is unaffected — `content.*` anchors carry + no operator (proven in tests).""" + if not hints: + return ir + anchors = tuple(h.to_anchor() for h in hints) + merged = tuple( + sorted( + (*ir.content_anchors, *anchors), + key=lambda e: (e.start_hop, e.end_hop, e.event_type, e.evidence_ids), + ) + ) + augmented = replace(ir, content_anchors=merged) + return replace(augmented, ir_sha256=ir_sha256_of(augmented)) + + +# --------------------------------------------------------------------------- +# Reference teacher (no model weights) — the working contract instance. +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class StubTranscriptTeacher: + """Deterministic reference teacher. Emits one whole-signal transcript hint + whose token is derived from the canonical hash — pure on the signal, so the + same signal always yields the same hint. Exercises the contract end-to-end + without a real ASR dependency.""" + + lane_id: str = "stub_transcript" + lane_version: str = "stub-v1" + hop_samples: int = 240 # 10 ms @ 24 kHz + + def annotate(self, signal: AudioSignal) -> tuple[TeacherHint, ...]: + n_hops = max(1, int(len(signal.samples) // self.hop_samples)) + token = signal.canonical_sha256[:8] + return ( + TeacherHint.make( + lane_id=self.lane_id, + lane_version=self.lane_version, + hint_type="transcript", + start_hop=0, + end_hop=n_hops, + payload=(("token", token),), + confidence_q=128, + ), + ) + + +# --------------------------------------------------------------------------- +# Real model lanes — declared, gated, import-guarded (eval-plan §4 table). +# --------------------------------------------------------------------------- + + +class TeacherUnavailable(RuntimeError): + """Raised when a declared teacher lane cannot be constructed — the extra is + not installed, or its adapter has not been wired yet.""" + + +@dataclass(frozen=True, slots=True) +class TeacherLaneSpec: + lane_id: str + extra: str # pip extra that provides the model + probe_module: str # importable module used to detect availability + role: str # what the lane is good for (never substrate) + builder: Callable[[], AudioTeacher] | None = None # None until an adapter lands + + +# eval-plan §4: best role in CORE / why not the substrate. Adapters are deferred +# until the extra + weights are present; the lane is declared and gated now. +KNOWN_TEACHER_LANES: dict[str, TeacherLaneSpec] = { + "whisper": TeacherLaneSpec( + "whisper", "audio-whisper", "whisper", + "offline transcript evidence, weak lexical labels, language ID", + ), + "nemo": TeacherLaneSpec( + "nemo", "audio-nemo", "nemo", + "timestamp/alignment teacher + streaming transcript evidence", + ), + "clap": TeacherLaneSpec( + "clap", "audio-clap", "laion_clap", + "coarse sound-event labels, audio-text alignment", + ), + "encodec": TeacherLaneSpec( + "encodec", "audio-encodec", "encodec", + "reconstruction shadow lane, transport, future speech-to-speech", + ), +} + + +def is_lane_available(lane_id: str) -> bool: + """True iff the lane's optional dependency is importable. Does not import it.""" + spec = KNOWN_TEACHER_LANES.get(lane_id) + if spec is None: + return False + return importlib.util.find_spec(spec.probe_module) is not None + + +def load_teacher(lane_id: str) -> AudioTeacher: + """Construct a teacher for a declared lane, or fail loudly with guidance. + + Raises `ValueError` for an unknown lane, `TeacherUnavailable` when the extra + is missing or no adapter is wired yet. Never returns a partially-built or + silent-fallback teacher — a teacher must be a real, typed-hint producer or + nothing at all. + """ + spec = KNOWN_TEACHER_LANES.get(lane_id) + if spec is None: + known = ", ".join(sorted(KNOWN_TEACHER_LANES)) + raise ValueError(f"unknown teacher lane {lane_id!r}; known lanes: {known}") + if spec.builder is None: + raise TeacherUnavailable( + f"teacher lane {lane_id!r} is declared but its adapter is not wired " + f"yet (install `{spec.extra}` and register a builder). " + f"Role: {spec.role}." + ) + if not is_lane_available(lane_id): + raise TeacherUnavailable( + f"teacher lane {lane_id!r} requires the optional extra " + f"`{spec.extra}` (module {spec.probe_module!r} not importable)." + ) + return spec.builder() diff --git a/tests/test_audio_teachers.py b/tests/test_audio_teachers.py new file mode 100644 index 00000000..bbf88255 --- /dev/null +++ b/tests/test_audio_teachers.py @@ -0,0 +1,175 @@ +""" +ADR-0181 PR-6 — teacher / shadow lane proof obligations. + +The load-bearing rule (eval-plan §4): teachers label or align, they NEVER define +the substrate. The structural guarantee is that admitting any teacher hint leaves +the **versor and `projection_sha256` byte-identical** — a hint adds IR evidence +(`content_anchors`) and moves only `ir_sha256`, never the geometry that crosses +the ProjectionHead boundary. + +Per CLAUDE.md §Schema-Defined Proof Obligations, the tests FAIL LOUDLY under the +violation each names: if a teacher anchor ever lowered to a rotor (e.g. attached +under an operator category instead of `content.*`), the substrate-invariance test +breaks; if `attach_teacher_hints` silently dropped hints, the evidence-recorded +test breaks. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import numpy as np +import pytest + +from evals.audio_sensorium.synth import synthesize +from sensorium.audio.canonical import canonicalize +from sensorium.audio.compiler import AudioCompiler +from sensorium.audio.operators import DEFAULT_OPERATOR_REGISTRY +from sensorium.audio.parser import ir_sha256_of +from sensorium.audio.teachers import ( + KNOWN_TEACHER_LANES, + StubTranscriptTeacher, + TeacherHint, + TeacherUnavailable, + attach_teacher_hints, + is_lane_available, + load_teacher, +) + +SR = 24_000 +_EVAL_DIR = Path("evals/audio_sensorium") + + +def _fixtures() -> list[dict]: + return json.loads((_EVAL_DIR / "fixtures.json").read_text())["fixtures"] + + +@pytest.fixture(scope="module") +def compiler() -> AudioCompiler: + return AudioCompiler() + + +@pytest.fixture(scope="module") +def signal(): + # a voiced fixture so the base versor is non-trivial (real rotors fold in) + fx = next(f for f in _fixtures() if f["id"] == "rise_question") + return canonicalize(synthesize(fx), SR) + + +@pytest.fixture +def base_unit(compiler, signal): + return compiler.compile_signal(signal) + + +@pytest.fixture +def hints(signal): + return StubTranscriptTeacher().annotate(signal) + + +# --- the load-bearing rule: teachers never define the substrate ------------ + + +def test_teacher_hint_does_not_change_versor(compiler, base_unit, hints): + augmented_ir = attach_teacher_hints(base_unit.audio_ir, hints) + taught = compiler.compile_ir(augmented_ir) + + # substrate is invariant to the teacher + assert np.array_equal(base_unit.versor, taught.versor) + assert base_unit.projection_sha256 == taught.projection_sha256 + assert taught.versor_condition < 1e-6 + + +def test_teacher_hint_is_recorded_as_evidence(compiler, base_unit, hints): + augmented_ir = attach_teacher_hints(base_unit.audio_ir, hints) + taught = compiler.compile_ir(augmented_ir) + + # evidence is recorded => ir_sha256 (and only the ir leg of merge_key) moves + assert taught.ir_sha256 != base_unit.ir_sha256 + assert taught.merge_key != base_unit.merge_key + assert taught.merge_key[2] == base_unit.merge_key[2] # projection leg unchanged + assert len(augmented_ir.content_anchors) == len(hints) + + +def test_teacher_anchors_carry_no_operator(base_unit, hints): + """Structural guarantee: content.* anchors are not operator keys, so + compile_events skips them (no rotor).""" + augmented_ir = attach_teacher_hints(base_unit.audio_ir, hints) + for anchor in augmented_ir.content_anchors: + assert anchor.event_type.startswith("content.") + assert anchor.event_type not in DEFAULT_OPERATOR_REGISTRY + + +def test_attach_is_immutable(base_unit, hints): + original_anchors = base_unit.audio_ir.content_anchors + original_sha = base_unit.audio_ir.ir_sha256 + _ = attach_teacher_hints(base_unit.audio_ir, hints) + assert base_unit.audio_ir.content_anchors == original_anchors + assert base_unit.audio_ir.ir_sha256 == original_sha + + +def test_attach_empty_is_noop(base_unit): + assert attach_teacher_hints(base_unit.audio_ir, ()) is base_unit.audio_ir + + +# --- ir_sha256_of refactor regression guard -------------------------------- + + +def test_ir_sha256_of_matches_parse(base_unit): + # the extracted hash function must reproduce what parse() stored + assert ir_sha256_of(base_unit.audio_ir) == base_unit.ir_sha256 + + +def test_attached_ir_sha_is_recomputable(base_unit, hints): + augmented = attach_teacher_hints(base_unit.audio_ir, hints) + assert ir_sha256_of(augmented) == augmented.ir_sha256 + + +# --- hint typing / versioning / checksum ----------------------------------- + + +def test_hint_is_versioned_and_checksummed(): + a = TeacherHint.make( + lane_id="x", lane_version="v1", hint_type="transcript", + start_hop=0, end_hop=5, payload=(("token", "abcd"),), confidence_q=100, + ) + same = TeacherHint.make( + lane_id="x", lane_version="v1", hint_type="transcript", + start_hop=0, end_hop=5, payload=(("token", "abcd"),), confidence_q=100, + ) + diff = TeacherHint.make( + lane_id="x", lane_version="v1", hint_type="transcript", + start_hop=0, end_hop=5, payload=(("token", "WXYZ"),), confidence_q=100, + ) + assert a.hint_sha256 == same.hint_sha256 # content-addressed, stable + assert a.hint_sha256 != diff.hint_sha256 # changes with payload + assert a.evidence_id == f"x:v1:{a.hint_sha256}" + + +def test_stub_teacher_is_deterministic(signal): + t = StubTranscriptTeacher() + assert t.annotate(signal)[0].hint_sha256 == t.annotate(signal)[0].hint_sha256 + + +# --- lane registry: declared, gated, graceful ------------------------------ + + +def test_known_lanes_match_eval_plan(): + assert set(KNOWN_TEACHER_LANES) == {"whisper", "nemo", "clap", "encodec"} + + +def test_load_unknown_lane_raises_value_error(): + with pytest.raises(ValueError, match="unknown teacher lane"): + load_teacher("does_not_exist") + + +def test_load_declared_lane_without_adapter_is_unavailable(): + # lanes are declared + gated; no adapter is wired yet -> loud, never silent + with pytest.raises(TeacherUnavailable): + load_teacher("whisper") + + +def test_is_lane_available_does_not_raise(): + # pure probe; returns a bool for known and unknown lanes + assert isinstance(is_lane_available("whisper"), bool) + assert is_lane_available("does_not_exist") is False