core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(adr-0181-p6): audio teacher/shadow lanes — typed hints, never substrate (#479)

Browse source 
Implements ADR-0181 PR-6 (eval-plan §4): teachers label or align; they never
define the substrate and never fold embeddings into the versor path.

- sensorium/audio/teachers.py:
  - TeacherHint: typed, versioned, checksummed annotation (no raw embeddings).
  - AudioTeacher protocol (pure on the signal).
  - attach_teacher_hints: the ONLY admission path — appends content.* anchors to
    the IR's content_anchors (immutable, recomputes ir_sha256). content.* is not
    an operator key, so compile_events skips it: versor + projection_sha256 stay
    byte-identical; only the ir leg of the merge_key moves (evidence recorded).
  - KNOWN_TEACHER_LANES (whisper/nemo/clap/encodec): declared + gated behind
    optional extras; load_teacher import-guards and fails loudly (never a silent
    fallback). StubTranscriptTeacher is the deterministic reference instance.
- parser.py: extract _ir_payload + ir_sha256_of (DRY single source of truth for
  ir_sha256; byte-identical to parse() output — regression-guarded).
- pyproject.toml: audio-whisper/nemo/clap/encodec optional extras (never
  runtime-required).

16 failable proof tests in tests/test_audio_teachers.py. Load-bearing:
test_teacher_hint_does_not_change_versor. Mutation-verified — giving a teacher
anchor an operator event_type (folding it into the versor) fails the
versor-invariance proof; reverted, all pass.

Additive only (ADR-0013): no core layer touched. Audio suite 57/57; eval-gate
ir_sha256 pins unchanged by the parser refactor; architectural invariants 40/40.
Real model adapters are deferred until extras+weights are present; this PR ships
the policy, the typed-hint contract, and the shadow-only guarantee.
This commit is contained in:
Shay 2026-05-29 13:36:33 -07:00 committed by GitHub
parent 938d2bfeb3
commit f017785a6d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 487 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
pyproject.toml
View file

@ -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"

25
sensorium/audio/__init__.py
View file

@ -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",

37
sensorium/audio/parser.py
View file

@ -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,
)
)

253
sensorium/audio/teachers.py Normal file
View file

@ -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()

175
tests/test_audio_teachers.py Normal file
View file

@ -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