From dd529ad408db3d9800cfd4417aa897f4de20796e Mon Sep 17 00:00:00 2001 From: Shay Date: Fri, 29 May 2026 13:22:44 -0700 Subject: [PATCH] feat(adr-0181-p5): audio Delta-CRDT arena + merge kernel (sequential==concurrent proof) (#477) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wires AudioCompilationUnits into the Delta-CRDT substrate (ADR-0180 §2.1/§2.2) at the Python layer — audio is the first concrete exerciser of ADR-0180. - sensorium/audio/arena.py: - AudioArena (§2.1): thread-local, share-nothing, lock-free accumulation; non-destructive snapshot(); thread_local_audio_arena() accessor. - AudioDelta (§2.2): canonical, content-addressed merge_key order + dedup; join is commutative/associative/idempotent. - merge_audio_deltas (§2.2): the Merge Kernel — folds arena deltas into one content-addressed, deduped, totally ordered set. - audio_merge_trace_hash: PCM-free, order-invariant trace hash — the sequential==concurrent proof anchor (ADR-0181 §4.2 A-3/A-6). Mirrors the Rust LocalArena/SemilatticeDelta/merge_kernel (core-rs/src/vault.rs, ADR-0180 §4.1) so the layers stay in parity when the Rust↔Python binding lands (§1.5.5, deferred — pure-CPU/Python path first). 10 failable proof tests in tests/test_audio_crdt_merge.py covering the load-bearing hash(Sequential)==hash(Concurrent) gate (real threads), the three semilattice legs, content-addressed ordering, dedup idempotence, and trace hygiene. Mutation-verified: disabling the content sort fails 6 of 10 loudly. Additive only: no core layer touched (ADR-0013). Audio tests 35/35; architectural invariants 40/40. Zero impact on existing eval lanes (no serving path changed). --- sensorium/audio/__init__.py | 19 +++- sensorium/audio/arena.py | 169 ++++++++++++++++++++++++++++++++ tests/test_audio_crdt_merge.py | 171 +++++++++++++++++++++++++++++++++ 3 files changed, 357 insertions(+), 2 deletions(-) create mode 100644 sensorium/audio/arena.py create mode 100644 tests/test_audio_crdt_merge.py diff --git a/sensorium/audio/__init__.py b/sensorium/audio/__init__.py index 9a360fb2..5b6f8348 100644 --- a/sensorium/audio/__init__.py +++ b/sensorium/audio/__init__.py @@ -6,10 +6,19 @@ Audio enters CORE as a compiler, not an embedding bridge: canonical waveform 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 only. Pack artifacts + the -AudioProjectionHead adapter land in PR-3; evals in PR-4; CRDT wiring in PR-5. +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``. """ +from sensorium.audio.arena import ( + AudioArena, + AudioDelta, + audio_merge_trace_hash, + merge_audio_deltas, + reset_thread_local_audio_arena, + thread_local_audio_arena, +) from sensorium.audio.compiler import AudioCompiler, compile_events from sensorium.audio.operators import ( DEFAULT_OPERATOR_REGISTRY, @@ -28,6 +37,12 @@ from sensorium.audio.types import ( ) __all__ = [ + "AudioArena", + "AudioDelta", + "audio_merge_trace_hash", + "merge_audio_deltas", + "reset_thread_local_audio_arena", + "thread_local_audio_arena", "AudioCompiler", "compile_events", "AudioOperatorRegistry", diff --git a/sensorium/audio/arena.py b/sensorium/audio/arena.py new file mode 100644 index 00000000..3e4d3c3d --- /dev/null +++ b/sensorium/audio/arena.py @@ -0,0 +1,169 @@ +""" +sensorium/audio/arena.py — audio Delta-CRDT arena + merge kernel (ADR-0181 PR-5). + +Wires `AudioCompilationUnit`s into the Delta-CRDT substrate (ADR-0180 §2.1/§2.2): +each compiled chunk is one delta; the audio adapter accumulates units in a +thread-local `AudioArena` (share-nothing, never writes global state); the merge +kernel folds arena snapshots into one content-addressed, deduplicated, totally +ordered set keyed by the unit's `merge_key` +(`canonical_sha256, ir_sha256, projection_sha256`). + +This is the **Python-layer mirror** of the Rust substrate +(`core-rs/src/vault.rs` — `LocalArena` / `SemilatticeDelta` / `merge_kernel`, +ADR-0180 §4.1): same content-addressed sort + byte-identical dedup, so the two +stay in parity when the Rust↔Python binding lands (ADR-0180 §1.5.5 — deferred to +a follow-up; the substrate works on the pure-CPU path first). Audio is the first +concrete exerciser of ADR-0180 (ADR-0181 §3.1). + +The merge result is **permutation- and duplicate-invariant** — the property +ADR-0180 §4.3 / ADR-0181 §4.2 (A-2, A-3) require: + + hash(Sequential_Ingest) == hash(Concurrent_CRDT_Ingest) + +Idempotence is structural, not asserted (ADR-0181 §2.2): identical canonical +bytes under an identical pack produce an identical `merge_key`, so the join +deduplicates them by construction. +""" + +from __future__ import annotations + +import hashlib +import json +import threading +from dataclasses import dataclass + +from sensorium.audio.trace import audio_evidence_trace +from sensorium.audio.types import AudioCompilationUnit + +MergeKey = tuple[str, str, str] + + +def _canonicalize( + units: tuple[AudioCompilationUnit, ...] | list[AudioCompilationUnit], +) -> tuple[AudioCompilationUnit, ...]: + """Order by content-addressed `merge_key`, drop byte-identical duplicates. + + `merge_key` encodes the canonical bytes, the IR, and the projection (the + versor) — so equal keys mean content-identical units; the dedup is exact and + arrival-independent, mirroring `core-rs` `Delta::from_entries`. + """ + ordered = sorted(units, key=lambda u: u.merge_key) + deduped: list[AudioCompilationUnit] = [] + last_key: MergeKey | None = None + for unit in ordered: + if unit.merge_key != last_key: + deduped.append(unit) + last_key = unit.merge_key + return tuple(deduped) + + +@dataclass(frozen=True, slots=True) +class AudioDelta: + """A canonical, order-invariant set of compiled audio chunks (ADR-0181 §2.2). + + Always held in content-addressed `merge_key` order with byte-identical + duplicates removed, so it is a canonical join-semilattice element regardless + of insertion order. Mirrors `core-rs` `Delta`. + """ + + units: tuple[AudioCompilationUnit, ...] + + @classmethod + def from_units( + cls, + units: tuple[AudioCompilationUnit, ...] | list[AudioCompilationUnit], + ) -> AudioDelta: + return cls(_canonicalize(units)) + + def join(self, other: AudioDelta) -> AudioDelta: + """Join semilattice op: commutative, associative, idempotent under + content-addressed equality (ADR-0180 §2.2).""" + return AudioDelta.from_units((*self.units, *other.units)) + + @property + def merge_keys(self) -> tuple[MergeKey, ...]: + return tuple(u.merge_key for u in self.units) + + def __len__(self) -> int: + return len(self.units) + + +class AudioArena: + """Thread-local, share-nothing accumulation arena for the audio adapter + (ADR-0180 §2.1). + + Push compiled units lock-free; nothing is ever written to global state from + an arena. `snapshot` emits a canonical, order-invariant `AudioDelta` and is + **non-destructive** — flush/GC is the Merge Kernel's concern, so a delayed + merge (the §3.2 eventual-consistency window) cannot lose a unit. Mirrors + `core-rs` `LocalArena`. + """ + + __slots__ = ("_units",) + + def __init__(self) -> None: + self._units: list[AudioCompilationUnit] = [] + + def push(self, unit: AudioCompilationUnit) -> None: + self._units.append(unit) + + def is_empty(self) -> bool: + return not self._units + + def snapshot(self) -> AudioDelta: + return AudioDelta.from_units(self._units) + + def __len__(self) -> int: + return len(self._units) + + +_THREAD_LOCAL = threading.local() + + +def thread_local_audio_arena() -> AudioArena: + """The calling thread's audio arena (ADR-0180 §2.1: each active adapter gets + a thread-local arena). Created on first access per thread; never shared + across threads, so it needs no lock.""" + arena = getattr(_THREAD_LOCAL, "audio_arena", None) + if arena is None: + arena = AudioArena() + _THREAD_LOCAL.audio_arena = arena + return arena + + +def reset_thread_local_audio_arena() -> None: + """Drop the calling thread's arena (flush/test helper). Pools reuse threads, + so callers that want a fresh arena per task must reset first.""" + if hasattr(_THREAD_LOCAL, "audio_arena"): + del _THREAD_LOCAL.audio_arena + + +def merge_audio_deltas(deltas: list[AudioDelta] | tuple[AudioDelta, ...]) -> AudioDelta: + """The Merge Kernel (ADR-0180 §2.2): fold arena deltas into one content- + addressed, deduplicated, totally ordered `AudioDelta`. + + Implemented as a single canonicalisation of the union rather than a + `fold(join)` chain; `tests/test_audio_crdt_merge.py` pins that the two are + equal, so the cheap path can never silently diverge from the semilattice + fold. Permutation- and duplicate-invariant — the property §4.3's + `hash(Sequential) == hash(Concurrent)` rides on. + """ + units: list[AudioCompilationUnit] = [] + for delta in deltas: + units.extend(delta.units) + return AudioDelta.from_units(units) + + +def audio_merge_trace_hash(merged: AudioDelta) -> str: + """Deterministic, PCM-free trace hash over a merged `AudioDelta` — the + `sequential == concurrent` anchor (ADR-0181 §3.1, §4.2 A-3/A-6). + + The payload is the per-unit evidence trace (no waveform — ADR-0180 §1.5.5) + in canonical `merge_key` order. Identical content in any arrival order yields + the same hash; distinct content yields a different hash. Determinism here is + order-invariance on a fixed platform; the compiler's A-1 gate owns + cross-platform float stability of the underlying versors. + """ + payload = [audio_evidence_trace(unit) for unit in merged.units] + blob = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(blob).hexdigest() diff --git a/tests/test_audio_crdt_merge.py b/tests/test_audio_crdt_merge.py new file mode 100644 index 00000000..788fd79d --- /dev/null +++ b/tests/test_audio_crdt_merge.py @@ -0,0 +1,171 @@ +""" +ADR-0181 PR-5 — audio Delta-CRDT merge proof obligations. + +The load-bearing gate is `hash(Sequential_Ingest) == hash(Concurrent_CRDT_Ingest)` +(ADR-0180 §4.3 / ADR-0181 §4.2 A-2, A-3): a set of `AudioCompilationUnit`s folds +to the same merged Vault contribution — and the same trace hash — regardless of +the order arenas were filled or flushed in. + +Per CLAUDE.md §Schema-Defined Proof Obligations, each test FAILS LOUDLY under the +violation it names: if `_canonicalize` stopped sorting by `merge_key` (ordered by +arrival instead), the permutation / sequential==concurrent / content-order tests +break; if it stopped deduplicating, the idempotence test breaks. +""" + +from __future__ import annotations + +import json +import random +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import numpy as np +import pytest + +from evals.audio_sensorium.synth import synthesize +from sensorium.audio.arena import ( + AudioArena, + AudioDelta, + audio_merge_trace_hash, + merge_audio_deltas, + reset_thread_local_audio_arena, + thread_local_audio_arena, +) +from sensorium.audio.canonical import canonicalize +from sensorium.audio.compiler import AudioCompiler +from sensorium.audio.trace import audio_evidence_trace + +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 units() -> list: + """Real AudioCompilationUnits — one per eval fixture, distinct merge_keys.""" + compiler = AudioCompiler() + out = [] + for fx in _fixtures(): + signal = canonicalize(synthesize(fx), SR) + out.append(compiler.compile_signal(signal)) + return out + + +# --- the load-bearing property (ADR-0181 §4.2 A-2 / A-3) ------------------- + + +def test_sequential_equals_concurrent(units): + """hash(Sequential) == hash(Concurrent). Single-arena fixture-order ingest + vs one unit per thread-local arena flushed under real thread scheduling.""" + seq_arena = AudioArena() + for unit in units: + seq_arena.push(unit) + seq_hash = audio_merge_trace_hash(merge_audio_deltas([seq_arena.snapshot()])) + + def worker(unit) -> AudioDelta: + # Pools reuse threads; reset so each task gets a fresh share-nothing arena. + reset_thread_local_audio_arena() + arena = thread_local_audio_arena() + arena.push(unit) + return arena.snapshot() + + for _ in range(5): # exercise scheduling non-determinism + order = units[:] + random.shuffle(order) + with ThreadPoolExecutor(max_workers=len(order)) as pool: + deltas = list(pool.map(worker, order)) + conc_hash = audio_merge_trace_hash(merge_audio_deltas(deltas)) + assert conc_hash == seq_hash + + +def test_merge_is_permutation_invariant(units): + base = merge_audio_deltas([AudioDelta.from_units(units)]) + for _ in range(8): + shuffled = units[:] + random.shuffle(shuffled) + merged = merge_audio_deltas([AudioDelta.from_units([u]) for u in shuffled]) + assert merged.merge_keys == base.merge_keys + assert audio_merge_trace_hash(merged) == audio_merge_trace_hash(base) + + +# --- semilattice legs (ADR-0180 §2.2) ------------------------------------- + + +def test_merge_is_idempotent(units): + delta = AudioDelta.from_units(units) + once = merge_audio_deltas([delta]) + twice = merge_audio_deltas([delta, delta]) + assert once.merge_keys == twice.merge_keys + assert audio_merge_trace_hash(once) == audio_merge_trace_hash(twice) + # distinct fixtures => no spurious duplicates introduced or dropped + assert len(twice) == len(units) + + +def test_join_is_commutative_and_associative(units): + a = AudioDelta.from_units(units[:2]) + b = AudioDelta.from_units(units[2:4]) + c = AudioDelta.from_units(units[4:]) + assert a.join(b).merge_keys == b.join(a).merge_keys + assert a.join(b).join(c).merge_keys == a.join(b.join(c)).merge_keys + + +def test_merge_kernel_equals_semilattice_fold(units): + deltas = [AudioDelta.from_units([u]) for u in units] + folded = deltas[0] + for d in deltas[1:]: + folded = folded.join(d) + assert merge_audio_deltas(deltas).merge_keys == folded.merge_keys + + +# --- content-addressing + dedup correctness -------------------------------- + + +def test_merge_order_is_content_addressed(units): + delta = AudioDelta.from_units(units) + assert list(delta.merge_keys) == sorted(delta.merge_keys) + # all distinct fixtures retained + assert len(delta) == len({u.merge_key for u in units}) + + +def test_arena_push_order_independent(units): + forward = AudioArena() + for u in units: + forward.push(u) + backward = AudioArena() + for u in reversed(units): + backward.push(u) + assert forward.snapshot().merge_keys == backward.snapshot().merge_keys + + +def test_arena_snapshot_non_draining(units): + arena = AudioArena() + arena.push(units[0]) + _ = arena.snapshot() + # flush/GC is the kernel's job; snapshot must not lose the unit across the + # eventual-consistency window (ADR-0180 §3.2). + assert len(arena) == 1 + assert len(arena.snapshot()) == 1 + + +# --- trace-hash sensitivity + hygiene (ADR-0181 §4.2 A-6) ------------------ + + +def test_trace_hash_changes_with_content(units): + """Guards against a vacuous (constant) hash: dropping a unit must change it.""" + full = AudioDelta.from_units(units) + subset = AudioDelta.from_units(units[:-1]) + assert audio_merge_trace_hash(full) != audio_merge_trace_hash(subset) + + +def test_merge_trace_hash_no_pcm(units): + delta = AudioDelta.from_units(units) + for unit in delta.units: + trace = audio_evidence_trace(unit) + for value in trace.values(): + assert not isinstance(value, (np.ndarray, bytes, bytearray)) + assert "samples" not in trace + # stable across repeated calls + assert audio_merge_trace_hash(delta) == audio_merge_trace_hash(delta)