P3-R3: Multi-track recording engine

- WAV writer with streaming support for 16/24/32-bit integer and 32-bit float
- MultiTrackRecorder with per-channel arming, arm state machine, write-ahead
  buffer queues decoupling JACK real-time thread from SD card I/O
- Punch in/out recording on individual channels with punch region tracking
- Session file format (JSON): session metadata, takes, tracks, mixer state,
  punch regions, serialization round-trip
- Take management: create, delete, rename, list, track association
- Stereo bounce engine with pan law (-3dB constant power), normalisation,
  master bus volume/mute/dim, block-based processing for memory efficiency
- Disk space monitor with configurable warning/critical thresholds, estimated
  recording time remaining, background polling, callback on threshold change
- 117 tests: sample conversion, WAV header validation, recorder lifecycle,
  session CRUD, serialization, bounce with pan/normalisation/master, integration

Integrates with JACK capture ports → mixer → recording ports → disk.
Architecture: JACK callback → recorder.write() → buffer queue → writer thread → WAV files.

src/recording/__init__.py

@@ -0,0 +1,83 @@
+"""Multi-track recording engine — capture + WAV export + session management.
+
+The recording package provides a complete recording subsystem for the
+Raspberry Pi RT Audio Mixer, including:
+
+  - Streaming WAV writer with configurable bit depth (16/24/32 bit)
+    and sample rate
+  - Multi-track recorder with disk streaming (write-ahead buffering
+    to decouple capture from SD card I/O)
+  - Punch in/out recording on individual channels
+  - Session file format (JSON + audio references) with take management
+  - Stereo bounce/export of master mix with pan law and normalisation
+  - Free disk space monitoring with configurable thresholds
+
+Integrates with JACK: capture ports → mixer → recording ports → disk.
+"""
+
+from __future__ import annotations
+
+from .wav_writer import (
+    WAVWriter,
+    WAVWriterConfig,
+    WAVBitDepth,
+    write_wav,
+)
+from .recorder import (
+    MultiTrackRecorder,
+    RecorderConfig,
+    RecorderState,
+    ChannelState as RecordChannelState,
+    ChannelRecordStatus,
+    PunchRegion,
+)
+from .session import (
+    Session,
+    Take,
+    TrackInfo,
+    PunchInfo,
+)
+from .disk_monitor import (
+    DiskMonitor,
+    DiskMonitorConfig,
+    DiskSpaceCallback,
+)
+from .bounce import (
+    BounceEngine,
+    BounceConfig,
+    BounceTrack,
+    BounceResult,
+    db_to_linear,
+    linear_to_db,
+)
+
+__all__ = [
+    # WAV
+    "WAVWriter",
+    "WAVWriterConfig",
+    "WAVBitDepth",
+    "write_wav",
+    # Recorder
+    "MultiTrackRecorder",
+    "RecorderConfig",
+    "RecorderState",
+    "RecordChannelState",
+    "ChannelRecordStatus",
+    "PunchRegion",
+    # Session
+    "Session",
+    "Take",
+    "TrackInfo",
+    "PunchInfo",
+    # Disk monitor
+    "DiskMonitor",
+    "DiskMonitorConfig",
+    "DiskSpaceCallback",
+    # Bounce
+    "BounceEngine",
+    "BounceConfig",
+    "BounceTrack",
+    "BounceResult",
+    "db_to_linear",
+    "linear_to_db",
+]

src/recording/bounce.py

@@ -0,0 +1,344 @@
+"""Stereo bounce / export engine.
+
+Mixes down multi-track recordings to a stereo master WAV file,
+applying channel volume, pan, and the master bus state.
+
+Supports:
+- Bounce any combination of track files to stereo
+- Per-channel volume and pan from mixer state
+- Master bus volume/mute/dim
+- Selectable bit depth (16/24/32 bit)
+- Normalisation (peak or LUFS target)
+- Preview: dry run that reports estimated file size
+
+Usage:
+    bounce = BounceEngine(sample_rate=48000)
+    bounce.add_track("tracks/ch00.wav", channel=0, volume_db=0, pan=0)
+    bounce.add_track("tracks/ch01.wav", channel=1, volume_db=-3, pan=0.5)
+    bounce.bounce("output/master.wav", bit_depth=24, normalise=True)
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+
+from .wav_writer import WAVWriter, WAVBitDepth, write_wav
+
+logger = logging.getLogger(__name__)
+
+
+# ── Constants ─────────────────────────────────────────────────────────────────
+
+# dB to linear conversion
+_DB_EPSILON = 1e-10  # floor for -inf dB
+
+
+def db_to_linear(db: float) -> float:
+    """Convert dB to linear gain.
+
+    -inf → 0.0, 0 dB → 1.0, +6 dB → 2.0, -6 dB → 0.5
+    """
+    return 10.0 ** (db / 20.0)
+
+
+def linear_to_db(linear: float) -> float:
+    """Convert linear gain to dB.
+
+    1.0 → 0 dB, 0.5 → -6 dB, 2.0 → +6 dB
+    """
+    if linear <= _DB_EPSILON:
+        return -144.0  # effectively -inf
+    return 20.0 * math.log10(linear)
+
+
+# ── Types ─────────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class BounceTrack:
+    """A single track to include in the bounce."""
+    file_path: str | Path   # path to WAV file (mono or stereo)
+    channel: int = 0        # mixer channel index (for reference)
+    volume_db: float = 0.0  # channel fader level in dB
+    pan: float = 0.0        # -1.0 (full left) to 1.0 (full right)
+    muted: bool = False
+
+    def __post_init__(self):
+        self.file_path = Path(self.file_path)
+
+
+@dataclass
+class BounceConfig:
+    """Configuration for a bounce operation."""
+    sample_rate: int = 48000
+    bit_depth: WAVBitDepth = WAVBitDepth.INT24
+    normalise: bool = True          # normalise peak to 0 dBFS
+    normalise_target_db: float = -0.3  # target peak in dBFS
+    master_volume_db: float = 0.0   # master fader dB
+    master_muted: bool = False
+    master_dim_db: float = -20.0
+    master_dim_active: bool = False
+    block_size_frames: int = 4096   # processing block size for memory efficiency
+
+
+@dataclass
+class BounceResult:
+    """Result of a bounce operation."""
+    output_path: Path
+    frames_written: int
+    duration_seconds: float
+    peak_db: float                  # peak level in dBFS
+    rms_db: float                   # RMS level in dBFS
+    files_processed: int
+
+
+# ── Bounce engine ─────────────────────────────────────────────────────────────
+
+
+class BounceEngine:
+    """Stereo bounce engine for multi-track mixdown.
+
+    Processes track files in blocks to keep memory usage low.
+    Supports large multi-track projects on embedded hardware.
+    """
+
+    def __init__(
+        self,
+        sample_rate: int = 48000,
+        *,
+        config: BounceConfig | None = None,
+    ):
+        self.sample_rate = sample_rate
+        self.config = config or BounceConfig(sample_rate=sample_rate)
+        self._tracks: list[BounceTrack] = []
+
+    # ── Track setup ───────────────────────────────────────────────────────
+
+    def add_track(
+        self,
+        file_path: str | Path,
+        channel: int = 0,
+        volume_db: float = 0.0,
+        pan: float = 0.0,
+        muted: bool = False,
+    ) -> None:
+        """Add a track to the bounce list."""
+        self._tracks.append(BounceTrack(
+            file_path=Path(file_path),
+            channel=channel,
+            volume_db=volume_db,
+            pan=pan,
+            muted=muted,
+        ))
+
+    def remove_track(self, channel: int) -> None:
+        """Remove all tracks for a given channel."""
+        self._tracks = [t for t in self._tracks if t.channel != channel]
+
+    def clear_tracks(self) -> None:
+        """Clear all tracks."""
+        self._tracks.clear()
+
+    # ── Bounce ────────────────────────────────────────────────────────────
+
+    def bounce(
+        self,
+        output_path: str | Path,
+        *,
+        bit_depth: WAVBitDepth | int | None = None,
+        normalise: bool | None = None,
+    ) -> BounceResult:
+        """Perform the stereo bounce.
+
+        Args:
+            output_path: Output WAV file path.
+            bit_depth: Override config bit depth.
+            normalise: Override config normalisation.
+
+        Returns:
+            BounceResult with output path, frame count, duration, and levels.
+        """
+        if not self._tracks:
+            raise ValueError("No tracks added — call add_track() first")
+
+        output_path = Path(output_path)
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+
+        bd = self.config.bit_depth if bit_depth is None else (
+            WAVBitDepth(bit_depth) if isinstance(bit_depth, int) else bit_depth
+        )
+        do_normalise = normalise if normalise is not None else self.config.normalise
+
+        # Compute master gain
+        master_gain = self._compute_master_gain()
+
+        # First pass: determine max length
+        max_frames = 0
+        samples_dict: dict[str, np.ndarray] = {}  # load all tracks
+
+        for track in self._tracks:
+            if track.muted:
+                continue
+            samples = self._load_audio(Path(track.file_path))
+            if samples.ndim == 1:
+                samples = samples.reshape(-1, 1)
+            if samples.shape[0] > max_frames:
+                max_frames = samples.shape[0]
+            samples_dict[str(track.file_path)] = samples
+
+        if max_frames == 0:
+            raise ValueError("No audio data to bounce (all tracks empty or muted)")
+
+        # Process in blocks
+        block_size = self.config.block_size_frames
+        stereo_out = np.zeros((max_frames, 2), dtype=np.float64)  # accumulate in float64
+
+        for track in self._tracks:
+            if track.muted:
+                continue
+
+            samples = samples_dict[str(track.file_path)]
+            ch_gain = db_to_linear(track.volume_db)
+
+            num_frames = samples.shape[0]
+            num_ch = samples.shape[1]
+
+            # Compute pan gains
+            left_gain, right_gain = _pan_to_gains(track.pan)
+
+            for start in range(0, num_frames, block_size):
+                end = min(start + block_size, num_frames)
+                block = samples[start:end, :].astype(np.float64)
+
+                # Convert to stereo
+                if num_ch == 1:
+                    stereo_out[start:end, 0] += block[:, 0] * ch_gain * left_gain
+                    stereo_out[start:end, 1] += block[:, 0] * ch_gain * right_gain
+                else:
+                    # Already stereo — pan is interpreted as balance
+                    stereo_out[start:end, 0] += block[:, 0] * ch_gain * left_gain
+                    stereo_out[start:end, 1] += block[:, 1] * ch_gain * right_gain
+
+        # Apply master gain
+        stereo_out *= master_gain
+
+        # Measure levels before normalisation
+        peak = float(np.max(np.abs(stereo_out)))
+        rms = float(np.sqrt(np.mean(stereo_out ** 2)))
+        peak_db = linear_to_db(peak) if peak > 0 else -144.0
+        rms_db = linear_to_db(rms) if rms > 0 else -144.0
+
+        # Normalise
+        if do_normalise and peak > 0:
+            target_linear = db_to_linear(self.config.normalise_target_db)
+            norm_gain = target_linear / peak
+            stereo_out *= norm_gain
+            peak_db = self.config.normalise_target_db
+
+        # Convert to float32 for writing
+        stereo_out_f32 = np.clip(stereo_out, -1.0, 1.0).astype(np.float32)
+
+        # Write
+        write_wav(output_path, stereo_out_f32, sample_rate=self.sample_rate, bit_depth=bd)
+
+        duration = max_frames / self.sample_rate if self.sample_rate > 0 else 0.0
+
+        logger.info(
+            "Bounce complete: %s (%d frames, %.1fs, peak=%+.1f dBFS, RMS=%+.1f dBFS)",
+            output_path.name, max_frames, duration, peak_db, rms_db,
+        )
+
+        return BounceResult(
+            output_path=output_path,
+            frames_written=max_frames,
+            duration_seconds=duration,
+            peak_db=peak_db,
+            rms_db=rms_db,
+            files_processed=len(self._tracks),
+        )
+
+    # ── Preview ───────────────────────────────────────────────────────────
+
+    def preview(self, output_path: str | Path | None = None) -> dict:
+        """Dry-run preview of bounce: returns expected stats without writing.
+
+        Returns a dict with estimated_file_size_bytes, total_frames, etc.
+        """
+        if not self._tracks:
+            return {"error": "No tracks added"}
+
+        max_frames = 0
+        for track in self._tracks:
+            if track.muted:
+                continue
+            samples = self._load_audio(Path(track.file_path))
+            if samples.ndim == 1:
+                nf = samples.shape[0]
+            else:
+                nf = samples.shape[0]
+            max_frames = max(max_frames, nf)
+
+        abs_depth = abs(int(self.config.bit_depth))
+        bytes_per_sample = abs_depth // 8
+        estimated_bytes = max_frames * 2 * bytes_per_sample  # stereo
+
+        return {
+            "total_frames": max_frames,
+            "duration_seconds": max_frames / self.sample_rate if self.sample_rate > 0 else 0,
+            "estimated_file_size_bytes": estimated_bytes,
+            "estimated_file_size_mb": estimated_bytes / (1024.0 * 1024.0),
+            "track_count": len([t for t in self._tracks if not t.muted]),
+            "sample_rate": self.sample_rate,
+            "bit_depth": self.config.bit_depth.name,
+        }
+
+    # ── Helpers ────────────────────────────────────────────────────────────
+
+    def _compute_master_gain(self) -> float:
+        """Compute master bus linear gain from config."""
+        if self.config.master_muted:
+            return 0.0
+
+        gain_db = self.config.master_volume_db
+        if self.config.master_dim_active:
+            gain_db += self.config.master_dim_db
+
+        return db_to_linear(gain_db)
+
+    @staticmethod
+    def _load_audio(file_path: Path) -> np.ndarray:
+        """Load audio file to float32 numpy array.
+
+        Uses soundfile if available, otherwise falls back to the backing
+        track loader.
+        """
+        try:
+            import soundfile as sf
+            samples, sr = sf.read(str(file_path), dtype="float32", always_2d=False)
+            return samples
+        except ImportError:
+            # Fallback: use backing track loader
+            from ..backing.loader import load_audio
+            audio = load_audio(file_path)
+            return audio.samples
+
+
+# ── Pan law ───────────────────────────────────────────────────────────────────
+
+def _pan_to_gains(pan: float) -> tuple[float, float]:
+    """Convert pan value [-1, 1] to left/right gain pair.
+
+    Uses -3 dB constant power pan law.
+    -1.0 = full left, 0.0 = center, 1.0 = full right.
+    """
+    # Normalise pan to angle [0, pi/2]
+    angle = (pan + 1.0) * 0.25 * math.pi  # maps [-1,1] → [0, pi/2]
+    left = math.cos(angle)
+    right = math.sin(angle)
+    return left, right

src/recording/disk_monitor.py

@@ -0,0 +1,252 @@
+"""Free disk space monitoring for recording sessions.
+
+The DiskMonitor periodically checks available disk space on the recording
+target and emits warnings when space falls below configurable thresholds.
+Critical for preventing data loss on SD card-based recording setups.
+
+Thresholds:
+- WARNING: low space, alert user but continue recording
+- CRITICAL: critically low, should stop recording to avoid corruption
+- MINIMUM_REQUIRED: absolute minimum required to start a recording session
+
+Usage:
+    monitor = DiskMonitor(path="/data/recordings")
+    monitor.start()
+    if monitor.is_low_space():
+        print(f"Warning: {monitor.free_gb:.1f} GB remaining")
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import shutil
+import threading
+import time
+from dataclasses import dataclass
+from typing import Callable, Optional
+
+logger = logging.getLogger(__name__)
+
+
+# ── Types ─────────────────────────────────────────────────────────────────────
+
+DiskSpaceCallback = Callable[[int, int, int], None]
+# callback(free_bytes, total_bytes, threshold_exceeded_level)
+# threshold_exceeded_level: 0=none, 1=warning, 2=critical
+
+
+@dataclass
+class DiskMonitorConfig:
+    """Configuration for disk space monitoring."""
+    # Thresholds in bytes
+    warning_threshold_bytes: int = 500 * 1024 * 1024       # 500 MB
+    critical_threshold_bytes: int = 100 * 1024 * 1024       # 100 MB
+    minimum_required_bytes: int = 50 * 1024 * 1024          # 50 MB
+
+    # Recording rate estimation
+    estimated_bytes_per_second: int = 172800               # 48kHz * 24bit * 1ch = ~144KB/s
+    safety_buffer_seconds: int = 120                        # 2 minutes of recording buffer
+
+    # Polling
+    poll_interval_seconds: float = 5.0                     # how often to check
+
+
+# ── Disk monitor ──────────────────────────────────────────────────────────────
+
+
+class DiskMonitor:
+    """Monitors free disk space on a recording target path.
+
+    Runs a background thread that periodically polls available disk
+    space. Fires callbacks when thresholds are crossed to alert the
+    user or trigger automatic stop.
+
+    Also estimates how much recording time remains based on configured
+    bit rate and current free space.
+    """
+
+    def __init__(
+        self,
+        path: str,
+        *,
+        config: DiskMonitorConfig | None = None,
+        callback: DiskSpaceCallback | None = None,
+    ):
+        self.path = os.path.abspath(path)
+        self.config = config or DiskMonitorConfig()
+        self._callback = callback
+
+        # State
+        self._running: bool = False
+        self._thread: Optional[threading.Thread] = None
+        self._lock = threading.Lock()
+
+        # Current stats (updated by background thread)
+        self._free_bytes: int = 0
+        self._total_bytes: int = 0
+        self._last_check: float = 0.0
+        self._current_level: int = 0  # 0=ok, 1=warning, 2=critical
+
+        # Run initial check
+        self._poll()
+
+    # ── Lifecycle ──────────────────────────────────────────────────────────
+
+    def start(self) -> None:
+        """Start background monitoring."""
+        with self._lock:
+            if self._running:
+                return
+            self._running = True
+            self._thread = threading.Thread(
+                target=self._monitor_loop,
+                name="disk-monitor",
+                daemon=True,
+            )
+            self._thread.start()
+            logger.info("Disk monitor started: %s", self.path)
+
+    def stop(self) -> None:
+        """Stop background monitoring."""
+        with self._lock:
+            self._running = False
+
+        if self._thread and self._thread.is_alive():
+            self._thread.join(timeout=5.0)
+
+    # ── Query ──────────────────────────────────────────────────────────────
+
+    def check_now(self) -> int:
+        """Force an immediate space check. Returns free bytes."""
+        self._poll()
+        return self._free_bytes
+
+    @property
+    def free_bytes(self) -> int:
+        return self._free_bytes
+
+    @property
+    def free_mb(self) -> float:
+        return self._free_bytes / (1024.0 * 1024.0)
+
+    @property
+    def free_gb(self) -> float:
+        return self._free_bytes / (1024.0 ** 3)
+
+    @property
+    def total_bytes(self) -> int:
+        return self._total_bytes
+
+    @property
+    def total_gb(self) -> float:
+        return self._total_bytes / (1024.0 ** 3)
+
+    @property
+    def usage_percent(self) -> float:
+        if self._total_bytes > 0:
+            return 100.0 * (1.0 - self._free_bytes / self._total_bytes)
+        return 0.0
+
+    def is_low_space(self) -> bool:
+        """True if free space is below warning threshold."""
+        return self._free_bytes < self.config.warning_threshold_bytes
+
+    def is_critical(self) -> bool:
+        """True if free space is below critical threshold."""
+        return self._free_bytes < self.config.critical_threshold_bytes
+
+    def has_minimum_for_recording(self) -> bool:
+        """True if enough space to start a recording session.
+
+        Checks that free space exceeds the minimum required plus
+        the safety buffer (estimated recording for safety_buffer_seconds).
+        """
+        required = (
+            self.config.minimum_required_bytes
+            + self.config.estimated_bytes_per_second * self.config.safety_buffer_seconds
+        )
+        return self._free_bytes >= required
+
+    def estimated_recording_time_seconds(self) -> float:
+        """Estimate how many more seconds can be recorded at current rate."""
+        if self.config.estimated_bytes_per_second <= 0:
+            return float("inf")
+        available = max(0, self._free_bytes - self.config.minimum_required_bytes)
+        return available / self.config.estimated_bytes_per_second
+
+    @property
+    def level(self) -> int:
+        """Current threshold level: 0=ok, 1=warning, 2=critical."""
+        return self._current_level
+
+    @property
+    def level_label(self) -> str:
+        return {0: "ok", 1: "warning", 2: "critical"}.get(self._current_level, "unknown")
+
+    # ── Callback management ────────────────────────────────────────────────
+
+    def set_callback(self, callback: DiskSpaceCallback | None) -> None:
+        self._callback = callback
+
+    # ── Internal ───────────────────────────────────────────────────────────
+
+    def _poll(self) -> None:
+        """Check disk space using statvfs."""
+        try:
+            stat = os.statvfs(self.path)
+            self._free_bytes = stat.f_frsize * stat.f_bavail
+            self._total_bytes = stat.f_frsize * stat.f_blocks
+            self._last_check = time.monotonic()
+
+            # Determine threshold level
+            old_level = self._current_level
+            if self._free_bytes < self.config.critical_threshold_bytes:
+                self._current_level = 2
+            elif self._free_bytes < self.config.warning_threshold_bytes:
+                self._current_level = 1
+            else:
+                self._current_level = 0
+
+            # Fire callback on level change
+            if self._current_level != old_level and self._callback:
+                try:
+                    self._callback(self._free_bytes, self._total_bytes, self._current_level)
+                except Exception:
+                    logger.error("Disk space callback error", exc_info=True)
+
+        except OSError as exc:
+            logger.error("Disk space check failed for %s: %s", self.path, exc)
+
+    def _monitor_loop(self) -> None:
+        """Background monitoring loop."""
+        while self._running:
+            self._poll()
+            time.sleep(self.config.poll_interval_seconds)
+
+        logger.debug("Disk monitor loop stopped")
+
+    # ── Utility ─────────────────────────────────────────────────────────────
+
+    @staticmethod
+    def get_free_bytes(path: str) -> int:
+        """Static helper: get free bytes on a path without instantiating."""
+        stat = os.statvfs(path)
+        return stat.f_frsize * stat.f_bavail
+
+    @staticmethod
+    def get_total_bytes(path: str) -> int:
+        """Static helper: get total bytes on a path."""
+        stat = os.statvfs(path)
+        return stat.f_frsize * stat.f_blocks
+
+    @staticmethod
+    def format_bytes(bytes_val: int) -> str:
+        """Human-readable byte count."""
+        if bytes_val >= 1024 ** 3:
+            return f"{bytes_val / (1024**3):.1f} GB"
+        elif bytes_val >= 1024 ** 2:
+            return f"{bytes_val / (1024**2):.1f} MB"
+        elif bytes_val >= 1024:
+            return f"{bytes_val / 1024:.1f} KB"
+        return f"{bytes_val} B"

src/recording/recorder.py

@@ -0,0 +1,535 @@
+"""Multi-track recorder with disk streaming and write-ahead buffering.
+
+The MultiTrackRecorder captures audio from JACK capture ports and writes
+each channel to a separate WAV file. A background writer thread drains
+a buffer queue to disk, preventing SD card write latency from causing
+xruns in the real-time audio thread.
+
+Key features:
+- Per-channel recording with independent arm/punch states
+- Write-ahead ring buffer (configurable size) to decouple capture from disk I/O
+- Punch in/out: start/stop recording on individual channels mid-session
+- Transport sync: start/stop/pause linked to JACK transport
+- Take management: each recording session produces a take with metadata
+
+Usage:
+    rec = MultiTrackRecorder(session_dir="/data/session_001")
+    rec.arm_channel(0)
+    rec.arm_channel(1)
+    rec.start()                # begin recording
+    # ... JACK process callback ...
+    rec.write(0, audio_chunk)  # float32 numpy array
+    rec.write(1, audio_chunk)
+    rec.stop()                 # finalise files
+"""
+
+from __future__ import annotations
+
+import logging
+import queue
+import threading
+import time
+from dataclasses import dataclass, field
+from enum import Enum, auto
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+
+from .wav_writer import WAVWriter, WAVBitDepth
+
+logger = logging.getLogger(__name__)
+
+
+# ── Types ─────────────────────────────────────────────────────────────────────
+
+
+class RecorderState(Enum):
+    """Top-level recorder state."""
+    IDLE = auto()
+    RECORDING = auto()
+    PAUSED = auto()
+
+
+class ChannelState(Enum):
+    """Per-channel record state."""
+    DISARMED = auto()   # not recording, not armed
+    ARMED = auto()      # ready to record when transport starts
+    RECORDING = auto()  # actively capturing
+
+
+@dataclass
+class PunchRegion:
+    """A punch-in/out time range on a channel."""
+    start_frame: int       # frame index when punch-in occurred
+    end_frame: int | None  # None if still recording
+
+
+@dataclass
+class ChannelRecordStatus:
+    """Per-channel recording metadata."""
+    channel: int
+    state: ChannelState = ChannelState.DISARMED
+    writer: Optional[WAVWriter] = None
+    frames_recorded: int = 0
+    punch_regions: list[PunchRegion] = field(default_factory=list)
+    current_punch_start: int | None = None
+    file_path: str = ""
+
+
+@dataclass
+class RecorderConfig:
+    """Configuration for the multi-track recorder."""
+    # Audio
+    sample_rate: int = 48000
+    channels: int = 16
+    bit_depth: WAVBitDepth = WAVBitDepth.INT24
+
+    # Buffer
+    buffer_chunk_frames: int = 256  # frames per queue item
+    max_queue_size: int = 128       # max chunks per channel in buffer queue
+    writer_flush_interval_frames: int = 2048  # flush writer every N frames
+
+    # Session
+    session_dir: str = "."
+
+    # JACK
+    jack_port_prefix: str = "system"  # capture port prefix
+
+
+# ── Multi-track recorder ──────────────────────────────────────────────────────
+
+
+class MultiTrackRecorder:
+    """Multi-track disk-streaming recorder.
+
+    Architecture:
+        JACK process callback → rec.write(ch, samples)
+            → buffer_queue[ch].put(chunk)
+        Writer thread → buffer_queue[ch].get(chunk)
+            → WAVWriter.write_frames(chunk)
+
+    The buffer queue decouples the real-time capture thread from
+    the SD card write latency, preventing xruns.
+    """
+
+    def __init__(self, config: RecorderConfig | None = None, *, session_dir: str | None = None):
+        self.config = config or RecorderConfig()
+
+        if session_dir:
+            self.config.session_dir = session_dir
+
+        # State
+        self._state: RecorderState = RecorderState.IDLE
+        self._start_time: float = 0.0
+        self._global_frame: int = 0
+
+        # Per-channel status
+        self._channels: list[ChannelRecordStatus] = [
+            ChannelRecordStatus(channel=i)
+            for i in range(self.config.channels)
+        ]
+
+        # Write buffers: one queue per channel
+        # Each queue item is a tuple: (np.ndarray, int) where int is the
+        # starting global frame index (for later alignment/stamping)
+        self._buffer_queues: list[queue.Queue] = [
+            queue.Queue(maxsize=self.config.max_queue_size)
+            for _ in range(self.config.channels)
+        ]
+
+        # Background writer
+        self._writer_thread: Optional[threading.Thread] = None
+        self._writer_stop_event = threading.Event()
+        self._writer_error: Optional[Exception] = None
+
+        # Thread safety
+        self._lock = threading.Lock()
+
+        # Ensure session directory exists
+        Path(self.config.session_dir).mkdir(parents=True, exist_ok=True)
+
+    # ── State ──────────────────────────────────────────────────────────────
+
+    @property
+    def state(self) -> RecorderState:
+        return self._state
+
+    @property
+    def is_recording(self) -> bool:
+        return self._state == RecorderState.RECORDING
+
+    @property
+    def global_frame(self) -> int:
+        return self._global_frame
+
+    def get_channel_state(self, channel: int) -> ChannelState:
+        if 0 <= channel < len(self._channels):
+            return self._channels[channel].state
+        return ChannelState.DISARMED
+
+    # ── Channel arming ────────────────────────────────────────────────────
+
+    def arm_channel(self, channel: int) -> bool:
+        """Arm a channel for recording.
+
+        Armed channels start recording when transport starts.
+        """
+        if not 0 <= channel < len(self._channels):
+            logger.warning("Invalid channel index: %d", channel)
+            return False
+
+        with self._lock:
+            ch = self._channels[channel]
+            if ch.state == ChannelState.DISARMED:
+                ch.state = ChannelState.ARMED
+                logger.debug("CH%d armed", channel)
+        return True
+
+    def disarm_channel(self, channel: int) -> bool:
+        """Disarm a channel. Stops recording if active."""
+        if not 0 <= channel < len(self._channels):
+            return False
+
+        with self._lock:
+            ch = self._channels[channel]
+            if ch.state == ChannelState.RECORDING:
+                self._punch_out_channel(ch)
+            ch.state = ChannelState.DISARMED
+            logger.debug("CH%d disarmed", channel)
+        return True
+
+    def is_armed(self, channel: int) -> bool:
+        if 0 <= channel < len(self._channels):
+            return self._channels[channel].state in (ChannelState.ARMED, ChannelState.RECORDING)
+        return False
+
+    def get_armed_channels(self) -> list[int]:
+        return [
+            i for i, ch in enumerate(self._channels)
+            if ch.state in (ChannelState.ARMED, ChannelState.RECORDING)
+        ]
+
+    # ── Recording lifecycle ───────────────────────────────────────────────
+
+    def start(self) -> bool:
+        """Start recording on all armed channels."""
+        if self._state == RecorderState.RECORDING:
+            logger.warning("Already recording")
+            return False
+
+        with self._lock:
+            self._state = RecorderState.RECORDING
+            self._start_time = time.monotonic()
+            self._global_frame = 0
+            self._writer_error = None
+            self._writer_stop_event.clear()
+
+            # Start recording on all armed channels
+            active_count = 0
+            for ch in self._channels:
+                if ch.state == ChannelState.ARMED:
+                    self._start_channel_recording(ch)
+                    active_count += 1
+
+        if active_count == 0:
+            logger.warning("No armed channels — starting with no active tracks")
+
+        # Start background writer thread
+        self._writer_thread = threading.Thread(
+            target=self._writer_loop,
+            name="rec-writer",
+            daemon=True,
+        )
+        self._writer_thread.start()
+
+        logger.info("Recording started: %d channels, sr=%d, bit=%s",
+                     active_count, self.config.sample_rate, self.config.bit_depth.name)
+        return True
+
+    def stop(self) -> None:
+        """Stop recording and finalise all files."""
+        if self._state == RecorderState.IDLE:
+            return
+
+        with self._lock:
+            self._state = RecorderState.IDLE
+
+            # Punch out all active channels
+            for ch in self._channels:
+                if ch.state == ChannelState.RECORDING:
+                    self._punch_out_channel(ch)
+
+        # Signal writer thread to finish
+        self._writer_stop_event.set()
+
+        # Send sentinel values to all queues to unblock writer
+        for q in self._buffer_queues:
+            try:
+                q.put(None, timeout=0.1)
+            except queue.Full:
+                pass
+
+        # Wait for writer thread
+        if self._writer_thread and self._writer_thread.is_alive():
+            self._writer_thread.join(timeout=10.0)
+            if self._writer_thread.is_alive():
+                logger.warning("Writer thread did not stop in 10s")
+
+        # Close all writers
+        with self._lock:
+            for ch in self._channels:
+                if ch.writer:
+                    ch.writer.close()
+                    ch.writer = None
+
+        total_frames = sum(ch.frames_recorded for ch in self._channels)
+        duration = self._global_frame / self.config.sample_rate if self.config.sample_rate else 0
+        logger.info("Recording stopped: %d total frames (%.1fs)",
+                     total_frames, duration)
+
+    def pause(self) -> None:
+        """Pause recording without closing files."""
+        if self._state != RecorderState.RECORDING:
+            return
+        with self._lock:
+            self._state = RecorderState.PAUSED
+            # Punch out all active channels
+            for ch in self._channels:
+                if ch.state == ChannelState.RECORDING:
+                    self._punch_out_channel(ch)
+        logger.info("Recording paused at frame %d", self._global_frame)
+
+    def resume(self) -> None:
+        """Resume recording after pause."""
+        if self._state != RecorderState.PAUSED:
+            return
+        with self._lock:
+            self._state = RecorderState.RECORDING
+            # Punch in on previously recording channels
+            for ch in self._channels:
+                if ch.state == ChannelState.ARMED:
+                    self._punch_in_channel(ch)
+        logger.info("Recording resumed at frame %d", self._global_frame)
+
+    # ── Punch in/out ──────────────────────────────────────────────────────
+
+    def punch_in(self, channel: int) -> bool:
+        """Start recording on a specific channel mid-session."""
+        if self._state != RecorderState.RECORDING:
+            logger.warning("Cannot punch in while not recording")
+            return False
+        if not 0 <= channel < len(self._channels):
+            return False
+
+        with self._lock:
+            ch = self._channels[channel]
+            if ch.state == ChannelState.DISARMED:
+                ch.state = ChannelState.ARMED  # arm first
+            if ch.state == ChannelState.ARMED:
+                self._punch_in_channel(ch)
+                logger.info("CH%d punch in at frame %d", channel, self._global_frame)
+                return True
+        return False
+
+    def punch_out(self, channel: int) -> bool:
+        """Stop recording on a specific channel mid-session."""
+        if self._state != RecorderState.RECORDING:
+            logger.warning("Cannot punch out while not recording")
+            return False
+        if not 0 <= channel < len(self._channels):
+            return False
+
+        with self._lock:
+            ch = self._channels[channel]
+            if ch.state == ChannelState.RECORDING:
+                self._punch_out_channel(ch)
+                logger.info("CH%d punch out at frame %d", channel, self._global_frame)
+                return True
+        return False
+
+    def _start_channel_recording(self, ch: ChannelRecordStatus) -> None:
+        """Start recording on a channel that was armed."""
+        self._punch_in_channel(ch)
+
+    def _punch_in_channel(self, ch: ChannelRecordStatus) -> None:
+        """Internal: start recording on a channel."""
+        # Close existing writer if any, create new file
+        if ch.writer:
+            ch.writer.close()
+
+        # Generate file path for this punch region
+        punch_count = len(ch.punch_regions)
+        if punch_count == 0:
+            suffix = ""
+        else:
+            suffix = f"_punch{punch_count}"
+
+        file_path = Path(self.config.session_dir) / f"ch{ch.channel:02d}{suffix}.wav"
+
+        ch.writer = WAVWriter(
+            file_path,
+            sample_rate=self.config.sample_rate,
+            channels=1,  # mono per channel
+            bit_depth=self.config.bit_depth,
+        )
+        ch.file_path = str(file_path)
+        ch.state = ChannelState.RECORDING
+        ch.current_punch_start = self._global_frame
+
+        # Register punch region
+        ch.punch_regions.append(PunchRegion(
+            start_frame=self._global_frame,
+            end_frame=None,
+        ))
+
+    def _punch_out_channel(self, ch: ChannelRecordStatus) -> None:
+        """Internal: stop recording on a channel."""
+        ch.state = ChannelState.ARMED
+        ch.current_punch_start = None
+
+        # Close punch region
+        if ch.punch_regions:
+            region = ch.punch_regions[-1]
+            region.end_frame = self._global_frame
+
+        # Flush and finalise writer
+        if ch.writer:
+            ch.writer.flush()
+            # Don't close — allow re-punch into same file
+            # For punch regions, we close and start new files each time
+
+    # ── Audio write (real-time path) ─────────────────────────────────────
+
+    def write(self, channel: int, samples: np.ndarray) -> bool:
+        """Write audio samples to a channel's buffer queue.
+
+        Call this from the JACK process callback or audio capture thread.
+        This is the real-time path — it must never block on disk I/O.
+
+        Args:
+            channel: 0-based channel index.
+            samples: float32 numpy array, shape (num_frames,). Values in [-1.0, 1.0].
+
+        Returns:
+            True if samples were queued, False if buffer full (drop).
+        """
+        if self._state != RecorderState.RECORDING:
+            return False
+
+        if not 0 <= channel < len(self._channels):
+            return False
+
+        ch = self._channels[channel]
+        if ch.state != ChannelState.RECORDING:
+            return False
+
+        if samples.ndim != 1:
+            samples = samples.ravel()
+
+        # Copy samples (must not reference JACK buffer directly)
+        chunk = np.array(samples, dtype=np.float32, copy=True)
+
+        try:
+            self._buffer_queues[channel].put_nowait(chunk)
+            return True
+        except queue.Full:
+            logger.warning("Buffer full for CH%d — dropping %d frames",
+                          channel, len(chunk))
+            return False
+
+    def advance_frame(self, frames: int = 1) -> None:
+        """Advance the global frame counter.
+
+        Call this after each buffer period to track recording position.
+        """
+        self._global_frame += frames
+
+    # ── Background writer thread ──────────────────────────────────────────
+
+    def _writer_loop(self) -> None:
+        """Background thread: drain buffer queues and write to disk.
+
+        Runs until _writer_stop_event is set and all queues are drained.
+        """
+        logger.debug("Writer thread started")
+
+        # Track which queues have been signalled (sentinel None received)
+        queues_signalled: set[int] = set()
+        num_queues = len(self._buffer_queues)
+
+        try:
+            while not self._writer_stop_event.is_set() or len(queues_signalled) < num_queues:
+                wrote_anything = False
+
+                for ch_idx, q in enumerate(self._buffer_queues):
+                    if ch_idx in queues_signalled:
+                        continue
+
+                    try:
+                        chunk = q.get(timeout=0.01)
+                    except queue.Empty:
+                        continue
+
+                    if chunk is None:
+                        # Sentinel: this queue is done
+                        queues_signalled.add(ch_idx)
+                        continue
+
+                    # Write chunk to disk
+                    with self._lock:
+                        ch = self._channels[ch_idx]
+                        if ch.writer is not None:
+                            ch.writer.write_frames(chunk)
+                            ch.frames_recorded += len(chunk)
+
+                    wrote_anything = True
+
+                if not wrote_anything and not self._writer_stop_event.is_set():
+                    # Short sleep to avoid busy-waiting when queues are empty
+                    time.sleep(0.005)
+
+        except Exception as exc:
+            self._writer_error = exc
+            logger.error("Writer thread error: %s", exc, exc_info=True)
+        finally:
+            logger.debug("Writer thread finished (wrote %d frames)",
+                        sum(ch.frames_recorded for ch in self._channels))
+
+    # ── Status / info ─────────────────────────────────────────────────────
+
+    @property
+    def writer_error(self) -> Optional[Exception]:
+        return self._writer_error
+
+    def get_recording_stats(self) -> dict:
+        """Get recording statistics."""
+        total_frames = sum(ch.frames_recorded for ch in self._channels)
+        return {
+            "state": self._state.name,
+            "global_frame": self._global_frame,
+            "total_frames_recorded": total_frames,
+            "elapsed_seconds": self._global_frame / self.config.sample_rate if self.config.sample_rate else 0,
+            "samplerate": self.config.sample_rate,
+            "channels": {
+                i: {
+                    "state": ch.state.name,
+                    "frames": ch.frames_recorded,
+                    "punch_regions": len(ch.punch_regions),
+                    "file": ch.file_path,
+                }
+                for i, ch in enumerate(self._channels)
+                if ch.state != ChannelState.DISARMED
+            },
+            "buffer_queues": {
+                i: q.qsize()
+                for i, q in enumerate(self._buffer_queues)
+                if q.qsize() > 0
+            },
+            "writer_error": str(self._writer_error) if self._writer_error else None,
+        }
+
+    def get_channel_record_status(self, channel: int) -> ChannelRecordStatus | None:
+        if 0 <= channel < len(self._channels):
+            return self._channels[channel]
+        return None

src/recording/session.py

@@ -0,0 +1,466 @@
+"""Session file format and take management.
+
+The session file is a JSON document that ties together the mixer state,
+audio references, and recording metadata. Sessions are portable — you
+can archive them, move them between machines, etc.
+
+Session format (JSON):
+    {
+      "version": 1,
+      "name": "Session Name",
+      "created_at": "2024-01-01T00:00:00",
+      "sample_rate": 48000,
+      "bit_depth": "int24",
+      "takes": [ ... ],
+      "mixer_state": { ... },
+      "notes": ""
+    }
+
+Takes:
+    Each take is a recording pass. A take contains one or more track
+    files (WAV), punch region metadata, and per-track mixer snapshots.
+    Multiple takes can exist in a single session (e.g., multiple
+    recording passes of the same song).
+
+Usage:
+    session = Session.create("My Session", sample_rate=48000)
+    take = session.new_take("Take 1")
+    session.add_track(take.id, 0, "tracks/take_01_ch00.wav", mixer_snapshot)
+    session.save("/data/sessions/mysession.session.json")
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+import uuid
+from dataclasses import dataclass, field, asdict
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+
+from .wav_writer import WAVBitDepth
+
+logger = logging.getLogger(__name__)
+
+# Current session file format version
+SESSION_VERSION = 1
+
+
+# ── Data classes ──────────────────────────────────────────────────────────────
+
+
+@dataclass
+class PunchInfo:
+    """Metadata for a punch region within a track."""
+    start_frame: int
+    end_frame: int
+
+
+@dataclass
+class TrackInfo:
+    """Metadata for one recorded track within a take."""
+    channel: int                     # 0-based mixer channel
+    file_path: str                   # relative path to WAV file (from session dir)
+    frames: int = 0                  # total recorded frames
+    punch_regions: list[PunchInfo] = field(default_factory=list)
+    mixer_snapshot: dict = field(default_factory=dict)  # channel mixer state at record time
+
+
+@dataclass
+class Take:
+    """A recording take with one or more tracks."""
+    id: str                          # unique take ID
+    name: str = ""                   # human-readable label
+    created_at: str = ""             # ISO-8601 timestamp
+    tracks: list[TrackInfo] = field(default_factory=list)
+    sample_rate: int = 48000
+    bit_depth: str = "int24"
+    notes: str = ""                  # free-form notes
+    start_frame: int = 0             # global frame when recording started
+    end_frame: int = 0               # global frame when recording ended
+    duration_seconds: float = 0.0    # derived from frames / sample_rate
+
+    def add_track(self, track: TrackInfo) -> None:
+        """Add a track to this take."""
+        self.tracks.append(track)
+
+    def get_track(self, channel: int) -> TrackInfo | None:
+        """Get a track by channel number."""
+        for t in self.tracks:
+            if t.channel == channel:
+                return t
+        return None
+
+    def get_track_file_paths(self) -> list[str]:
+        """Get all track file paths."""
+        return [t.file_path for t in self.tracks]
+
+    def total_frames(self) -> int:
+        """Total frames across all tracks."""
+        if self.end_frame is not None and self.start_frame is not None:
+            return self.end_frame - self.start_frame
+        return max((t.frames for t in self.tracks), default=0)
+
+    def to_dict(self) -> dict:
+        """Serialize take to a dict."""
+        return {
+            "id": self.id,
+            "name": self.name,
+            "created_at": self.created_at,
+            "sample_rate": self.sample_rate,
+            "bit_depth": self.bit_depth,
+            "tracks": [
+                {
+                    "channel": t.channel,
+                    "file_path": t.file_path,
+                    "frames": t.frames,
+                    "punch_regions": [
+                        {"start_frame": p.start_frame, "end_frame": p.end_frame}
+                        for p in t.punch_regions
+                    ],
+                    "mixer_snapshot": t.mixer_snapshot,
+                }
+                for t in self.tracks
+            ],
+            "notes": self.notes,
+            "start_frame": self.start_frame,
+            "end_frame": self.end_frame,
+            "duration_seconds": self.duration_seconds,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict) -> Take:
+        """Deserialize a Take from a dict."""
+        tracks = []
+        for td in data.get("tracks", []):
+            punch_regions = [
+                PunchInfo(start_frame=p["start_frame"], end_frame=p["end_frame"])
+                for p in td.get("punch_regions", [])
+            ]
+            tracks.append(TrackInfo(
+                channel=td["channel"],
+                file_path=td["file_path"],
+                frames=td.get("frames", 0),
+                punch_regions=punch_regions,
+                mixer_snapshot=td.get("mixer_snapshot", {}),
+            ))
+
+        return cls(
+            id=data["id"],
+            name=data.get("name", ""),
+            created_at=data.get("created_at", ""),
+            tracks=tracks,
+            sample_rate=data.get("sample_rate", 48000),
+            bit_depth=data.get("bit_depth", "int24"),
+            notes=data.get("notes", ""),
+            start_frame=data.get("start_frame", 0),
+            end_frame=data.get("end_frame", 0),
+            duration_seconds=data.get("duration_seconds", 0.0),
+        )
+
+
+# ── Session ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class Session:
+    """A recording session.
+
+    A session contains:
+    - Session metadata (name, created_at, sample_rate, bit_depth)
+    - Takes (multiple recording passes)
+    - Mixer state snapshot
+    - Free-form notes
+
+    Sessions are serialised as JSON files. Audio files are referenced
+    by relative paths from the session file's directory.
+    """
+
+    version: int = SESSION_VERSION
+    name: str = "Untitled Session"
+    created_at: str = ""               # ISO-8601
+    modified_at: str = ""              # ISO-8601
+    sample_rate: int = 48000
+    bit_depth: str = "int24"           # "int16", "int24", "int32", "float32"
+    takes: list[Take] = field(default_factory=list)
+    mixer_state: dict = field(default_factory=dict)
+    notes: str = ""
+    _file_path: Optional[str] = None   # path to the session file, set after load/save
+
+    @classmethod
+    def create(
+        cls,
+        name: str,
+        *,
+        sample_rate: int = 48000,
+        bit_depth: str | WAVBitDepth = "int24",
+    ) -> Session:
+        """Create a new session."""
+        if isinstance(bit_depth, WAVBitDepth):
+            bit_depth = _bit_depth_to_str(bit_depth)
+
+        now = _now_iso()
+        return cls(
+            name=name,
+            created_at=now,
+            modified_at=now,
+            sample_rate=sample_rate,
+            bit_depth=bit_depth,
+        )
+
+    # ── Take management ─────────────────────────────────────────────────
+
+    def new_take(
+        self,
+        name: str = "",
+        *,
+        start_frame: int = 0,
+        end_frame: int = 0,
+    ) -> Take:
+        """Create a new take in this session.
+
+        Args:
+            name: Human-readable label (e.g. "Take 1", "Vocal overdub").
+            start_frame: Global frame when recording started.
+            end_frame: Global frame when recording ended (0 if not yet known).
+
+        Returns:
+            The new Take.
+        """
+        take = Take(
+            id=f"take_{uuid.uuid4().hex[:12]}",
+            name=name or f"Take {len(self.takes) + 1}",
+            created_at=_now_iso(),
+            sample_rate=self.sample_rate,
+            bit_depth=self.bit_depth,
+            start_frame=start_frame,
+            end_frame=end_frame,
+            duration_seconds=(end_frame - start_frame) / self.sample_rate if end_frame > start_frame and self.sample_rate > 0 else 0.0,
+        )
+        self.takes.append(take)
+        self.modified_at = _now_iso()
+        logger.info("New take created: %s (%s)", take.name, take.id)
+        return take
+
+    def get_take(self, take_id: str) -> Take | None:
+        """Get a take by ID."""
+        for t in self.takes:
+            if t.id == take_id:
+                return t
+        return None
+
+    def delete_take(self, take_id: str) -> bool:
+        """Delete a take (does NOT delete WAV files)."""
+        for i, t in enumerate(self.takes):
+            if t.id == take_id:
+                del self.takes[i]
+                self.modified_at = _now_iso()
+                logger.info("Take deleted: %s", take_id)
+                return True
+        return False
+
+    def rename_take(self, take_id: str, new_name: str) -> bool:
+        """Rename a take."""
+        take = self.get_take(take_id)
+        if take:
+            take.name = new_name
+            self.modified_at = _now_iso()
+            return True
+        return False
+
+    def list_takes(self) -> list[Take]:
+        """Return all takes, ordered by creation time."""
+        return list(self.takes)
+
+    def get_last_take(self) -> Take | None:
+        """Get the most recent take."""
+        return self.takes[-1] if self.takes else None
+
+    # ── Track management ────────────────────────────────────────────────
+
+    def add_track(
+        self,
+        take_id: str,
+        channel: int,
+        file_path: str,
+        mixer_snapshot: dict | None = None,
+        frames: int = 0,
+    ) -> TrackInfo | None:
+        """Add a track to an existing take.
+
+        Args:
+            take_id: The take to add the track to.
+            channel: 0-based mixer channel number.
+            file_path: Relative path to the WAV file.
+            mixer_snapshot: Channel mixer state at recording time.
+            frames: Number of recorded frames.
+
+        Returns:
+            The new TrackInfo, or None if take not found.
+        """
+        take = self.get_take(take_id)
+        if not take:
+            logger.warning("Take not found: %s", take_id)
+            return None
+
+        track = TrackInfo(
+            channel=channel,
+            file_path=file_path,
+            frames=frames,
+            mixer_snapshot=mixer_snapshot or {},
+        )
+        take.add_track(track)
+        self.modified_at = _now_iso()
+        return track
+
+    def get_all_track_files(self) -> list[str]:
+        """Get all track WAV file paths across all takes."""
+        paths = []
+        for take in self.takes:
+            paths.extend(take.get_track_file_paths())
+        return paths
+
+    # ── Serialisation ───────────────────────────────────────────────────
+
+    def to_dict(self) -> dict:
+        """Serialize session to a dict."""
+        return {
+            "version": self.version,
+            "name": self.name,
+            "created_at": self.created_at,
+            "modified_at": self.modified_at,
+            "sample_rate": self.sample_rate,
+            "bit_depth": self.bit_depth,
+            "takes": [t.to_dict() for t in self.takes],
+            "mixer_state": self.mixer_state,
+            "notes": self.notes,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict) -> Session:
+        """Deserialize a Session from a dict."""
+        takes = [Take.from_dict(td) for td in data.get("takes", [])]
+        return cls(
+            version=data.get("version", SESSION_VERSION),
+            name=data.get("name", "Untitled Session"),
+            created_at=data.get("created_at", ""),
+            modified_at=data.get("modified_at", ""),
+            sample_rate=data.get("sample_rate", 48000),
+            bit_depth=data.get("bit_depth", "int24"),
+            takes=takes,
+            mixer_state=data.get("mixer_state", {}),
+            notes=data.get("notes", ""),
+        )
+
+    def save(self, file_path: str | Path | None = None) -> Path:
+        """Save the session to a JSON file.
+
+        Args:
+            file_path: Target file path. If None, uses the path from a
+                       previous load() or save() call.
+
+        Returns:
+            Path to the saved file.
+        """
+        if file_path is None:
+            if self._file_path:
+                file_path = self._file_path
+            else:
+                raise ValueError("No file path specified and no previous path known")
+
+        file_path = Path(file_path)
+        self.modified_at = _now_iso()
+
+        file_path.parent.mkdir(parents=True, exist_ok=True)
+
+        data = self.to_dict()
+
+        with open(file_path, "w", encoding="utf-8") as f:
+            json.dump(data, f, indent=2, ensure_ascii=False)
+
+        self._file_path = str(file_path)
+        logger.info("Session saved: %s (%d takes)", file_path, len(self.takes))
+        return file_path
+
+    @classmethod
+    def load(cls, file_path: str | Path) -> Session:
+        """Load a session from a JSON file.
+
+        Args:
+            file_path: Path to the session JSON file.
+
+        Returns:
+            The loaded Session.
+        """
+        file_path = Path(file_path)
+
+        with open(file_path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+
+        session = cls.from_dict(data)
+        session._file_path = str(file_path)
+        logger.info("Session loaded: %s (%d takes)", file_path, len(session.takes))
+        return session
+
+    # ── Utility ──────────────────────────────────────────────────────────
+
+    @property
+    def take_count(self) -> int:
+        return len(self.takes)
+
+    @property
+    def total_recorded_frames(self) -> int:
+        """Total recorded frames across all takes."""
+        return sum(t.total_frames() for t in self.takes)
+
+    @property
+    def total_duration_seconds(self) -> float:
+        """Total recording duration across all takes."""
+        if self.sample_rate == 0:
+            return 0.0
+        return self.total_recorded_frames / self.sample_rate
+
+    @property
+    def directory(self) -> Path | None:
+        """The directory containing the session file, or None if unsaved."""
+        if self._file_path:
+            return Path(self._file_path).parent
+        return None
+
+    def resolve_track_path(self, track: TrackInfo) -> Path:
+        """Resolve a relative track file path to an absolute path.
+
+        Relative paths are resolved against the session file's directory.
+        """
+        if os.path.isabs(track.file_path):
+            return Path(track.file_path)
+        if self._file_path:
+            return Path(self._file_path).parent / track.file_path
+        return Path(track.file_path)
+
+    def update_mixer_state(self, state: dict) -> None:
+        """Update the mixer state snapshot."""
+        self.mixer_state = state
+        self.modified_at = _now_iso()
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+
+def _now_iso() -> str:
+    """Return current UTC time as ISO-8601 string."""
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _bit_depth_to_str(bit_depth: WAVBitDepth) -> str:
+    """Convert WAVBitDepth enum to string representation."""
+    mapping = {
+        WAVBitDepth.INT16: "int16",
+        WAVBitDepth.INT24: "int24",
+        WAVBitDepth.INT32: "int32",
+        WAVBitDepth.FLOAT32: "float32",
+    }
+    return mapping.get(bit_depth, "int24")

src/recording/wav_writer.py

@@ -0,0 +1,395 @@
+"""Streaming WAV file writer with configurable bit depth and sample rate.
+
+Writes standard RIFF/WAVE files in chunks for low-latency recording.
+Supports 16-bit integer, 24-bit integer, 32-bit integer, and 32-bit
+float formats. Designed for SD card streaming — writes are buffered
+and flushed to minimise seek operations.
+
+Usage:
+    writer = WAVWriter("track_01.wav", sample_rate=48000, channels=1, bit_depth=24)
+    writer.write_frames(frames)  # np.ndarray, float32 in [-1.0, 1.0]
+    writer.close()               # finalises header
+"""
+
+from __future__ import annotations
+
+import logging
+import struct
+from dataclasses import dataclass
+from enum import IntEnum
+from pathlib import Path
+from typing import BinaryIO, Callable, Optional
+
+import numpy as np
+
+try:
+    import soundfile as sf
+    HAS_SOUNDFILE = True
+except ImportError:
+    HAS_SOUNDFILE = False
+
+logger = logging.getLogger(__name__)
+
+
+# ── WAV format constants ──────────────────────────────────────────────────────
+
+class WAVBitDepth(IntEnum):
+    """Supported WAV bit depths."""
+    INT16 = 16
+    INT24 = 24
+    INT32 = 32
+    FLOAT32 = -32  # negative to distinguish from int32
+
+
+# PCM format tags
+_WAV_FORMAT_PCM = 1        # integer PCM
+_WAV_FORMAT_IEEE_FLOAT = 3 # IEEE 754 float
+
+
+# ── Sample conversion ─────────────────────────────────────────────────────────
+
+def _float32_to_int16(samples: np.ndarray) -> np.ndarray:
+    """Convert float32 [-1.0, 1.0] → int16."""
+    clamped = np.clip(samples, -1.0, 1.0)
+    return (clamped * 32767.0).astype(np.int16)
+
+
+def _float32_to_int24(samples: np.ndarray) -> bytes:
+    """Convert float32 [-1.0, 1.0] → packed 24-bit little-endian bytes.
+
+    Each sample occupies 3 bytes. Numpy has no int24 dtype, so we pack
+    manually via int32 scaling and struct.
+    """
+    clamped = np.clip(samples, -1.0, 1.0)
+    scaled = (clamped * 8388607.0).astype(np.int32)  # 2^23 - 1
+    # Pack each sample as 3-byte little-endian
+    data = bytearray(len(scaled) * 3)
+    for i, val in enumerate(scaled):
+        # Mask to 24 bits signed, then pack
+        packed = val & 0xFFFFFF
+        if val < 0:
+            packed |= 0x800000  # sign-extend
+        data[i * 3] = packed & 0xFF
+        data[i * 3 + 1] = (packed >> 8) & 0xFF
+        data[i * 3 + 2] = (packed >> 16) & 0xFF
+    return bytes(data)
+
+
+_FAST_INT24 = False  # set by first call to _float32_to_int24_fast
+
+
+def _float32_to_int24_fast(samples: np.ndarray) -> np.ndarray:
+    """Convert float32 → 24-bit packed as 3-byte little-endian using numpy vectorisation.
+
+    Faster than per-sample struct packing. Returns uint8 array of shape (n*3,).
+    """
+    clamped = np.clip(samples, -1.0, 1.0)
+    # Scale to 24-bit signed range
+    scaled = np.round(clamped * 8388607.0).astype(np.int32)
+    # Clamp to 24-bit signed range
+    scaled = np.clip(scaled, -8388608, 8388607)
+    # Convert to unsigned 24-bit for packing
+    mask = np.where(scaled < 0, scaled + 16777216, scaled).astype(np.uint32)
+
+    n = len(mask)
+    packed = np.empty(n * 3, dtype=np.uint8)
+    packed[0::3] = (mask & 0xFF).astype(np.uint8)
+    packed[1::3] = ((mask >> 8) & 0xFF).astype(np.uint8)
+    packed[2::3] = ((mask >> 16) & 0xFF).astype(np.uint8)
+    return packed
+
+
+def _float32_to_int32(samples: np.ndarray) -> np.ndarray:
+    """Convert float32 [-1.0, 1.0] → int32."""
+    clamped = np.clip(samples, -1.0, 1.0)
+    # Use float64 to preserve precision at the int32 range boundary
+    scaled = (clamped.astype(np.float64) * 2147483647.0)
+    return np.clip(np.round(scaled), -2147483648, 2147483647).astype(np.int32)
+
+
+def _float32_to_float32(samples: np.ndarray) -> np.ndarray:
+    """Pass through float32 (already correct format)."""
+    return samples.astype(np.float32)
+
+
+# ── WAV writer ───────────────────────────────────────────────────────────────
+
+
+@dataclass
+class WAVWriterConfig:
+    """Configuration for a WAV writer."""
+    sample_rate: int = 48000
+    channels: int = 1
+    bit_depth: WAVBitDepth = WAVBitDepth.INT24
+    # Internal buffering: flush every N bytes (0 = only at close)
+    flush_interval_bytes: int = 65536  # 64KB
+
+
+class WAVWriter:
+    """Streaming WAV file writer.
+
+    Opens a file, writes the initial header with placeholder sizes,
+    accepts float32 audio frames in chunks, and finalises the header
+    on close.
+
+    Usage::
+
+        with WAVWriter("output.wav", sample_rate=48000, channels=2, bit_depth=24) as w:
+            w.write_frames(audio_chunk)  # shape (frames, channels), float32
+    """
+
+    def __init__(
+        self,
+        file_path: str | Path,
+        *,
+        sample_rate: int = 48000,
+        channels: int = 1,
+        bit_depth: WAVBitDepth | int = WAVBitDepth.INT24,
+        config: Optional[WAVWriterConfig] = None,
+    ):
+        self.file_path = Path(file_path)
+        self.sample_rate = sample_rate
+        self.channels = channels
+        self.bit_depth = WAVBitDepth(bit_depth) if isinstance(bit_depth, int) else bit_depth
+        self.config = config or WAVWriterConfig(
+            sample_rate=sample_rate, channels=channels, bit_depth=self.bit_depth
+        )
+
+        self._file: Optional[BinaryIO] = None
+        self._frames_written: int = 0
+        self._data_bytes_written: int = 0
+        self._bytes_since_flush: int = 0
+        self._closed: bool = False
+        self._header_written: bool = False
+
+        # Select conversion function based on bit depth
+        self._convert = _CONVERTERS.get(self.bit_depth)
+        if self._convert is None:
+            raise ValueError(f"Unsupported bit depth: {self.bit_depth}")
+
+        # Determine bytes per sample frame (all channels)
+        abs_depth = abs(int(self.bit_depth))
+        self._bytes_per_sample = abs_depth // 8
+        self._bytes_per_frame = self._bytes_per_sample * channels
+
+        # Open the file
+        self._open()
+
+    def __enter__(self) -> WAVWriter:
+        return self
+
+    def __exit__(self, *args) -> None:
+        self.close()
+
+    # ── File I/O ──────────────────────────────────────────────────────────
+
+    def _open(self) -> None:
+        """Create the file and write placeholder header."""
+        self.file_path.parent.mkdir(parents=True, exist_ok=True)
+        self._file = open(self.file_path, "wb")
+        self._write_placeholder_header()
+
+    def _write_placeholder_header(self) -> None:
+        """Write RIFF/WAVE header with placeholder (0) data size.
+
+        The data size and overall file size will be patched on close.
+        """
+        f = self._file
+        assert f is not None, "File must be open before writing header"
+
+        # Determine format
+        abs_depth = abs(int(self.bit_depth))
+        if self.bit_depth == WAVBitDepth.FLOAT32:
+            audio_format = _WAV_FORMAT_IEEE_FLOAT
+        else:
+            audio_format = _WAV_FORMAT_PCM
+
+        byte_rate = self.sample_rate * self.channels * self._bytes_per_sample
+        block_align = self.channels * self._bytes_per_sample
+
+        # RIFF header
+        f.write(b"RIFF")
+        f.write(struct.pack("<I", 0))  # file size placeholder
+        f.write(b"WAVE")
+
+        # fmt sub-chunk
+        f.write(b"fmt ")
+        f.write(struct.pack("<I", 16))  # chunk size (16 for PCM/IEEE)
+        f.write(struct.pack("<H", audio_format))
+        f.write(struct.pack("<H", self.channels))
+        f.write(struct.pack("<I", self.sample_rate))
+        f.write(struct.pack("<I", byte_rate))
+        f.write(struct.pack("<H", block_align))
+        f.write(struct.pack("<H", abs_depth))
+
+        # data sub-chunk header
+        f.write(b"data")
+        f.write(struct.pack("<I", 0))  # data size placeholder
+
+        f.flush()
+        self._header_written = True
+
+    def write_frames(self, frames: np.ndarray) -> int:
+        """Write audio frames to the file.
+
+        Args:
+            frames: float32 numpy array, shape (num_frames, channels) or (num_frames,).
+                    Values should be in [-1.0, 1.0].
+
+        Returns:
+            Number of frames written.
+
+        Raises:
+            ValueError: If writer is already closed.
+        """
+        if self._closed:
+            raise ValueError("Cannot write to closed WAVWriter")
+        if self._file is None:
+            raise RuntimeError("File not open")
+
+        # Ensure 2D
+        if frames.ndim == 1:
+            frames = frames.reshape(-1, 1)
+
+        num_frames = frames.shape[0]
+
+        if num_frames == 0:
+            return 0
+
+        # Handle channel mismatch
+        if frames.shape[1] != self.channels:
+            if frames.shape[1] == 1 and self.channels > 1:
+                # Mono → multi: duplicate
+                frames = np.tile(frames, (1, self.channels))
+            elif self.channels == 1 and frames.shape[1] > 1:
+                # Multi → mono: take first channel
+                frames = frames[:, 0:1]
+            else:
+                raise ValueError(
+                    f"Frame channels ({frames.shape[1]}) != writer channels ({self.channels})"
+                )
+
+        # Interleave channels: (num_frames, channels) → flat
+        interleaved = frames.ravel()  # row-major = interleaved
+
+        # Convert to target format
+        data = self._convert(interleaved)
+
+        # Write
+        if isinstance(data, np.ndarray):
+            self._file.write(data.tobytes())
+            byte_count = data.nbytes
+        else:
+            self._file.write(data)
+            byte_count = len(data)
+
+        self._frames_written += num_frames
+        self._data_bytes_written += byte_count
+        self._bytes_since_flush += byte_count
+
+        # Periodic flush
+        if self.config.flush_interval_bytes > 0 and self._bytes_since_flush >= self.config.flush_interval_bytes:
+            self._file.flush()
+            self._bytes_since_flush = 0
+
+        return num_frames
+
+    def flush(self) -> None:
+        """Force flush buffered data to disk."""
+        if self._file and not self._closed:
+            self._file.flush()
+            self._bytes_since_flush = 0
+
+    def close(self) -> None:
+        """Finalise header and close the file."""
+        if self._closed:
+            return
+        self._closed = True
+
+        if self._file is None:
+            return
+
+        try:
+            # Flush remaining data
+            self._file.flush()
+
+            # Patch header sizes
+            data_size = self._data_bytes_written
+            file_size = 36 + data_size  # 44 - 8 = 36 (RIFF header after "WAVE")
+
+            self._file.seek(4)
+            self._file.write(struct.pack("<I", file_size))
+
+            self._file.seek(40)  # data chunk size field
+            self._file.write(struct.pack("<I", data_size))
+
+            self._file.flush()
+        except Exception as exc:
+            logger.error("Error finalising WAV header for %s: %s", self.file_path, exc)
+        finally:
+            self._file.close()
+            self._file = None
+
+        logger.debug(
+            "WAV closed: %s — %d frames, %d bytes, %dHz/%dch/%s",
+            self.file_path.name,
+            self._frames_written,
+            self._data_bytes_written,
+            self.sample_rate,
+            self.channels,
+            self.bit_depth.name,
+        )
+
+    @property
+    def frames_written(self) -> int:
+        return self._frames_written
+
+    @property
+    def duration_seconds(self) -> float:
+        return self._frames_written / self.sample_rate if self.sample_rate > 0 else 0.0
+
+    @property
+    def closed(self) -> bool:
+        return self._closed
+
+
+# ── Converter registry ───────────────────────────────────────────────────────
+
+_SampleConverter = Callable[[np.ndarray], np.ndarray]
+
+_CONVERTERS: dict[WAVBitDepth, _SampleConverter] = {
+    WAVBitDepth.INT16:   _float32_to_int16,
+    WAVBitDepth.INT24:   _float32_to_int24_fast,
+    WAVBitDepth.INT32:   _float32_to_int32,
+    WAVBitDepth.FLOAT32: _float32_to_float32,
+}
+
+
+# ── Utility: quick WAV write ─────────────────────────────────────────────────
+
+def write_wav(
+    file_path: str | Path,
+    samples: np.ndarray,
+    sample_rate: int = 48000,
+    bit_depth: WAVBitDepth | int = WAVBitDepth.INT24,
+) -> Path:
+    """Write a complete WAV file in one shot (convenience wrapper).
+
+    Args:
+        file_path: Output path.
+        samples: float32 audio data, shape (frames,) or (frames, channels).
+        sample_rate: Sample rate in Hz.
+        bit_depth: Target bit depth.
+
+    Returns:
+        Path to written file.
+    """
+    if samples.ndim == 1:
+        channels = 1
+    else:
+        channels = samples.shape[1]
+
+    with WAVWriter(file_path, sample_rate=sample_rate, channels=channels, bit_depth=bit_depth) as w:
+        w.write_frames(samples)
+
+    return Path(file_path)