//! Windows audio backend — cpal / WASAPI (Phase 1).
//!
//! Implements [`AudioBackend`] on top of [`cpal`], which wraps WASAPI on Windows.
//! It is the Windows counterpart to `pipewire_impl.rs` and deliberately preserves
//! the exact same contract so the rest of the app (mixer, encoder, jitter buffer)
//! is unchanged:
//!
//! - **Capture**: mono, 48 kHz, S16 PCM, emitted as `Vec<i16>` frames of
//!   [`CAPTURE_FRAME`] (960 = 20 ms) samples — matching the encoder/jitter frame.
//!   The RT capture callback only downmixes and pushes samples into a lock-free
//!   ring; the owning thread drains that ring, frames it, and sends — so the
//!   callback never allocates, locks, or touches an mpsc channel.
//! - **Playback**: stereo interleaved ([`PLAYBACK_CHANNELS`]) S16 PCM at 48 kHz,
//!   drained from a ring buffer that is paced to the device's hardware clock via
//!   `ring_fill` exactly as the PipeWire backend does.
//!
//! ## Threading and the `!Send` stream
//!
//! `cpal::Stream` is `!Send` (some backends require it to be created and dropped
//! on the same thread), but [`AudioBackend`] is `Send + Sync` and the backend is
//! shared through an `Arc`. So the stream never lives in the struct: each of
//! `start_capture`/`start_playback` spawns one owning thread that builds the
//! stream, plays it, and keeps it alive until the per-worker `running` flag flips
//! (set by `stop`). The struct holds only `Send` handles (the flag + the join
//! handle). The stream's RT callback does the actual audio work; the owning
//! thread additionally feeds the playback ring (or drains the capture ring).
//!
//! `start_*` does not return until the owning thread reports back over a readiness
//! channel that the device resolved and the stream is built and playing — so a
//! device/format/WASAPI failure surfaces as a real `Err` to the caller instead of
//! leaving the UI in a joined-but-silent room.
//!
//! ## Sample rate and channel layout (W4)
//!
//! The whole pipeline runs internally at 48 kHz (Opus + the 960-sample frame) and
//! mono capture / stereo playback. We prefer a native-48 kHz device config so the
//! common case is conversion-free and bit-exact. When the device can't do 48 kHz
//! (commonly a 44.1 kHz-only endpoint) or can't do stereo output, we fall back to
//! the device's default config and convert at the boundary with the dep-free
//! [`super::resample`] linear resamplers instead of hard-erroring:
//!
//! - **Capture**: the device-rate mono stream is resampled to 48 kHz on the
//!   capture drain thread (off the RT callback) before framing.
//! - **Playback**: the internal 48 kHz stereo bus is resampled to the device rate
//!   and remapped to the device channel count inside the output RT callback, which
//!   pulls internal frames from the ring on demand (allocation-free, so RT-safe).
//!   The ring, prefill, and `ring_fill` pacing stay in internal 48 kHz-stereo
//!   units, so the mixer is unchanged.
//!
//! Linear interpolation has no anti-aliasing filter (see [`super::resample`] docs);
//! it is adequate for speech and keeps the matching-rate path bit-exact, with the
//! seam ready for a higher-quality resampler later.

use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, AtomicUsize, Ordering};
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use cpal::{
    Device, FromSample, Sample, SampleFormat, SampleRate, SizedSample, Stream, StreamConfig,
};
use ringbuf::{
    HeapRb,
    traits::{Consumer, Producer, Split},
};

use super::resample::{PushResampler, StereoPullResampler};
use super::{AudioBackend, AudioDevice, AudioError, PLAYBACK_CHANNELS, PLAYBACK_TARGET_SAMPLES};

/// The one sample rate the pipeline supports (Opus + the 20 ms frame).
const SAMPLE_RATE: u32 = 48_000;
/// Mono capture frame: 960 samples = 20 ms @ 48 kHz. Matches the PipeWire backend
/// and `core::jitter::FRAME_SAMPLES`.
const CAPTURE_FRAME: usize = 960;
/// Lock-free capture ring capacity (mono samples) between the RT callback and the
/// owning drain thread: 8 frames = 160 ms of headroom, so a scheduling hiccup on
/// the drain thread doesn't immediately overrun the RT producer.
const CAPTURE_RING_CAPACITY: usize = CAPTURE_FRAME * 8;
/// How long the capture drain thread sleeps when the ring is momentarily empty,
/// before polling again. Small enough to stay well under the 20 ms frame cadence.
const CAPTURE_POLL: Duration = Duration::from_millis(5);
/// Playback ring capacity in interleaved samples: 200 ms of stereo @ 48 kHz.
/// Comfortably above [`PLAYBACK_TARGET_SAMPLES`] so the clock-paced producer has
/// headroom and never has to drop frames in steady state.
const RING_CAPACITY: usize = 9600 * PLAYBACK_CHANNELS;
/// How often a blocked playback worker re-checks its `running` flag, bounding how
/// long `stop()` can take to join it (mirrors the PipeWire backend's `WORKER_POLL`).
const WORKER_POLL: Duration = Duration::from_millis(100);
/// How long a freshly-played stream has to prove itself (deliver its first RT
/// callbacks) before the start is treated as failed. cpal's `play()` only *queues*
/// the WASAPI `Start()`, so a queued-but-failed start would otherwise masquerade as
/// success and join the UI into a silent room (review W1).
const STREAM_START_TIMEOUT: Duration = Duration::from_secs(3);
/// Number of completed RT callbacks the owner waits for before declaring the stream
/// live. One callback isn't proof: a stream can fire once and immediately fail in
/// the same processing cycle, so requiring a couple of cycles (plus the terminal
/// error check) keeps a one-shot-then-dead stream from being reported Ok (B1).
const MIN_START_CALLBACKS: usize = 2;
/// Backstop for [`finish_start`]: bounds the WHOLE owner path (resolve + build +
/// play + the [`STREAM_START_TIMEOUT`] callback wait + any wedged-stream cleanup).
/// Sized as a generous setup budget plus the callback wait plus slack so a slow but
/// valid device (e.g. a Bluetooth endpoint that takes seconds to spin up) is not
/// falsely failed, while a driver that wedges before the owner can report is still
/// released eventually (review W6, B4).
const FINISH_START_TIMEOUT: Duration = Duration::from_secs(10);
/// Lowest / highest device sample rate the backend will drive. The floor bounds
/// the playback pull-resampler's input-pulls-per-output-frame (≈48000/rate) so a
/// pathological low rate can't blow the RT callback's deadline; the ceiling and a
/// nonzero floor also reject the 0 Hz / absurd values a misbehaving driver could
/// report, which would otherwise panic or spin (review W7).
const MIN_DEVICE_RATE: u32 = 8_000;
const MAX_DEVICE_RATE: u32 = 384_000;

// Stream-error categories carried from the RT error callback to the owner thread
// through an `AtomicU8`, so the callback itself never allocates or logs — both of
// which it previously did via `format!`/`log_msg` on the time-critical stream
// thread (review W2). The owner/logger translates the code off the RT path.
const STREAM_ERR_NONE: u8 = 0;
const STREAM_ERR_DEVICE_UNAVAILABLE: u8 = 1;
const STREAM_ERR_BACKEND: u8 = 2;

/// Map a cpal stream error to its [`STREAM_ERR_*`](STREAM_ERR_NONE) code. Pure +
/// allocation-free, so it is safe to call from the RT error callback.
fn stream_err_code(e: &cpal::StreamError) -> u8 {
    match e {
        cpal::StreamError::DeviceNotAvailable => STREAM_ERR_DEVICE_UNAVAILABLE,
        _ => STREAM_ERR_BACKEND,
    }
}

/// Human-readable text for a [`STREAM_ERR_*`](STREAM_ERR_NONE) code, logged off the
/// RT path by the owner/health-logger thread.
fn stream_err_text(code: u8) -> &'static str {
    match code {
        STREAM_ERR_DEVICE_UNAVAILABLE => "audio device became unavailable",
        _ => "audio backend stream error",
    }
}

/// Wait for a just-played stream to prove it actually started: its RT data
/// callback bumps `callbacks`, or an error callback sets `err_code`. Returns `Ok`
/// once [`MIN_START_CALLBACKS`] cycles have run with no error, `Err` on an
/// error-callback code or [`STREAM_START_TIMEOUT`], or a clean abort if `stop()`
/// cleared `running` mid-start. Polls a few cheap atomics on the owner thread —
/// never the RT thread (review W1).
///
/// The error is **terminal and wins any race** with `callbacks`: a stream can run a
/// callback and then fail in the same processing cycle, so `err_code` is checked
/// first each loop AND re-checked before declaring success (review B1).
fn wait_for_stream_start(
    callbacks: &AtomicUsize,
    err_code: &AtomicU8,
    running: &AtomicBool,
) -> Result<(), AudioError> {
    let deadline = Instant::now() + STREAM_START_TIMEOUT;
    let as_err = |code: u8| Err(AudioError::Stream(stream_err_text(code).to_string()));
    loop {
        let code = err_code.load(Ordering::Relaxed);
        if code != STREAM_ERR_NONE {
            return as_err(code);
        }
        if callbacks.load(Ordering::Relaxed) >= MIN_START_CALLBACKS {
            // Re-check: a callback that pushed us to the threshold may have been the
            // last before a same-cycle failure. Let a terminal error win.
            let code = err_code.load(Ordering::Relaxed);
            if code != STREAM_ERR_NONE {
                return as_err(code);
            }
            return Ok(());
        }
        if !running.load(Ordering::Relaxed) {
            return Err(AudioError::Stream("stream start aborted".to_string()));
        }
        if Instant::now() >= deadline {
            return Err(AudioError::Stream(
                "stream did not start within timeout (no WASAPI callback)".to_string(),
            ));
        }
        thread::sleep(Duration::from_millis(5));
    }
}

/// Windows audio backend. See module docs.
pub struct CpalBackend {
    capture: Mutex<SlotState>,
    playback: Mutex<SlotState>,
}

/// A spawned owning thread plus the flags that coordinate its lifetime: `running`
/// tells it to drop its stream and exit; `exited` is flipped true (by [`ExitGuard`]
/// in the thread body) when it actually returns, so a *detached* wedged start can be
/// detected as finished later (review B3).
struct StreamWorker {
    running: Arc<AtomicBool>,
    exited: Arc<AtomicBool>,
    thread: JoinHandle<()>,
}

/// Flips its flag true when dropped, marking a worker thread as exited. Lives at the
/// top of the worker closure so it fires on normal return, panic unwind, or whenever
/// a wedged driver call finally releases the thread — which is what lets a [`SlotState::Wedged`]
/// tombstone (B3) know its orphan is gone.
struct ExitGuard(Arc<AtomicBool>);
impl Drop for ExitGuard {
    fn drop(&mut self) {
        self.0.store(true, Ordering::Relaxed);
    }
}

/// The lifecycle state of a capture or playback slot.
enum SlotState {
    /// No stream — a new start may proceed.
    Idle,
    /// A live, started stream owned by its worker thread.
    Live(StreamWorker),
    /// A start that timed out wedged in a driver call (review B3). Its worker thread
    /// was *detached* rather than joined — joining would re-introduce the unbounded
    /// hang [`FINISH_START_TIMEOUT`] exists to prevent — so it may still be alive,
    /// holding the COM/device handle. `exited` flips true when that orphan finally
    /// returns. New starts are rejected until then, so retries against a permanently
    /// wedged device don't pile up more orphan threads.
    Wedged { exited: Arc<AtomicBool> },
}

/// Inspect a slot before starting a stream into it. Clears a [`SlotState::Wedged`]
/// tombstone whose orphan has since exited (the slot becomes reusable), but rejects
/// a start while a wedged orphan is still alive or a live stream already owns the
/// slot. Pure w.r.t. the passed state, so the tombstone logic is unit-testable (B3).
fn ensure_idle(state: &mut SlotState, what: &str) -> Result<(), AudioError> {
    match state {
        SlotState::Idle => Ok(()),
        SlotState::Live(_) => Err(AudioError::Stream(format!("{what} already started"))),
        SlotState::Wedged { exited } => {
            if exited.load(Ordering::Relaxed) {
                *state = SlotState::Idle;
                Ok(())
            } else {
                Err(AudioError::Stream(format!(
                    "{what} is recovering from an unresponsive audio device; retry shortly"
                )))
            }
        }
    }
}

impl CpalBackend {
    pub fn new() -> Self {
        Self {
            capture: Mutex::new(SlotState::Idle),
            playback: Mutex::new(SlotState::Idle),
        }
    }
}

impl Default for CpalBackend {
    fn default() -> Self {
        Self::new()
    }
}

impl AudioBackend for CpalBackend {
    fn start_capture(
        &self,
        tx: Sender<Vec<i16>>,
        target_node: Option<String>,
    ) -> Result<(), AudioError> {
        let mut guard = self.capture.lock().unwrap();
        ensure_idle(&mut guard, "capture")?;
        let running = Arc::new(AtomicBool::new(true));
        let exited = Arc::new(AtomicBool::new(false));
        let running_thread = running.clone();
        let exited_thread = exited.clone();
        let (ready_tx, ready_rx) = channel::<Result<(), AudioError>>();
        let thread = thread::Builder::new()
            .name("peerspeak-cpal-capture".to_string())
            .spawn(move || {
                let _exit = ExitGuard(exited_thread);
                run_capture(tx, target_node, running_thread, ready_tx);
            })
            .map_err(|e| AudioError::Init(e.to_string()))?;
        finish_start(
            guard,
            StreamWorker {
                running,
                exited,
                thread,
            },
            ready_rx,
            "capture",
        )
    }

    fn start_playback(
        &self,
        rx: Receiver<Vec<i16>>,
        target_node: Option<String>,
        ring_fill: Arc<AtomicUsize>,
    ) -> Result<(), AudioError> {
        let mut guard = self.playback.lock().unwrap();
        ensure_idle(&mut guard, "playback")?;
        let running = Arc::new(AtomicBool::new(true));
        let exited = Arc::new(AtomicBool::new(false));
        let running_thread = running.clone();
        let exited_thread = exited.clone();
        let (ready_tx, ready_rx) = channel::<Result<(), AudioError>>();
        let thread = thread::Builder::new()
            .name("peerspeak-cpal-playback".to_string())
            .spawn(move || {
                let _exit = ExitGuard(exited_thread);
                run_playback(rx, target_node, ring_fill, running_thread, ready_tx);
            })
            .map_err(|e| AudioError::Init(e.to_string()))?;
        finish_start(
            guard,
            StreamWorker {
                running,
                exited,
                thread,
            },
            ready_rx,
            "playback",
        )
    }

    fn stop(&self) -> Result<(), AudioError> {
        for slot in [&self.capture, &self.playback] {
            let mut guard = slot.lock().unwrap();
            match std::mem::replace(&mut *guard, SlotState::Idle) {
                SlotState::Live(worker) => {
                    worker.running.store(false, Ordering::Relaxed);
                    let _ = worker.thread.join();
                }
                // A wedged orphan was detached and can't be joined. If it has since
                // exited the slot is now clear; otherwise restore the tombstone so a
                // later start still sees the device is recovering (B3).
                SlotState::Wedged { exited } => {
                    if !exited.load(Ordering::Relaxed) {
                        *guard = SlotState::Wedged { exited };
                    }
                }
                SlotState::Idle => {}
            }
        }
        Ok(())
    }
}

/// Block until the just-spawned worker reports (over `ready_rx`) that its stream
/// is built and playing, then either install it (`Ok`) or join it and surface the
/// real error. This is what makes `start_capture`/`start_playback` fail loudly
/// instead of returning `Ok` into a joined-but-silent room (Codex review W1).
fn finish_start(
    mut guard: std::sync::MutexGuard<'_, SlotState>,
    worker: StreamWorker,
    ready_rx: Receiver<Result<(), AudioError>>,
    what: &str,
) -> Result<(), AudioError> {
    // Bounded wait. An unbounded `recv()` here would hang `start_*` forever — and
    // any concurrent `stop()` behind the same slot mutex — if a WASAPI/driver call
    // wedged the worker before it could report (review W6).
    match ready_rx.recv_timeout(FINISH_START_TIMEOUT) {
        Ok(Ok(())) => {
            *guard = SlotState::Live(worker);
            Ok(())
        }
        // Setup failed (Err) or the worker disconnected before reporting: either
        // way it has stopped, so reap it and surface the error.
        Ok(Err(e)) => {
            worker.running.store(false, Ordering::Relaxed);
            let _ = worker.thread.join();
            Err(e)
        }
        Err(RecvTimeoutError::Disconnected) => {
            worker.running.store(false, Ordering::Relaxed);
            let _ = worker.thread.join();
            Err(AudioError::Init(format!(
                "cpal {what} worker exited before reporting readiness"
            )))
        }
        Err(RecvTimeoutError::Timeout) => {
            // The worker is wedged in a driver call. Signal it to exit, but DETACH
            // rather than join — joining would re-introduce the unbounded hang this
            // timeout exists to prevent. Leave a Wedged tombstone so subsequent
            // starts are rejected until the orphan's ExitGuard flips `exited`, rather
            // than spawning more orphan threads against the same dead device (B3).
            let StreamWorker {
                running,
                exited,
                thread,
            } = worker;
            running.store(false, Ordering::Relaxed);
            drop(thread);
            *guard = SlotState::Wedged { exited };
            Err(AudioError::Init(format!(
                "cpal {what} did not start within {FINISH_START_TIMEOUT:?}"
            )))
        }
    }
}

// ---------------------------------------------------------------------------
// Device enumeration (for the settings device pickers)
// ---------------------------------------------------------------------------

/// Enumerate WASAPI input/output devices via cpal, sorted by description to match
/// the PipeWire backend's stable UI ordering.
///
/// cpal exposes a single friendly name per device, which is also what [`resolve`]
/// matches `target_node` against — so `name` and `description` are the same string
/// and a saved selection round-trips. Note: WASAPI device names are less stable
/// across driver/endpoint changes than PipeWire node names, so a saved device may
/// not always be found again; selection then falls back to the system default.
pub fn enumerate_audio_devices() -> Vec<AudioDevice> {
    let host = cpal::default_host();
    let mut devices = Vec::new();

    if let Ok(inputs) = host.input_devices() {
        for device in inputs {
            if let Ok(name) = device.name() {
                devices.push(AudioDevice {
                    description: name.clone(),
                    name,
                    is_input: true,
                });
            }
        }
    }
    if let Ok(outputs) = host.output_devices() {
        for device in outputs {
            if let Ok(name) = device.name() {
                devices.push(AudioDevice {
                    description: name.clone(),
                    name,
                    is_input: false,
                });
            }
        }
    }

    devices.sort_by(|a, b| a.description.cmp(&b.description));
    devices
}

// ---------------------------------------------------------------------------
// Device / config selection
// ---------------------------------------------------------------------------

/// Resolve a device (by `target` name, else the system default) and a stream
/// config. We prefer a config running natively at [`SAMPLE_RATE`] (conversion-free);
/// if the device has none, we fall back to its default config and resample/remap at
/// the boundary (W4 — see module docs and [`choose_config`]).
fn resolve(
    output: bool,
    target: Option<String>,
) -> Result<(Device, StreamConfig, SampleFormat), AudioError> {
    let host = cpal::default_host();

    let default = || {
        if output {
            host.default_output_device()
        } else {
            host.default_input_device()
        }
    };
    let device = match target {
        // A saved device name that no longer resolves falls back to the system
        // default — but log it, because WASAPI friendly names can change across
        // driver/endpoint changes, so a silent fallback otherwise looks like
        // "audio went to the wrong device for no reason" (review W7).
        Some(ref name) => match find_device_by_name(&host, output, name) {
            Some(dev) => Some(dev),
            None => {
                crate::log_msg(&format!(
                    "cpal: saved {} device '{name}' not found; using system default",
                    if output { "output" } else { "input" },
                ));
                default()
            }
        },
        None => default(),
    }
    .ok_or_else(|| AudioError::Device("no audio device available".to_string()))?;

    let supported = choose_config(&device, output)?;
    let sample_format = supported.sample_format();
    let config = supported.config();

    // Validate the OS-reported geometry before any code divides by it or sizes a
    // loop from it (review W7). Zero channels would panic `chunks_exact(0)` /
    // `chunks_mut(0)`; a zero or absurd rate would yield an infinite/huge resample
    // ratio. Reject up front with a real error instead of panicking or spinning.
    if config.channels == 0 {
        return Err(AudioError::Device(
            "audio device reports zero channels".to_string(),
        ));
    }
    if !(MIN_DEVICE_RATE..=MAX_DEVICE_RATE).contains(&config.sample_rate.0) {
        return Err(AudioError::Device(format!(
            "audio device sample rate {} Hz is outside the supported {MIN_DEVICE_RATE}–{MAX_DEVICE_RATE} Hz range",
            config.sample_rate.0,
        )));
    }
    Ok((device, config, sample_format))
}

fn find_device_by_name(host: &cpal::Host, output: bool, name: &str) -> Option<Device> {
    let devices = if output {
        host.output_devices().ok()?
    } else {
        host.input_devices().ok()?
    };
    devices
        .into_iter()
        .find(|d| d.name().is_ok_and(|n| n == name))
}

/// Whether the backend can actually open this config range. The workers only build
/// `F32`/`I16`/`U16` streams ([`build_input`]/[`build_output`] — every other sample
/// format hits the `other => Err(...)` arm), and a zero-channel range would later be
/// rejected by [`resolve`]'s geometry check. Filtering both here keeps
/// [`choose_config`] from *ranking* a range it can't drive ahead of a usable one and
/// then hard-failing the start instead of trying the next candidate (Codex B5
/// re-review, P3).
fn usable_range(r: &cpal::SupportedStreamConfigRange) -> bool {
    r.channels() > 0 && format_supported(r.sample_format())
}

/// Sample formats the capture/playback stream builders accept. Pure, so it's
/// unit-testable independently of the cpal range types.
fn format_supported(fmt: SampleFormat) -> bool {
    matches!(
        fmt,
        SampleFormat::F32 | SampleFormat::I16 | SampleFormat::U16
    )
}

/// Pick a sample rate inside both a device's supported `[r_min, r_max]` span and the
/// backend's drivable `[MIN_DEVICE_RATE, MAX_DEVICE_RATE]` window, preferring
/// [`SAMPLE_RATE`] when it's reachable and otherwise the nearest in-window bound.
/// Returns `None` when the device span doesn't overlap the window at all. Pure and
/// integer-only, so the selection policy is unit-testable (review B5).
fn bounded_rate(r_min: u32, r_max: u32) -> Option<u32> {
    let lo = r_min.max(MIN_DEVICE_RATE);
    let hi = r_max.min(MAX_DEVICE_RATE);
    (lo <= hi).then(|| SAMPLE_RATE.clamp(lo, hi))
}

/// Pick a stream config. Preference order, best (no conversion) first:
///  1. exactly [`SAMPLE_RATE`] at the preferred layout (stereo out / mono in),
///  2. exactly [`SAMPLE_RATE`] at any channel count (rate-exact, backend remaps),
///  3. a supported config at a [`bounded_rate`] near 48 kHz (backend resamples + remaps),
///  4. the device's default config (only if nothing above is drivable).
///
/// Cases 3–4 incur resampling; the backend reads the returned config's rate and
/// channel count and converts at the boundary (W4). Case 3 (review B5) is what keeps
/// an oddball endpoint whose default rate is outside the drivable window — but which
/// also exposes a usable in-window config — from being rejected by [`resolve`]. A
/// device that exposes no config at all is still a hard error.
fn choose_config(device: &Device, output: bool) -> Result<cpal::SupportedStreamConfig, AudioError> {
    let ranges: Vec<cpal::SupportedStreamConfigRange> = if output {
        device
            .supported_output_configs()
            .map_err(|e| AudioError::Device(e.to_string()))?
            .collect()
    } else {
        device
            .supported_input_configs()
            .map_err(|e| AudioError::Device(e.to_string()))?
            .collect()
    };

    // A range covers a sample-rate span and a fixed channel count.
    let supports_48k = |r: &cpal::SupportedStreamConfigRange| {
        r.min_sample_rate().0 <= SAMPLE_RATE && SAMPLE_RATE <= r.max_sample_rate().0
    };
    let pick = |channels: Option<u16>| {
        ranges
            .iter()
            .find(|r| usable_range(r) && supports_48k(r) && channels.is_none_or(|c| r.channels() == c))
            .cloned()
    };

    // Cases 1 + 2: an exact-48 kHz config, preferring the native layout but
    // accepting any channel count (the backend remaps without resampling).
    let exact = if output {
        pick(Some(PLAYBACK_CHANNELS as u16)).or_else(|| pick(None))
    } else {
        pick(Some(1)).or_else(|| pick(None))
    };
    if let Some(r) = exact {
        return Ok(r.with_sample_rate(SampleRate(SAMPLE_RATE)));
    }

    // Case 3: no native 48 kHz. Before falling back to the device default — which
    // resolve() rejects outright if its rate is outside the drivable window — look
    // for a supported config whose rate range overlaps that window and drive it at a
    // bounded rate, resampling at the boundary (review B5). Prefer the native layout,
    // then the bounded rate closest to 48 kHz.
    let pick_bounded = |channels: Option<u16>| -> Option<(cpal::SupportedStreamConfigRange, u32)> {
        ranges
            .iter()
            .filter(|r| usable_range(r) && channels.is_none_or(|c| r.channels() == c))
            .filter_map(|r| {
                bounded_rate(r.min_sample_rate().0, r.max_sample_rate().0)
                    .map(|rate| (r.clone(), rate))
            })
            .min_by_key(|(_, rate)| rate.abs_diff(SAMPLE_RATE))
    };
    let preferred_channels = if output { PLAYBACK_CHANNELS as u16 } else { 1 };
    if let Some((r, rate)) = pick_bounded(Some(preferred_channels)).or_else(|| pick_bounded(None)) {
        crate::log_msg(&format!(
            "cpal: device '{}' has no native {SAMPLE_RATE} Hz {} config; using bounded {rate} Hz / {} ch with linear resampling (W4/B5)",
            device.name().unwrap_or_else(|_| "<unknown>".to_string()),
            if output { "output" } else { "input" },
            r.channels(),
        ));
        return Ok(r.with_sample_rate(SampleRate(rate)));
    }

    // Case 4: last resort — the device's default config. If its rate is outside the
    // drivable window, resolve() rejects it with a clear device error, which is the
    // honest outcome: the device exposes nothing this backend can drive.
    let def = if output {
        device.default_output_config()
    } else {
        device.default_input_config()
    }
    .map_err(|e| AudioError::Device(e.to_string()))?;
    crate::log_msg(&format!(
        "cpal: device '{}' has no bounded {} config near {SAMPLE_RATE} Hz; falling back to default {} Hz / {} ch (W4)",
        device.name().unwrap_or_else(|_| "<unknown>".to_string()),
        if output { "output" } else { "input" },
        def.sample_rate().0,
        def.channels(),
    ));
    Ok(def)
}

// ---------------------------------------------------------------------------
// Capture
// ---------------------------------------------------------------------------

fn run_capture(
    tx: Sender<Vec<i16>>,
    target: Option<String>,
    running: Arc<AtomicBool>,
    ready: Sender<Result<(), AudioError>>,
) {
    // The RT callback pushes mono samples into this lock-free ring; we drain it on
    // this (non-RT) thread, so the callback never allocates or sends on a channel.
    let rb = HeapRb::<i16>::new(CAPTURE_RING_CAPACITY);
    let (producer, mut consumer) = rb.split();
    let overrun = Arc::new(AtomicU64::new(0));
    // Stream-liveness signals read by `wait_for_stream_start`: each RT data callback
    // bumps `callbacks`; the RT error callback sets `err_code` (it does NOT log —
    // that would allocate/syscall on the time-critical thread). See W1/W2/B1.
    let callbacks = Arc::new(AtomicUsize::new(0));
    let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE));

    // Fallible device/stream setup: resolve, build, and *queue* the stream start.
    let setup = || -> Result<(Stream, String, SampleFormat, usize, u32), AudioError> {
        let (device, config, sample_format) = resolve(false, target)?;
        let channels = config.channels as usize;
        let device_rate = config.sample_rate.0;
        let stream = match sample_format {
            SampleFormat::F32 => build_input::<f32, _>(
                &device, &config, producer, channels, overrun.clone(), callbacks.clone(),
                err_code.clone(),
            ),
            SampleFormat::I16 => build_input::<i16, _>(
                &device, &config, producer, channels, overrun.clone(), callbacks.clone(),
                err_code.clone(),
            ),
            SampleFormat::U16 => build_input::<u16, _>(
                &device, &config, producer, channels, overrun.clone(), callbacks.clone(),
                err_code.clone(),
            ),
            other => Err(AudioError::Stream(format!(
                "unsupported capture sample format: {other:?}"
            ))),
        }?;
        stream
            .play()
            .map_err(|e| AudioError::Stream(e.to_string()))?;
        let name = device.name().unwrap_or_else(|_| "<unknown>".to_string());
        Ok((stream, name, sample_format, channels, device_rate))
    };

    // Build + queue, then wait for the stream to actually prove it's live before
    // reporting readiness. `play()` returning Ok only means WASAPI's `Start()` was
    // queued; a later Start failure would otherwise leave us joined-but-silent (W1).
    let (stream, dev_name, sample_format, channels, device_rate) = match setup() {
        Ok(v) => match wait_for_stream_start(&callbacks, &err_code, &running) {
            Ok(()) => {
                let _ = ready.send(Ok(()));
                v
            }
            Err(e) => {
                // Drop the (possibly wedged) stream BEFORE reporting: cpal's
                // Stream::drop joins its WASAPI worker, so if that wedges we want
                // the Err withheld and finish_start's backstop to detach, rather
                // than finish_start joining this owner forever (B2).
                drop(v.0);
                let _ = ready.send(Err(e));
                return;
            }
        },
        Err(e) => {
            let _ = ready.send(Err(e));
            return;
        }
    };
    crate::log_msg(&format!(
        "cpal capture started: device='{dev_name}' format={sample_format:?} channels={channels} device_rate={device_rate} Hz -> {SAMPLE_RATE} Hz"
    ));

    // If the device isn't at 48 kHz, resample its mono stream up/down to 48 kHz on
    // this (non-RT) thread before framing (W4). At 48 kHz this stays None and the
    // samples pass straight through, bit-exact.
    let mut resampler =
        (device_rate != SAMPLE_RATE).then(|| PushResampler::new(device_rate, SAMPLE_RATE));
    // Reused scratch for a sample's resampled output (off-RT alloc; tiny — at most
    // a couple of samples per input). Avoids a nested-closure borrow over `acc`/`tx`.
    let mut resampled: Vec<i16> = Vec::new();

    // Drain the RT ring on this thread: pop mono samples, (resample,) frame them
    // (the `Vec` allocation lives here, off the RT path), and send completed
    // frames. Keep `stream` alive until `stop()` flips the flag.
    let mut acc = FrameAccumulator::new(CAPTURE_FRAME);
    let mut last_overrun = 0u64;
    let mut last_err = STREAM_ERR_NONE;
    while running.load(Ordering::Relaxed) {
        let mut drained = false;
        while let Some(sample) = consumer.try_pop() {
            drained = true;
            resampled.clear();
            match resampler {
                Some(ref mut rs) => {
                    rs.push(i16_to_f32(sample), |out| resampled.push(f32_to_i16(out)));
                }
                None => resampled.push(sample),
            }
            for s in resampled.drain(..) {
                if let Some(frame) = acc.push(s) {
                    // Consumer gone (call ended) → stop feeding; the stream is
                    // dropped below on the way out.
                    if tx.send(frame).is_err() {
                        drop(stream);
                        return;
                    }
                }
            }
        }
        let o = overrun.load(Ordering::Relaxed);
        if o != last_overrun {
            crate::log_msg(&format!(
                "cpal capture overrun: dropped {} samples (drain thread fell behind)",
                o - last_overrun
            ));
            last_overrun = o;
        }
        // Surface a stream error the RT callback flagged (it can't log itself).
        let ec = err_code.load(Ordering::Relaxed);
        if ec != STREAM_ERR_NONE && ec != last_err {
            crate::log_msg(&format!("cpal capture stream error: {}", stream_err_text(ec)));
            last_err = ec;
        }
        if !drained {
            thread::sleep(CAPTURE_POLL);
        }
    }
    drop(stream);
}

#[allow(clippy::too_many_arguments)]
fn build_input<T, P>(
    device: &Device,
    config: &StreamConfig,
    mut producer: P,
    channels: usize,
    overrun: Arc<AtomicU64>,
    callbacks: Arc<AtomicUsize>,
    err_code: Arc<AtomicU8>,
) -> Result<Stream, AudioError>
where
    T: SizedSample + Send + 'static,
    i16: FromSample<T>,
    P: Producer<Item = i16> + Send + 'static,
{
    // RT-safe error callback: record a category in an atomic only. Formatting +
    // logging happen on the owner thread (the cpal/WASAPI error callback runs on
    // the time-critical stream thread, where alloc/syscall are forbidden — W2).
    let err_fn = move |e: cpal::StreamError| err_code.store(stream_err_code(&e), Ordering::Relaxed);
    device
        .build_input_stream::<T, _, _>(
            config,
            move |data: &[T], _| {
                // Count callbacks so the owner can confirm the stream is really
                // running before reporting Ok (W1/B1).
                callbacks.fetch_add(1, Ordering::Relaxed);
                // RT-safe: downmix + wait-free push only. A full ring means the
                // drain thread stalled; count the drop and keep going.
                for frame in data.chunks_exact(channels) {
                    let mono = downmix_to_mono(frame);
                    if producer.try_push(mono).is_err() {
                        overrun.fetch_add(1, Ordering::Relaxed);
                    }
                }
            },
            err_fn,
            None,
        )
        .map_err(|e| AudioError::Stream(e.to_string()))
}

/// Average a device frame's channels down to a single mono i16. For a 1-channel
/// device this is just the converted sample.
fn downmix_to_mono<T>(frame: &[T]) -> i16
where
    T: Copy,
    i16: FromSample<T>,
{
    if frame.is_empty() {
        return 0;
    }
    let sum: i32 = frame.iter().map(|&s| i16::from_sample(s) as i32).sum();
    (sum / frame.len() as i32) as i16
}

/// Scale an i16 PCM sample to f32 in roughly `[-1, 1]` for interpolation.
#[inline]
fn i16_to_f32(s: i16) -> f32 {
    s as f32 / 32768.0
}

/// Convert an interpolated f32 sample back to i16, clamping to range.
#[inline]
fn f32_to_i16(x: f32) -> i16 {
    (x * 32768.0).clamp(i16::MIN as f32, i16::MAX as f32) as i16
}

/// Accumulates mono samples into fixed-size [`CAPTURE_FRAME`] frames. Pulled out
/// of the RT callback so the framing is unit-testable.
struct FrameAccumulator {
    buf: Vec<i16>,
    frame_len: usize,
}

impl FrameAccumulator {
    fn new(frame_len: usize) -> Self {
        Self {
            buf: Vec::with_capacity(frame_len),
            frame_len,
        }
    }

    /// Push one sample; returns a completed frame when the buffer fills.
    fn push(&mut self, sample: i16) -> Option<Vec<i16>> {
        self.buf.push(sample);
        if self.buf.len() == self.frame_len {
            Some(std::mem::replace(
                &mut self.buf,
                Vec::with_capacity(self.frame_len),
            ))
        } else {
            None
        }
    }
}

// ---------------------------------------------------------------------------
// Playback
// ---------------------------------------------------------------------------

fn run_playback(
    rx: Receiver<Vec<i16>>,
    target: Option<String>,
    ring_fill: Arc<AtomicUsize>,
    running: Arc<AtomicBool>,
    ready: Sender<Result<(), AudioError>>,
) {
    let rb = HeapRb::<i16>::new(RING_CAPACITY);
    let (mut producer, consumer) = rb.split();

    // Prefill to the steady-state depth so playout starts at target. `ring_fill`
    // is an EXACT occupancy counter maintained by deltas (worker fetch_add on
    // push, RT callback fetch_sub on pop) — not ringbuf's cached `occupied_len`,
    // which is stale across the split halves and would lie high and starve the
    // ring. See pipewire_impl.rs for the full rationale.
    for _ in 0..PLAYBACK_TARGET_SAMPLES {
        let _ = producer.try_push(0);
    }
    ring_fill.store(PLAYBACK_TARGET_SAMPLES, Ordering::Relaxed);

    // Diagnostics (mirrors the PipeWire backend's playout-health line).
    let underrun = Arc::new(AtomicU64::new(0));
    let dropped = Arc::new(AtomicU64::new(0));
    // Largest single output-callback length seen (interleaved samples). WASAPI
    // shared-mode picks its own period, so this can exceed the prefill target —
    // which would force an underrun every cycle (review W2). The callback only
    // does a wait-free fetch_max; the health logger reports/warns off the RT path.
    let max_cb = Arc::new(AtomicUsize::new(0));
    // Stream-liveness signals (see the capture path / W1, W2, B1): each RT callback
    // bumps `callbacks`; the RT error callback sets `err_code` without logging.
    let callbacks = Arc::new(AtomicUsize::new(0));
    let err_code = Arc::new(AtomicU8::new(STREAM_ERR_NONE));

    // Fallible device/stream setup. `consumer` is moved into the output callback.
    let setup = || -> Result<(Stream, String, SampleFormat, usize, u32), AudioError> {
        let (device, config, sample_format) = resolve(true, target)?;
        let channels = config.channels as usize;
        let device_rate = config.sample_rate.0;
        let stream = match sample_format {
            SampleFormat::F32 => build_output::<f32, _>(
                &device, &config, consumer, ring_fill.clone(), underrun.clone(),
                max_cb.clone(), callbacks.clone(), err_code.clone(),
            ),
            SampleFormat::I16 => build_output::<i16, _>(
                &device, &config, consumer, ring_fill.clone(), underrun.clone(),
                max_cb.clone(), callbacks.clone(), err_code.clone(),
            ),
            SampleFormat::U16 => build_output::<u16, _>(
                &device, &config, consumer, ring_fill.clone(), underrun.clone(),
                max_cb.clone(), callbacks.clone(), err_code.clone(),
            ),
            other => Err(AudioError::Stream(format!(
                "unsupported playback sample format: {other:?}"
            ))),
        }?;
        stream
            .play()
            .map_err(|e| AudioError::Stream(e.to_string()))?;
        let name = device.name().unwrap_or_else(|_| "<unknown>".to_string());
        Ok((stream, name, sample_format, channels, device_rate))
    };

    // Build + queue, then wait for real callbacks before reporting readiness (W1).
    let (stream, dev_name, sample_format, channels, device_rate) = match setup() {
        Ok(v) => match wait_for_stream_start(&callbacks, &err_code, &running) {
            Ok(()) => {
                let _ = ready.send(Ok(()));
                v
            }
            Err(e) => {
                // Drop before reporting so a wedged Stream::drop withholds the Err
                // and lets finish_start's backstop detach instead of hanging (B2).
                drop(v.0);
                let _ = ready.send(Err(e));
                return;
            }
        },
        Err(e) => {
            let _ = ready.send(Err(e));
            return;
        }
    };
    crate::log_msg(&format!(
        "cpal playback started: device='{dev_name}' format={sample_format:?} channels={channels} device_rate={device_rate} Hz <- {SAMPLE_RATE} Hz"
    ));

    let logger = spawn_health_logger(
        running.clone(),
        ring_fill.clone(),
        underrun.clone(),
        dropped.clone(),
        max_cb.clone(),
        err_code.clone(),
    );

    // Feed the ring from the network mixer until `stop()` flips `running` or the
    // sender disconnects (call ended). Clock-paced production keeps the ring near
    // target, so the drop path below should never fire in steady state.
    drain_loop(&rx, &running, |frame| {
        if ring_fill.load(Ordering::Relaxed) + frame.len() > RING_CAPACITY {
            dropped.fetch_add(1, Ordering::Relaxed);
            return;
        }
        // Reserve occupancy BEFORE publishing samples, and publish the whole frame
        // in one `push_slice` (review W3). Per-sample pushes let the RT consumer
        // observe a half-written stereo pair (L without R) → channel tear, and a
        // pop that raced the post-loop `fetch_add` could drive `ring_fill` below
        // zero and wrap it to usize::MAX, wedging the mixer's pacing. Reserving
        // first means the consumer can never pop a sample that isn't yet counted.
        ring_fill.fetch_add(frame.len(), Ordering::Relaxed);
        let pushed = producer.push_slice(&frame);
        if pushed != frame.len() {
            // The capacity check above should make this unreachable (the consumer
            // only drains), but stay exact if it ever isn't.
            ring_fill.fetch_sub(frame.len() - pushed, Ordering::Relaxed);
            dropped.fetch_add(1, Ordering::Relaxed);
        }
    });

    // We're shutting down (either stop() or disconnect). Ensure the logger sees it
    // even on the disconnect path, then drop the stream.
    running.store(false, Ordering::Relaxed);
    let _ = logger.join();
    drop(stream);
}

#[allow(clippy::too_many_arguments)]
fn build_output<T, C>(
    device: &Device,
    config: &StreamConfig,
    mut consumer: C,
    ring_fill: Arc<AtomicUsize>,
    underrun: Arc<AtomicU64>,
    max_cb: Arc<AtomicUsize>,
    callbacks: Arc<AtomicUsize>,
    err_code: Arc<AtomicU8>,
) -> Result<Stream, AudioError>
where
    T: SizedSample + FromSample<i16> + Send + 'static,
    C: Consumer<Item = i16> + Send + 'static,
{
    // RT-safe error callback: atomic store only, no alloc/log (review W2).
    let err_fn = move |e: cpal::StreamError| err_code.store(stream_err_code(&e), Ordering::Relaxed);
    let device_rate = config.sample_rate.0;
    let device_channels = config.channels as usize;
    if device_rate == SAMPLE_RATE && device_channels == PLAYBACK_CHANNELS {
        device
            .build_output_stream::<T, _, _>(
                config,
                move |data: &mut [T], _| {
                    callbacks.fetch_add(1, Ordering::Relaxed);
                    // Record demand in INTERNAL 48 kHz-stereo samples (not raw
                    // device samples) so the health logger's prefill-target
                    // comparison is apples-to-apples for any rate/layout (W4).
                    max_cb.fetch_max(
                        internal_demand(data.len(), device_channels, device_rate),
                        Ordering::Relaxed,
                    );
                    let (popped, starved) = fill_output(&mut consumer, data);
                    if starved > 0 {
                        underrun.fetch_add(starved, Ordering::Relaxed);
                    }
                    if popped > 0 {
                        // Decrement the exact occupancy by what we actually pulled
                        // (underruns removed nothing) so the mixer paces against the
                        // true ring depth.
                        ring_fill.fetch_sub(popped, Ordering::Relaxed);
                    }
                },
                err_fn,
                None,
            )
            .map_err(|e| AudioError::Stream(e.to_string()))
    } else {
        let mut resampler = StereoPullResampler::new(SAMPLE_RATE, device_rate);
        device
            .build_output_stream::<T, _, _>(
                config,
                move |data: &mut [T], _| {
                    callbacks.fetch_add(1, Ordering::Relaxed);
                    max_cb.fetch_max(
                        internal_demand(data.len(), device_channels, device_rate),
                        Ordering::Relaxed,
                    );
                    let (popped, starved) =
                        fill_output_remap(&mut consumer, data, device_channels, &mut resampler);
                    if starved > 0 {
                        underrun.fetch_add(starved, Ordering::Relaxed);
                    }
                    if popped > 0 {
                        // Decrement the exact occupancy by what we actually pulled
                        // (underruns removed nothing) so the mixer paces against the
                        // true ring depth.
                        ring_fill.fetch_sub(popped, Ordering::Relaxed);
                    }
                },
                err_fn,
                None,
            )
            .map_err(|e| AudioError::Stream(e.to_string()))
    }
}

/// Convert an output callback's raw device-sample length into the equivalent
/// internal 48 kHz-stereo sample demand, so the prefill-target comparison stays
/// meaningful regardless of the device's rate/channel layout (W4 diagnostic fix).
/// Integer-only and allocation-free, so it is safe on the RT callback thread.
#[inline]
fn internal_demand(device_len: usize, device_channels: usize, device_rate: u32) -> usize {
    let device_frames = device_len / device_channels.max(1);
    let need_frames =
        (device_frames as u64 * SAMPLE_RATE as u64).div_ceil(device_rate.max(1) as u64) as usize;
    need_frames * PLAYBACK_CHANNELS
}

/// Drain the ring into the device buffer, substituting silence on underrun.
/// Returns `(samples_popped, samples_starved)`. RT-safe (wait-free `try_pop`).
fn fill_output<T, C>(consumer: &mut C, out: &mut [T]) -> (usize, u64)
where
    T: Sample + FromSample<i16>,
    C: Consumer<Item = i16>,
{
    let mut popped = 0usize;
    let mut starved = 0u64;
    for slot in out.iter_mut() {
        match consumer.try_pop() {
            Some(v) => {
                *slot = T::from_sample(v);
                popped += 1;
            }
            None => {
                *slot = T::from_sample(0i16);
                starved += 1;
            }
        }
    }
    (popped, starved)
}

/// Resample/remap internal 48 kHz stereo ring samples into the device buffer.
/// Returns `(internal_samples_popped, device_samples_starved)`. RT-safe.
fn fill_output_remap<T, C>(
    consumer: &mut C,
    out: &mut [T],
    device_channels: usize,
    resampler: &mut StereoPullResampler,
) -> (usize, u64)
where
    T: Sample + FromSample<i16>,
    C: Consumer<Item = i16>,
{
    let mut popped = 0usize;
    let mut starved = 0u64;
    for frame in out.chunks_mut(device_channels) {
        match resampler.next(|| {
            let l = match consumer.try_pop() {
                Some(v) => {
                    popped += 1;
                    v
                }
                None => return None,
            };
            let r = match consumer.try_pop() {
                Some(v) => {
                    popped += 1;
                    v
                }
                None => return None,
            };
            Some((i16_to_f32(l), i16_to_f32(r)))
        }) {
            Some((l, r)) => {
                if device_channels == 1 {
                    frame[0] = T::from_sample(f32_to_i16((l + r) * 0.5));
                } else {
                    frame[0] = T::from_sample(f32_to_i16(l));
                    frame[1] = T::from_sample(f32_to_i16(r));
                    for slot in &mut frame[2..] {
                        *slot = T::from_sample(0i16);
                    }
                }
            }
            None => {
                for slot in frame {
                    *slot = T::from_sample(0i16);
                }
                starved += device_channels as u64;
            }
        }
    }
    (popped, starved)
}

/// Once-per-second playout-health line (mirrors the PipeWire backend). Quiet
/// unless a second actually glitched, or `PEERSPEAK_AUDIO_VERBOSE` is set.
fn spawn_health_logger(
    running: Arc<AtomicBool>,
    ring_fill: Arc<AtomicUsize>,
    underrun: Arc<AtomicU64>,
    dropped: Arc<AtomicU64>,
    max_cb: Arc<AtomicUsize>,
    err_code: Arc<AtomicU8>,
) -> JoinHandle<()> {
    let verbose = std::env::var_os("PEERSPEAK_AUDIO_VERBOSE").is_some();
    thread::spawn(move || {
        let (mut last_u, mut last_d) = (0u64, 0u64);
        let mut reported_cb = 0usize;
        let mut last_err = STREAM_ERR_NONE;
        while running.load(Ordering::Relaxed) {
            thread::sleep(Duration::from_secs(1));
            let u = underrun.load(Ordering::Relaxed);
            let d = dropped.load(Ordering::Relaxed);
            let fill = ring_fill.load(Ordering::Relaxed);
            let (du, dd) = (u - last_u, d - last_d);
            last_u = u;
            last_d = d;
            if verbose || du > 0 || dd > 0 {
                crate::log_msg(&format!(
                    "playout-health: fill={fill} samples (~{}ms) | underrun +{du} samples/s (total {u}) | dropped +{dd} frames/s (total {d})",
                    fill / (48 * PLAYBACK_CHANNELS),
                ));
            }
            // Surface a stream error the RT callback flagged (it can't log itself).
            let ec = err_code.load(Ordering::Relaxed);
            if ec != STREAM_ERR_NONE && ec != last_err {
                crate::log_msg(&format!("cpal playback stream error: {}", stream_err_text(ec)));
                last_err = ec;
            }
            // Report the device's per-cycle demand (in internal 48 kHz-stereo
            // samples) the first time it's seen, and on any new high. If a callback
            // demands more than the prefill target, the ring can't satisfy it and
            // underruns every cycle — the W2 signature; warn so a real-host log
            // shows whether it's biting.
            let cb = max_cb.load(Ordering::Relaxed);
            if cb > reported_cb {
                reported_cb = cb;
                let ms = cb / (48 * PLAYBACK_CHANNELS);
                if cb > PLAYBACK_TARGET_SAMPLES {
                    crate::log_msg(&format!(
                        "cpal output callback demands up to {cb} internal samples/cycle (~{ms}ms) EXCEEDS prefill target {PLAYBACK_TARGET_SAMPLES} — expect periodic underruns; needs a larger target or a fixed buffer size (review W2)",
                    ));
                } else if verbose {
                    crate::log_msg(&format!(
                        "cpal output callback demands up to {cb} internal samples/cycle (~{ms}ms), target {PLAYBACK_TARGET_SAMPLES}",
                    ));
                }
            }
        }
    })
}

/// Pump frames from `rx` to `on_frame` until `running` goes false or the sender
/// disconnects. The timed receive re-checks `running` at least every
/// [`WORKER_POLL`], so `stop()` can join the worker promptly instead of hanging
/// on a parked blocking `recv()` (same A7 fix as the PipeWire backend). Pure
/// w.r.t. its inputs, so it's unit-testable.
fn drain_loop(rx: &Receiver<Vec<i16>>, running: &AtomicBool, mut on_frame: impl FnMut(Vec<i16>)) {
    while running.load(Ordering::Relaxed) {
        match rx.recv_timeout(WORKER_POLL) {
            Ok(frame) => on_frame(frame),
            Err(RecvTimeoutError::Timeout) => continue,
            Err(RecvTimeoutError::Disconnected) => return,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::mpsc;

    #[test]
    fn downmix_averages_channels() {
        assert_eq!(downmix_to_mono::<i16>(&[100, 100]), 100);
        assert_eq!(downmix_to_mono::<i16>(&[100, -100]), 0);
        assert_eq!(downmix_to_mono::<i16>(&[50]), 50);
        assert_eq!(downmix_to_mono::<i16>(&[]), 0);
        // 4-channel average rounds toward zero (integer division).
        assert_eq!(downmix_to_mono::<i16>(&[10, 20, 30, 41]), 25);
    }

    #[test]
    fn frame_accumulator_emits_full_frames() {
        let mut acc = FrameAccumulator::new(3);
        assert_eq!(acc.push(1), None);
        assert_eq!(acc.push(2), None);
        assert_eq!(acc.push(3), Some(vec![1, 2, 3]));
        // Resets for the next frame.
        assert_eq!(acc.push(4), None);
        assert_eq!(acc.push(5), None);
        assert_eq!(acc.push(6), Some(vec![4, 5, 6]));
    }

    #[test]
    fn fill_output_pops_then_substitutes_silence() {
        let rb = HeapRb::<i16>::new(8);
        let (mut prod, mut cons) = rb.split();
        for v in [1, 2, 3] {
            prod.try_push(v).unwrap();
        }
        let mut out = [0i16; 5];
        let (popped, starved) = fill_output(&mut cons, &mut out);
        assert_eq!(popped, 3);
        assert_eq!(starved, 2);
        assert_eq!(out, [1, 2, 3, 0, 0]);
    }

    #[test]
    fn fill_output_remap_downmixes_to_mono() {
        let rb = HeapRb::<i16>::new(8);
        let (mut prod, mut cons) = rb.split();
        for v in [100, 300, 500, -100, 7, 9] {
            prod.try_push(v).unwrap();
        }
        let mut out = [0i16; 2];
        let mut resampler = StereoPullResampler::new(SAMPLE_RATE, SAMPLE_RATE);
        let (popped, starved) = fill_output_remap(&mut cons, &mut out, 1, &mut resampler);
        assert_eq!(popped, 6);
        assert_eq!(starved, 0);
        assert_eq!(out, [200, 200]);
    }

    #[test]
    fn fill_output_remap_silences_underrun() {
        let rb = HeapRb::<i16>::new(8);
        let (_prod, mut cons) = rb.split();
        let mut out = [11i16; 4];
        let mut resampler = StereoPullResampler::new(SAMPLE_RATE, SAMPLE_RATE);
        let (popped, starved) = fill_output_remap(&mut cons, &mut out, 2, &mut resampler);
        assert_eq!(popped, 0);
        assert_eq!(starved, out.len() as u64);
        assert_eq!(out, [0, 0, 0, 0]);
    }

    #[test]
    fn fill_output_remap_copies_stereo_at_matching_rate() {
        let rb = HeapRb::<i16>::new(8);
        let (mut prod, mut cons) = rb.split();
        for v in [1, -1, 2, -2, 3, -3] {
            prod.try_push(v).unwrap();
        }
        let mut out = [0i16; 4];
        let mut resampler = StereoPullResampler::new(SAMPLE_RATE, SAMPLE_RATE);
        let (popped, starved) = fill_output_remap(&mut cons, &mut out, 2, &mut resampler);
        assert_eq!(popped, 6);
        assert_eq!(starved, 0);
        assert_eq!(out, [1, -1, 2, -2]);
    }

    #[test]
    fn drain_loop_exits_when_running_flips_even_with_sender_alive() {
        let (tx, rx) = mpsc::channel::<Vec<i16>>();
        let running = Arc::new(AtomicBool::new(true));
        let r2 = running.clone();
        let h = thread::spawn(move || drain_loop(&rx, &r2, |_| {}));
        thread::sleep(Duration::from_millis(50));
        running.store(false, Ordering::Relaxed);
        thread::sleep(WORKER_POLL + Duration::from_millis(150));
        assert!(
            h.is_finished(),
            "drain_loop must exit after running=false even while the sender is alive"
        );
        drop(tx);
        h.join().unwrap();
    }

    #[test]
    fn drain_loop_returns_on_disconnect() {
        let (tx, rx) = mpsc::channel::<Vec<i16>>();
        let running = Arc::new(AtomicBool::new(true));
        drop(tx);
        drain_loop(&rx, &running, |_| panic!("no frame should arrive"));
    }

    #[test]
    fn bounded_rate_prefers_48k_when_in_window() {
        // A device span that contains 48 kHz resolves exactly.
        assert_eq!(bounded_rate(44_100, 96_000), Some(SAMPLE_RATE));
        assert_eq!(
            bounded_rate(MIN_DEVICE_RATE, MAX_DEVICE_RATE),
            Some(SAMPLE_RATE)
        );
    }

    #[test]
    fn bounded_rate_clamps_to_nearest_in_window_bound() {
        // Entirely below 48 kHz → the top bound (closest reachable to 48 kHz).
        assert_eq!(bounded_rate(8_000, 16_000), Some(16_000));
        // Entirely above 48 kHz → the bottom bound.
        assert_eq!(bounded_rate(88_200, 192_000), Some(88_200));
    }

    #[test]
    fn bounded_rate_rejects_spans_outside_the_window() {
        assert_eq!(bounded_rate(1_000, 4_000), None); // below the floor
        assert_eq!(bounded_rate(400_000, 500_000), None); // above the ceiling
    }

    #[test]
    fn bounded_rate_intersects_window_edges() {
        // Overlaps only the floor: [4k, 8k] ∩ [8k, 384k] = {8k}.
        assert_eq!(bounded_rate(4_000, MIN_DEVICE_RATE), Some(MIN_DEVICE_RATE));
        // Overlaps only the ceiling.
        assert_eq!(
            bounded_rate(MAX_DEVICE_RATE, 500_000),
            Some(MAX_DEVICE_RATE)
        );
    }

    #[test]
    fn format_supported_matches_the_stream_builders() {
        // Exactly the three the build_input/build_output match arms accept.
        for f in [SampleFormat::F32, SampleFormat::I16, SampleFormat::U16] {
            assert!(format_supported(f), "{f:?} should be drivable");
        }
        // Everything else cpal can expose must be filtered out before ranking, or a
        // start could pick it and then hit the `unsupported sample format` arm (P3).
        for f in [
            SampleFormat::I8,
            SampleFormat::U8,
            SampleFormat::I32,
            SampleFormat::U32,
            SampleFormat::I64,
            SampleFormat::U64,
            SampleFormat::F64,
        ] {
            assert!(!format_supported(f), "{f:?} must not be reported drivable");
        }
    }

    #[test]
    fn ensure_idle_allows_an_idle_slot() {
        let mut s = SlotState::Idle;
        assert!(ensure_idle(&mut s, "capture").is_ok());
        assert!(matches!(s, SlotState::Idle));
    }

    #[test]
    fn ensure_idle_rejects_a_live_wedged_orphan_then_clears_when_it_exits() {
        let exited = Arc::new(AtomicBool::new(false));
        let mut s = SlotState::Wedged {
            exited: exited.clone(),
        };
        // Orphan still alive → reject, tombstone preserved.
        assert!(ensure_idle(&mut s, "playback").is_err());
        assert!(matches!(s, SlotState::Wedged { .. }));
        // Orphan's ExitGuard fired → the next start clears the tombstone and proceeds.
        exited.store(true, Ordering::Relaxed);
        assert!(ensure_idle(&mut s, "playback").is_ok());
        assert!(matches!(s, SlotState::Idle));
    }

    #[test]
    fn drain_loop_delivers_frames() {
        let (tx, rx) = mpsc::channel::<Vec<i16>>();
        let running = Arc::new(AtomicBool::new(true));
        let r2 = running.clone();
        let got = Arc::new(Mutex::new(Vec::new()));
        let g2 = got.clone();
        let h = thread::spawn(move || drain_loop(&rx, &r2, |f| g2.lock().unwrap().push(f)));
        tx.send(vec![1, 2, 3]).unwrap();
        tx.send(vec![4, 5]).unwrap();
        thread::sleep(Duration::from_millis(50));
        running.store(false, Ordering::Relaxed);
        drop(tx);
        h.join().unwrap();
        assert_eq!(*got.lock().unwrap(), vec![vec![1, 2, 3], vec![4, 5]]);
    }
}