feat(adr-0181-p5): audio Delta-CRDT arena + merge kernel (sequential==concurrent proof) (#477)

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

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

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

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)