feat(adr-0181-p3): audio_core_v1 pack + AudioProjectionHead (gate-closed) (#467)
PR-3 of ADR-0181. Externalises the PR-2 in-code operator registry to a
versioned, checksum-locked pack and wires the sensorium adapter. Mounts
gate-closed until the PR-4 eval gates pass. No core mutation (ADR-0013).
Pack (packs/audio/audio_core_v1/):
- manifest.json canonical params, resampling config, gating defaults, ordering
- operators.jsonl the 8 elliptic operators (byte-equivalent to DEFAULT_OPERATOR_REGISTRY)
- basis_map.json alias→index over the six elliptic planes {6,7,8,10,11,13}
- resample_fir_v1.npy pinned odd-length symmetric windowed-sinc low-pass (unit DC)
- checksums.json sha256 of every artifact
Loader (packs/audio/loader.py) — trust boundary (ADR-0051 / CLAUDE.md §Security):
- _validate_pack_id rejects traversal/separators/dotfiles before any path join
- fail-closed checksum verification: every file re-hashed vs checksums.json
- resolved path must stay under the packs root (defense in depth)
- JSON format (matches every other CORE pack loader; spec §6.1 TOML was
illustrative; no tomllib dependency added)
Adapter (sensorium/adapters/audio.py):
- AudioProjectionHead wraps AudioCompiler; project(AudioSignal)->(32,) f32
- make_audio_pack loads+verifies the pack, builds a gate-closed ModalityPack
Tests (19): PR-2↔PR-3 registry byte-parity via manifest_sha256, elliptic-only
operators, FIR integrity, fail-closed checksum mismatch, path-traversal guard,
gate-closed refusal, mount-time unitarity (+ bad-unitarity blocks mount),
gate-engage requires checksum_verified, engaged pack projects (32,) f32.
All audio 32 + arch-invariants 40 + smoke 67 green.
This commit is contained in:
parent
4d59e66043
commit
7f619d44e5
10 changed files with 485 additions and 0 deletions
5
packs/audio/__init__.py
Normal file
5
packs/audio/__init__.py
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
"""packs.audio — audio modality packs (ADR-0181). See loader.load_audio_pack."""
|
||||
|
||||
from packs.audio.loader import AudioPackError, LoadedAudioPack, load_audio_pack
|
||||
|
||||
__all__ = ["load_audio_pack", "LoadedAudioPack", "AudioPackError"]
|
||||
15
packs/audio/audio_core_v1/basis_map.json
Normal file
15
packs/audio/audio_core_v1/basis_map.json
Normal file
|
|
@ -0,0 +1,15 @@
|
|||
{
|
||||
"basis_version": "audio-basis-v1",
|
||||
"comment": "Elliptic grade-2 planes only (e_a e_b, a,b in {e1..e4}) — square to -1. Indices 9,12,14,15 touch e5 and are hyperbolic; excluded from v1. See ADR-0181 / sensorium/audio/operators.py.",
|
||||
"elliptic_planes": [6, 7, 8, 10, 11, 13],
|
||||
"aliases": {
|
||||
"B_PAUSE_SHORT": 6,
|
||||
"B_PAUSE_LONG": 7,
|
||||
"B_SPEECH": 8,
|
||||
"B_PITCH_RISE": 10,
|
||||
"B_PITCH_FALL": 11,
|
||||
"B_EMPHASIS": 13,
|
||||
"B_TURN": 6,
|
||||
"B_NOISE": 7
|
||||
}
|
||||
}
|
||||
9
packs/audio/audio_core_v1/checksums.json
Normal file
9
packs/audio/audio_core_v1/checksums.json
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
{
|
||||
"files": {
|
||||
"basis_map.json": "sha256:423fa846961bd66f8ed67da9ebf92d3f2a26ce71e671415cb9a13ba8ff4d2c71",
|
||||
"manifest.json": "sha256:e1a2f13e84cb6b51f07adae490004ac3c2bde6db4b9115dcf5b60cf459919fc4",
|
||||
"operators.jsonl": "sha256:a240e5e9963654e3500cf3aa9f8560273db4c12c1e9280f960dfac35f71dd594",
|
||||
"resample_fir_v1.npy": "sha256:bcbea0eb13e8217cb001a962863cc430b89d8912091e0cfe635dd1a3d4e9a06b"
|
||||
},
|
||||
"pack_id": "audio_core_v1"
|
||||
}
|
||||
38
packs/audio/audio_core_v1/manifest.json
Normal file
38
packs/audio/audio_core_v1/manifest.json
Normal file
|
|
@ -0,0 +1,38 @@
|
|||
{
|
||||
"pack_id": "audio_core_v1",
|
||||
"modality": "audio",
|
||||
"cl41_dim": 32,
|
||||
"compiler_version": "0.1.0",
|
||||
"basis_version": "audio-basis-v1",
|
||||
"canonical": {
|
||||
"sample_rate": 24000,
|
||||
"channels": 1,
|
||||
"frame_ms": 20,
|
||||
"hop_ms": 10,
|
||||
"output_dtype": "float32",
|
||||
"internal_dtype": "float64"
|
||||
},
|
||||
"resampling": {
|
||||
"algorithm": "polyphase_fir",
|
||||
"fir_path": "resample_fir_v1.npy",
|
||||
"fir_sha256": "bcbea0eb13e8217cb001a962863cc430b89d8912091e0cfe635dd1a3d4e9a06b",
|
||||
"padtype": "constant",
|
||||
"cval": 0.0
|
||||
},
|
||||
"gating": {
|
||||
"gate_engaged": false,
|
||||
"checksum_verified": false,
|
||||
"versor_condition_max": 1e-06
|
||||
},
|
||||
"ordering": {
|
||||
"event_precedence": [
|
||||
"channel",
|
||||
"pause",
|
||||
"speech",
|
||||
"prosody",
|
||||
"turn",
|
||||
"non_speech",
|
||||
"content_anchor"
|
||||
]
|
||||
}
|
||||
}
|
||||
8
packs/audio/audio_core_v1/operators.jsonl
Normal file
8
packs/audio/audio_core_v1/operators.jsonl
Normal file
|
|
@ -0,0 +1,8 @@
|
|||
{"operator_id":"audio.pause.short.v1","event_type":"pause.short","blade_alias":"B_PAUSE_SHORT","blade_index":6,"rotor_kind":"elliptic","base_theta_q":48,"gain_rules":{"dur_hops":2},"theta_clip_q":768,"version":"1"}
|
||||
{"operator_id":"audio.pause.long.v1","event_type":"pause.long","blade_alias":"B_PAUSE_LONG","blade_index":7,"rotor_kind":"elliptic","base_theta_q":96,"gain_rules":{"dur_hops":2},"theta_clip_q":768,"version":"1"}
|
||||
{"operator_id":"audio.speech.voiced.v1","event_type":"speech.voiced","blade_alias":"B_SPEECH","blade_index":8,"rotor_kind":"elliptic","base_theta_q":64,"gain_rules":{"dur_hops":1},"theta_clip_q":768,"version":"1"}
|
||||
{"operator_id":"audio.prosody.rise.v1","event_type":"prosody.rise","blade_alias":"B_PITCH_RISE","blade_index":10,"rotor_kind":"elliptic","base_theta_q":64,"gain_rules":{"slope_q":3},"theta_clip_q":768,"version":"1"}
|
||||
{"operator_id":"audio.prosody.fall.v1","event_type":"prosody.fall","blade_alias":"B_PITCH_FALL","blade_index":11,"rotor_kind":"elliptic","base_theta_q":64,"gain_rules":{"slope_q":3},"theta_clip_q":768,"version":"1"}
|
||||
{"operator_id":"audio.prosody.emph.v1","event_type":"prosody.emphasis","blade_alias":"B_EMPHASIS","blade_index":13,"rotor_kind":"elliptic","base_theta_q":32,"gain_rules":{"delta_db_q":4},"theta_clip_q":768,"version":"1"}
|
||||
{"operator_id":"audio.turn.boundary.v1","event_type":"turn.boundary","blade_alias":"B_TURN","blade_index":6,"rotor_kind":"elliptic","base_theta_q":160,"gain_rules":{"boundary_q":2},"theta_clip_q":768,"version":"1"}
|
||||
{"operator_id":"audio.nonspeech.noise.v1","event_type":"nonspeech.noise","blade_alias":"B_NOISE","blade_index":7,"rotor_kind":"elliptic","base_theta_q":200,"gain_rules":{"noise_q":2},"theta_clip_q":768,"version":"1"}
|
||||
BIN
packs/audio/audio_core_v1/resample_fir_v1.npy
Normal file
BIN
packs/audio/audio_core_v1/resample_fir_v1.npy
Normal file
Binary file not shown.
135
packs/audio/loader.py
Normal file
135
packs/audio/loader.py
Normal file
|
|
@ -0,0 +1,135 @@
|
|||
"""
|
||||
packs/audio/loader.py — audio pack loader (ADR-0181 PR-3).
|
||||
|
||||
Trust boundary (CLAUDE.md §Security / ADR-0051):
|
||||
|
||||
* The pack id is validated for filesystem safety *before* any path join —
|
||||
traversal tokens, separators, and dotfiles fail closed.
|
||||
* The loader is fail-closed on checksum mismatch: every file named in
|
||||
``checksums.json`` is re-hashed and compared before the pack is observable
|
||||
to runtime code. A tampered or stale artifact blocks the load (and therefore
|
||||
the mount — bad checksum blocks pack mount, eval-plan §2).
|
||||
* The loader never mutates a pack on disk.
|
||||
|
||||
The pack format is JSON (matching every other CORE pack loader and avoiding a
|
||||
TOML parser dependency); ADR-0181 spec §6.1 showed TOML illustratively only.
|
||||
The loaded operator registry is byte-equivalent to the in-code
|
||||
``DEFAULT_OPERATOR_REGISTRY`` (asserted by tests via ``manifest_sha256``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
from sensorium.audio.operators import AudioOperatorRegistry, OperatorSpec
|
||||
|
||||
_PACKS_ROOT = Path(__file__).resolve().parent
|
||||
|
||||
|
||||
class AudioPackError(ValueError):
|
||||
"""Raised when an audio pack is missing, malformed, or fails verification."""
|
||||
|
||||
|
||||
def _validate_pack_id(pack_id: object) -> str:
|
||||
"""Reject unsafe pack ids before any filesystem access (ADR-0051)."""
|
||||
from core._safe_display import safe_pack_id as _disp
|
||||
|
||||
if not isinstance(pack_id, str):
|
||||
raise AudioPackError(f"pack_id must be a string, got {_disp(pack_id)!r}")
|
||||
if pack_id == "":
|
||||
raise AudioPackError("pack_id must not be empty")
|
||||
if ".." in pack_id:
|
||||
raise AudioPackError(f"pack_id must not contain '..': {_disp(pack_id)!r}")
|
||||
if "/" in pack_id or "\\" in pack_id:
|
||||
raise AudioPackError(f"pack_id must be a simple pack id, not a path: {_disp(pack_id)!r}")
|
||||
if pack_id.startswith("."):
|
||||
raise AudioPackError(f"pack_id must not start with '.': {_disp(pack_id)!r}")
|
||||
for ch in pack_id:
|
||||
if not (ch.isascii() and (ch.isalnum() or ch in {"_", "-"})):
|
||||
raise AudioPackError(f"pack_id must be alphanumeric/_/-, got {_disp(pack_id)!r}")
|
||||
return pack_id
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class LoadedAudioPack:
|
||||
pack_id: str
|
||||
manifest: dict
|
||||
registry: AudioOperatorRegistry
|
||||
fir: np.ndarray | None
|
||||
fir_sha256: str
|
||||
|
||||
|
||||
def _verify_checksums(pack_dir: Path) -> None:
|
||||
"""Re-hash every file in checksums.json; fail closed on any mismatch."""
|
||||
checks_path = pack_dir / "checksums.json"
|
||||
if not checks_path.is_file():
|
||||
raise AudioPackError(f"checksums.json missing for pack at {pack_dir.name}")
|
||||
checks = json.loads(checks_path.read_text())
|
||||
for fname, expected in checks.get("files", {}).items():
|
||||
fpath = pack_dir / fname
|
||||
if not fpath.is_file():
|
||||
raise AudioPackError(f"pack file '{fname}' named in checksums.json is missing")
|
||||
actual = "sha256:" + hashlib.sha256(fpath.read_bytes()).hexdigest()
|
||||
if actual != expected:
|
||||
raise AudioPackError(
|
||||
f"checksum mismatch for '{fname}': expected {expected}, got {actual}"
|
||||
)
|
||||
|
||||
|
||||
def _load_registry(pack_dir: Path) -> AudioOperatorRegistry:
|
||||
specs: dict[str, OperatorSpec] = {}
|
||||
for line in (pack_dir / "operators.jsonl").read_text().splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
row = json.loads(line)
|
||||
# gain_rules dict -> sorted (name, gain) pairs for stable serialization.
|
||||
gains = tuple(sorted((k, int(v)) for k, v in row["gain_rules"].items()))
|
||||
spec = OperatorSpec(
|
||||
operator_id=row["operator_id"],
|
||||
event_type=row["event_type"],
|
||||
blade_alias=row["blade_alias"],
|
||||
blade_index=int(row["blade_index"]),
|
||||
base_theta_q=int(row["base_theta_q"]),
|
||||
gain_rules=gains,
|
||||
theta_clip_q=int(row["theta_clip_q"]),
|
||||
version=str(row.get("version", "1")),
|
||||
)
|
||||
specs[spec.event_type] = spec
|
||||
return AudioOperatorRegistry(specs)
|
||||
|
||||
|
||||
def load_audio_pack(
|
||||
pack_id: str = "audio_core_v1",
|
||||
*,
|
||||
packs_root: Path | None = None,
|
||||
verify: bool = True,
|
||||
) -> LoadedAudioPack:
|
||||
"""Load and verify an audio pack from ``packs/audio/<pack_id>/``."""
|
||||
safe_id = _validate_pack_id(pack_id)
|
||||
root = packs_root if packs_root is not None else _PACKS_ROOT
|
||||
pack_dir = (root / safe_id).resolve()
|
||||
# Defense in depth: the resolved path must stay under the packs root.
|
||||
if not str(pack_dir).startswith(str(root.resolve())):
|
||||
raise AudioPackError(f"resolved pack path escapes packs root: {safe_id!r}")
|
||||
if not pack_dir.is_dir():
|
||||
raise AudioPackError(f"no audio pack mounted at {safe_id!r}")
|
||||
|
||||
if verify:
|
||||
_verify_checksums(pack_dir)
|
||||
|
||||
manifest = json.loads((pack_dir / "manifest.json").read_text())
|
||||
registry = _load_registry(pack_dir)
|
||||
|
||||
fir_path = pack_dir / manifest.get("resampling", {}).get("fir_path", "resample_fir_v1.npy")
|
||||
fir: np.ndarray | None = np.load(fir_path) if fir_path.is_file() else None
|
||||
fir_sha256 = (
|
||||
"sha256:" + hashlib.sha256(fir_path.read_bytes()).hexdigest()
|
||||
if fir_path.is_file() else ""
|
||||
)
|
||||
return LoadedAudioPack(safe_id, manifest, registry, fir, fir_sha256)
|
||||
101
sensorium/adapters/audio.py
Normal file
101
sensorium/adapters/audio.py
Normal file
|
|
@ -0,0 +1,101 @@
|
|||
"""
|
||||
sensorium/adapters/audio.py — Audio modality adapter (ADR-0181 PR-3).
|
||||
|
||||
The thin, pack-governed ProjectionHead for audio. It wraps the deterministic
|
||||
AudioCompiler (PR-2): a canonical AudioSignal is compiled to one
|
||||
AudioCompilationUnit and the unit's (32,) float32 versor crosses the
|
||||
Logos-recovery boundary.
|
||||
|
||||
The pack mounts **gate-closed** (``gate_engaged=False``) until the eval gates
|
||||
(PR-4) pass — `ModalityRegistry.project` refuses a closed gate, so audio
|
||||
contributes no field state until determinism/checksum/versor gates are green
|
||||
(ADR-0181 §2.5). Adding audio touches no existing layer (ADR-0013).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
import numpy as np
|
||||
|
||||
from sensorium.audio.canonical import CANONICAL_SAMPLE_RATE, canonicalize
|
||||
from sensorium.audio.compiler import AudioCompiler
|
||||
from sensorium.audio.types import AudioSignal
|
||||
from sensorium.protocol import (
|
||||
CL41_DIM,
|
||||
Modality,
|
||||
ModalityPack,
|
||||
ModalityVocabulary,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class AudioProjectionHead:
|
||||
"""D1 projection head for audio.
|
||||
|
||||
Determinism class: D1 — deterministic given a pinned pack (operator
|
||||
registry + basis). ``project`` accepts an already-canonical AudioSignal;
|
||||
raw waveform is canonicalised upstream via ``canonicalize`` (a helper is
|
||||
exposed as ``project_raw`` for convenience).
|
||||
"""
|
||||
|
||||
compiler: AudioCompiler
|
||||
modality: Modality = Modality.AUDIO
|
||||
|
||||
@property
|
||||
def embedding_dim(self) -> int:
|
||||
return CL41_DIM
|
||||
|
||||
def project(self, signal: AudioSignal) -> np.ndarray:
|
||||
out = self.compiler.compile_signal(signal).versor
|
||||
if out.shape != (CL41_DIM,):
|
||||
raise ValueError(f"expected ({CL41_DIM},), got {out.shape}")
|
||||
if out.dtype != np.float32:
|
||||
raise TypeError(f"expected float32, got {out.dtype}")
|
||||
return out
|
||||
|
||||
def project_batch(self, signals: list[AudioSignal]) -> np.ndarray:
|
||||
return np.stack([self.project(s) for s in signals], axis=0)
|
||||
|
||||
def project_raw(
|
||||
self, samples: np.ndarray, sample_rate: int, *, fir: np.ndarray | None = None
|
||||
) -> np.ndarray:
|
||||
"""Convenience: canonicalise raw samples then project."""
|
||||
signal = canonicalize(samples, sample_rate, target_sr=CANONICAL_SAMPLE_RATE, fir=fir)
|
||||
return self.project(signal)
|
||||
|
||||
def verify_unitarity(self, sample: AudioSignal) -> bool:
|
||||
try:
|
||||
return self.compiler.compile_signal(sample).versor_condition < 1e-6
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def make_audio_pack(
|
||||
pack_id: str = "audio_core_v1",
|
||||
*,
|
||||
gate_engaged: bool = False,
|
||||
checksum_verified: bool = False,
|
||||
packs_root=None,
|
||||
) -> ModalityPack:
|
||||
"""Construct a gate-closed audio ModalityPack from a verified pack.
|
||||
|
||||
Loads + checksum-verifies the pack (fail-closed), builds the compiler over
|
||||
the pack's operator registry, and wraps it in an AudioProjectionHead. The
|
||||
pack ships gate-closed; engaging the gate is the PR-4 eval-gate decision.
|
||||
"""
|
||||
from packs.audio.loader import load_audio_pack
|
||||
|
||||
loaded = load_audio_pack(pack_id, packs_root=packs_root)
|
||||
compiler = AudioCompiler(loaded.registry, pack_id=loaded.pack_id)
|
||||
head = AudioProjectionHead(compiler)
|
||||
return ModalityPack(
|
||||
pack_id=loaded.pack_id,
|
||||
modality_type=Modality.AUDIO,
|
||||
projection=head,
|
||||
decoder=None,
|
||||
vocabulary=ModalityVocabulary(),
|
||||
grammar_scaffold=None,
|
||||
checksum_verified=checksum_verified,
|
||||
gate_engaged=gate_engaged,
|
||||
)
|
||||
71
tests/test_audio_pack_manifest.py
Normal file
71
tests/test_audio_pack_manifest.py
Normal file
|
|
@ -0,0 +1,71 @@
|
|||
"""
|
||||
ADR-0181 PR-3 — audio pack manifest + loader trust boundary.
|
||||
|
||||
Covers:
|
||||
- the loaded operator registry is byte-equivalent to the in-code
|
||||
DEFAULT_OPERATOR_REGISTRY (PR-2 ↔ PR-3 parity, via manifest_sha256);
|
||||
- fail-closed checksum verification (tampered artifact blocks load);
|
||||
- the ADR-0051 path-traversal guard;
|
||||
- the pinned FIR artifact integrity.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
|
||||
import pytest
|
||||
|
||||
from packs.audio.loader import AudioPackError, load_audio_pack
|
||||
from sensorium.audio.operators import DEFAULT_OPERATOR_REGISTRY, ELLIPTIC_PLANES
|
||||
|
||||
|
||||
def test_pack_loads_and_verifies():
|
||||
pack = load_audio_pack("audio_core_v1")
|
||||
assert pack.pack_id == "audio_core_v1"
|
||||
assert pack.manifest["cl41_dim"] == 32
|
||||
assert pack.manifest["gating"]["gate_engaged"] is False # ships gate-closed
|
||||
|
||||
|
||||
def test_loaded_registry_is_byte_equivalent_to_in_code_default():
|
||||
"""The pack artifact and the in-code DEFAULT_OPERATOR_REGISTRY must produce
|
||||
the identical content hash — PR-3 externalises PR-2 without drift."""
|
||||
pack = load_audio_pack("audio_core_v1")
|
||||
assert pack.registry.manifest_sha256() == DEFAULT_OPERATOR_REGISTRY.manifest_sha256()
|
||||
assert set(pack.registry.specs) == set(DEFAULT_OPERATOR_REGISTRY.specs)
|
||||
|
||||
|
||||
def test_all_loaded_operators_are_elliptic():
|
||||
pack = load_audio_pack("audio_core_v1")
|
||||
for spec in pack.registry.specs.values():
|
||||
assert spec.blade_index in ELLIPTIC_PLANES
|
||||
|
||||
|
||||
def test_fir_artifact_is_odd_length_and_matches_manifest():
|
||||
pack = load_audio_pack("audio_core_v1")
|
||||
assert pack.fir is not None
|
||||
assert pack.fir.ndim == 1 and pack.fir.size % 2 == 1 # zero-phase symmetric
|
||||
assert pack.fir_sha256.endswith(pack.manifest["resampling"]["fir_sha256"])
|
||||
|
||||
|
||||
# --- trust boundary -------------------------------------------------------
|
||||
|
||||
def test_checksum_mismatch_fails_closed(tmp_path):
|
||||
"""A tampered artifact must block the load (eval-plan §2 mount validation)."""
|
||||
from packs.audio.loader import _PACKS_ROOT
|
||||
dst = tmp_path / "audio_core_v1"
|
||||
shutil.copytree(_PACKS_ROOT / "audio_core_v1", dst)
|
||||
# Corrupt operators.jsonl without updating checksums.json.
|
||||
(dst / "operators.jsonl").write_text((dst / "operators.jsonl").read_text() + "\n")
|
||||
with pytest.raises(AudioPackError, match="checksum mismatch"):
|
||||
load_audio_pack("audio_core_v1", packs_root=tmp_path)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad_id", ["..", "../secrets", "a/b", "a\\b", ".hidden", ""])
|
||||
def test_path_traversal_rejected(bad_id):
|
||||
with pytest.raises(AudioPackError):
|
||||
load_audio_pack(bad_id)
|
||||
|
||||
|
||||
def test_missing_pack_fails_closed(tmp_path):
|
||||
with pytest.raises(AudioPackError, match="no audio pack"):
|
||||
load_audio_pack("audio_core_v1", packs_root=tmp_path)
|
||||
103
tests/test_audio_sensorium_mount.py
Normal file
103
tests/test_audio_sensorium_mount.py
Normal file
|
|
@ -0,0 +1,103 @@
|
|||
"""
|
||||
ADR-0181 PR-3 — audio ProjectionHead mount + gate governance.
|
||||
|
||||
Covers the registry contract for audio:
|
||||
- the pack mounts gate-closed; projection is refused while the gate is closed;
|
||||
- mount-time unitarity runs on a probe AudioSignal and passes;
|
||||
- bad unitarity blocks the mount;
|
||||
- engaging the gate requires checksum_verified (ModalityPack invariant);
|
||||
- an engaged pack projects to (32,) float32.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from sensorium.adapters.audio import AudioProjectionHead, make_audio_pack
|
||||
from sensorium.audio.canonical import canonicalize
|
||||
from sensorium.protocol import CL41_DIM, Modality, ModalityPack, ModalityVocabulary
|
||||
from sensorium.registry import ModalityRegistry
|
||||
|
||||
SR = 24_000
|
||||
|
||||
|
||||
def _signal():
|
||||
n = int(SR * 0.3)
|
||||
t = np.arange(n, dtype=np.float64) / SR
|
||||
samples = (0.5 * np.sin(2 * np.pi * 160.0 * t)).astype(np.float32)
|
||||
return canonicalize(samples, SR)
|
||||
|
||||
|
||||
def test_pack_ships_gate_closed():
|
||||
pack = make_audio_pack("audio_core_v1")
|
||||
assert pack.modality_type is Modality.AUDIO
|
||||
assert pack.gate_engaged is False
|
||||
assert pack.projection is not None
|
||||
|
||||
|
||||
def test_mount_runs_unitarity_and_succeeds():
|
||||
reg = ModalityRegistry()
|
||||
pack = make_audio_pack("audio_core_v1")
|
||||
reg.mount(pack, sample=_signal()) # runs verify_unitarity on the probe
|
||||
assert "audio_core_v1" in reg
|
||||
|
||||
|
||||
def test_closed_gate_refuses_projection():
|
||||
reg = ModalityRegistry()
|
||||
reg.mount(make_audio_pack("audio_core_v1"), sample=_signal())
|
||||
with pytest.raises(RuntimeError, match="gate is not engaged"):
|
||||
reg.project("audio_core_v1", _signal())
|
||||
|
||||
|
||||
def test_bad_unitarity_blocks_mount():
|
||||
"""A projection whose verify_unitarity fails must not mount (eval-plan §2)."""
|
||||
class _BrokenHead:
|
||||
modality = Modality.AUDIO
|
||||
embedding_dim = CL41_DIM
|
||||
|
||||
def project(self, signal): # pragma: no cover - not reached
|
||||
return np.zeros(CL41_DIM, dtype=np.float32)
|
||||
|
||||
def project_batch(self, signals): # pragma: no cover
|
||||
return np.zeros((len(signals), CL41_DIM), dtype=np.float32)
|
||||
|
||||
def verify_unitarity(self, sample) -> bool:
|
||||
return False
|
||||
|
||||
pack = ModalityPack(
|
||||
pack_id="audio_broken",
|
||||
modality_type=Modality.AUDIO,
|
||||
projection=_BrokenHead(),
|
||||
vocabulary=ModalityVocabulary(),
|
||||
grammar_scaffold=None,
|
||||
checksum_verified=False,
|
||||
gate_engaged=False,
|
||||
)
|
||||
with pytest.raises(ValueError, match="Unitarity check failed"):
|
||||
ModalityRegistry().mount(pack, sample=_signal())
|
||||
|
||||
|
||||
def test_engaging_gate_requires_checksum_verified():
|
||||
with pytest.raises(ValueError, match="checksum_verified"):
|
||||
make_audio_pack("audio_core_v1", gate_engaged=True, checksum_verified=False)
|
||||
|
||||
|
||||
def test_engaged_pack_projects_32_float32():
|
||||
reg = ModalityRegistry()
|
||||
reg.mount(
|
||||
make_audio_pack("audio_core_v1", gate_engaged=True, checksum_verified=True),
|
||||
sample=_signal(),
|
||||
)
|
||||
mv = reg.project("audio_core_v1", _signal())
|
||||
assert mv.shape == (CL41_DIM,)
|
||||
assert mv.dtype == np.float32
|
||||
|
||||
|
||||
def test_project_raw_canonicalises_then_projects():
|
||||
head: AudioProjectionHead = make_audio_pack("audio_core_v1").projection
|
||||
n = int(SR * 0.25)
|
||||
t = np.arange(n, dtype=np.float64) / SR
|
||||
raw = (0.4 * np.sin(2 * np.pi * 200.0 * t)).astype(np.float32)
|
||||
mv = head.project_raw(raw, SR)
|
||||
assert mv.shape == (CL41_DIM,) and mv.dtype == np.float32
|
||||
Loading…
Reference in a new issue