//! `cpal`-backed [`Listener`] implementation.
//!
//! Captures from the OS default input device on a dedicated `std::thread`
//! (cpal streams are `!Send`), runs the audio through a [`VadBackend`], and
//! ferries finalized speech segments to a `tokio` STT worker. The worker
//! calls [`elevenlabs_stt::transcribe`] and forwards results to the consumer
//! as [`VoiceEvent::Transcript`].

use crate::enrollment::TranscriptRole;
use crate::provider::build_stt_provider;
#[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
use crate::vad_backend::MlxSileroVadBackend;
use crate::vad_backend::{EnergyVadBackend, SileroVadBackend, VadBackend};
use crate::{Listener, Result, VoiceConfig, VoiceError, VoiceEvent};
use async_trait::async_trait;
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use tokio::sync::mpsc;

/// Bounded receiver capacity for VoiceEvents going to the consumer.
const EVENT_CHANNEL_CAPACITY: usize = 64;
/// Bounded receiver capacity for raw audio segments going to the STT worker.
///
/// At cap=8 with `max_segment_ms = 6000`, the consumer has 48 s of
/// runway before the channel fills and `try_send` starts dropping.
/// Fine when whisper transcribes faster than realtime; not fine when
/// it doesn't (#120: linux-x64 binary running through Rosetta on
/// Apple Silicon takes 30–60 s per 6 s segment, fills the channel in
/// the first minute, drops everything after). Bumping to 32 buys
/// roughly 3 minutes of buffer — enough to absorb a slow first-call
/// warm-up while still bounded against unbounded memory growth.
/// 32 × ~1 MB/segment ≈ 32 MB at peak.
const AUDIO_CHANNEL_CAPACITY: usize = 32;
/// Minimum segment length to bother sending to STT.
const MIN_SEGMENT_MS: u64 = 300;
/// Sample rate the listener delivers to VAD and STT regardless of what
/// the device negotiates. Whisper expects 16 kHz mono; macOS VPIO
/// already delivers 16 kHz mono after its sample-rate converter, so
/// matching that on Linux/Windows keeps every downstream consumer on
/// one consistent shape.
const TARGET_SAMPLE_RATE: u32 = 16_000;

/// Default microphone listener using `cpal` + ElevenLabs STT.
pub struct CpalListener {
    running: Arc<AtomicBool>,
    stt_task: Option<tokio::task::JoinHandle<()>>,
    capture_thread: Option<std::thread::JoinHandle<()>>,
    /// Optional speaker diarizer applied to each finalized segment
    /// before the Transcript event is emitted. When `None`, the
    /// `role` field defaults to [`TranscriptRole::Unknown`].
    #[cfg(feature = "diarization")]
    diarizer: Option<crate::diarization::SharedDiarizer>,
    /// Optional Parakeet provider used for streaming partials. When
    /// set, the STT worker bypasses the configured provider and uses
    /// `transcribe_streaming` to emit `VoiceEvent::Partial` events as
    /// the decoder progresses, plus a final `Transcript` event.
    #[cfg(feature = "parakeet")]
    parakeet_streaming: Option<std::sync::Arc<crate::parakeet_stt::ParakeetSttProvider>>,
    /// Optional Apple-Speech provider used for streaming partials on
    /// macOS. Same role as `parakeet_streaming` — STT worker uses
    /// `transcribe_streaming` and emits `VoiceEvent::Partial` per
    /// snapshot, then a final `Transcript`. macOS only.
    #[cfg(target_os = "macos")]
    apple_speech_streaming: Option<std::sync::Arc<crate::apple_speech_stt::AppleSpeechSttProvider>>,
    /// Optional enrollment-based speaker pipeline. When set, the STT
    /// worker classifies finalized segments via voiceprint cosine
    /// similarity — segments matching the enrolled user get
    /// `role: "enrolled_user"`. Layered with the diarizer: the
    /// pipeline runs first; if it can identify the speaker we keep
    /// that role, otherwise we fall back to the diarizer's cluster
    /// label.
    speaker_pipeline: Option<std::sync::Arc<crate::enrollment::SpeakerPipeline>>,
}

impl CpalListener {
    pub fn new() -> Self {
        Self {
            running: Arc::new(AtomicBool::new(false)),
            stt_task: None,
            capture_thread: None,
            #[cfg(feature = "diarization")]
            diarizer: None,
            #[cfg(feature = "parakeet")]
            parakeet_streaming: None,
            #[cfg(target_os = "macos")]
            apple_speech_streaming: None,
            speaker_pipeline: None,
        }
    }

    /// Attach an enrollment-based [`crate::enrollment::SpeakerPipeline`].
    /// Each finalized segment is classified against the pipeline's
    /// voiceprint(s); known speakers get tagged in
    /// [`crate::enrollment::TranscriptRole::EnrolledUser`].
    pub fn with_speaker_pipeline(
        mut self,
        pipeline: std::sync::Arc<crate::enrollment::SpeakerPipeline>,
    ) -> Self {
        self.speaker_pipeline = Some(pipeline);
        self
    }

    /// Attach a speaker diarizer. Each finalized speech segment is
    /// classified before the Transcript event is emitted, so
    /// consumers see `role: "other:speaker_N"` rather than
    /// `role: "unknown"`.
    #[cfg(feature = "diarization")]
    pub fn with_diarizer(mut self, diarizer: crate::diarization::SharedDiarizer) -> Self {
        self.diarizer = Some(diarizer);
        self
    }

    /// Use Parakeet TDT for streaming transcription. The STT worker
    /// will emit `VoiceEvent::Partial` per non-blank token plus a
    /// final `VoiceEvent::Transcript` when each VAD-finalized
    /// segment completes. Bypasses the configured `SttProvider`
    /// (`config.stt_provider`) for transcription; only the VAD path
    /// uses it.
    #[cfg(feature = "parakeet")]
    pub fn with_parakeet_streaming(
        mut self,
        provider: std::sync::Arc<crate::parakeet_stt::ParakeetSttProvider>,
    ) -> Self {
        self.parakeet_streaming = Some(provider);
        self
    }

    /// Use Apple's SFSpeechRecognizer for streaming transcription on
    /// macOS. Same shape as [`Self::with_parakeet_streaming`] — STT
    /// worker emits `VoiceEvent::Partial` per snapshot and a final
    /// `VoiceEvent::Transcript` when each VAD-finalized segment
    /// completes. Apple's snapshots can revise (whisper-style); the
    /// terminal Transcript event is the canonical text.
    ///
    /// macOS only. Bypasses the configured `SttProvider` for the
    /// transcription step; only the VAD path uses it.
    #[cfg(target_os = "macos")]
    pub fn with_apple_speech_streaming(
        mut self,
        provider: std::sync::Arc<crate::apple_speech_stt::AppleSpeechSttProvider>,
    ) -> Self {
        self.apple_speech_streaming = Some(provider);
        self
    }
}

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

impl Drop for CpalListener {
    fn drop(&mut self) {
        if self.running.load(Ordering::SeqCst) {
            tracing::warn!("[voice] CpalListener dropped while running; joining capture thread");
            self.running.store(false, Ordering::SeqCst);
            if let Some(handle) = self.capture_thread.take() {
                let _ = handle.join();
            }
            if let Some(task) = self.stt_task.take() {
                task.abort();
            }
        }
    }
}

#[async_trait]
impl Listener for CpalListener {
    async fn start(&mut self, config: VoiceConfig) -> Result<mpsc::Receiver<VoiceEvent>> {
        if self.running.load(Ordering::SeqCst) {
            return Err(VoiceError::AlreadyRunning);
        }

        // Validate before flipping the running flag — any error here must
        // leave the listener idle so the caller can fix and retry.
        let stt_provider = build_stt_provider(&config)?;
        self.running.store(true, Ordering::SeqCst);

        let (event_tx, event_rx) = mpsc::channel::<VoiceEvent>(EVENT_CHANNEL_CAPACITY);
        let (audio_tx, mut audio_rx) = mpsc::channel::<AudioSegment>(AUDIO_CHANNEL_CAPACITY);

        // STT worker — drains finalized segments, calls the configured STT
        // provider, forwards transcripts to the consumer. Lives until
        // audio_tx is dropped or the task is aborted.
        //
        // STT errors are propagated as `Transcript` events with the text
        // prefixed by `[stt-error] ` so the consumer can surface them in
        // the UI. Without this, STT failures are silent — only visible in
        // the logs the user never reads.
        let event_tx_for_stt = event_tx.clone();
        #[cfg(feature = "diarization")]
        let diarizer_for_stt = self.diarizer.clone();
        #[cfg(feature = "parakeet")]
        let parakeet_for_stt = self.parakeet_streaming.clone();
        #[cfg(target_os = "macos")]
        let apple_speech_for_stt = self.apple_speech_streaming.clone();
        let pipeline_for_stt = self.speaker_pipeline.clone();
        // Spawn on the process-global voice runtime, not the caller's
        // tokio runtime. The caller is typically a `#[napi] pub async
        // fn` that returns to JS as soon as `start()` finishes — at
        // which point the per-call runtime context can vanish, and
        // the spawned task never gets polled. #120 root cause.
        let stt_handle = crate::voice_runtime_handle().spawn(async move {
            tracing::info!("[voice] STT consumer task started");
            while let Some(segment) = audio_rx.recv().await {
                tracing::trace!(
                    samples_len = segment.samples.len(),
                    sample_rate = segment.sample_rate,
                    duration_ms = segment.duration_ms,
                    "[voice] consumer dequeued segment"
                );
                #[cfg(feature = "parakeet")]
                if let Some(parakeet) = parakeet_for_stt.as_ref() {
                    // Streaming path — emit Partial per token, final
                    // Transcript at the end. Diarization runs on the
                    // full segment after streaming completes.
                    let event_tx_partials = event_tx_for_stt.clone();
                    let duration_ms = segment.duration_ms;
                    let on_partial = move |p: crate::parakeet_stt::ParakeetPartial| {
                        if p.is_final {
                            // Skip the final partial — we'll send the
                            // canonical Transcript event next, with
                            // diarization applied. Avoids duplicate UI
                            // updates.
                            return;
                        }
                        let tx = event_tx_partials.clone();
                        // Fire-and-forget: try_send drops the partial
                        // if the consumer is backed up rather than
                        // stalling the decoder.
                        let _ = tx.try_send(VoiceEvent::Partial {
                            text: p.text,
                            duration_ms,
                        });
                    };
                    match parakeet
                        .transcribe_streaming(&segment.samples, segment.sample_rate, on_partial)
                        .await
                    {
                        Ok(text) if !text.is_empty() => {
                            let role = classify_segment(
                                #[cfg(feature = "diarization")]
                                diarizer_for_stt.as_ref(),
                                pipeline_for_stt.as_ref(),
                                &segment.samples,
                                segment.sample_rate,
                            );
                            let _ = event_tx_for_stt
                                .send(VoiceEvent::Transcript {
                                    text,
                                    duration_ms: segment.duration_ms,
                                    role,
                                })
                                .await;
                        }
                        Ok(_) => tracing::debug!("[voice] empty parakeet transcript"),
                        Err(e) => {
                            tracing::error!("[voice] parakeet streaming failed: {e}");
                            let _ = event_tx_for_stt
                                .send(VoiceEvent::Transcript {
                                    text: format!("[stt-error] {e}"),
                                    duration_ms: segment.duration_ms,
                                    role: TranscriptRole::Unknown,
                                })
                                .await;
                        }
                    }
                    continue;
                }
                #[cfg(target_os = "macos")]
                if let Some(apple) = apple_speech_for_stt.as_ref() {
                    // Apple-Speech streaming path — emit Partial per
                    // recognizer snapshot, final Transcript at the end.
                    // Same fire-and-forget try_send pattern as the
                    // parakeet branch — partials are advisory, not
                    // canonical, so dropping them under backpressure is
                    // preferable to stalling the recognizer.
                    let event_tx_partials = event_tx_for_stt.clone();
                    let duration_ms = segment.duration_ms;
                    let on_partial = move |p: crate::apple_speech_stt::AppleSpeechPartial| {
                        let tx = event_tx_partials.clone();
                        let _ = tx.try_send(VoiceEvent::Partial {
                            text: p.text,
                            duration_ms,
                        });
                    };
                    match apple
                        .transcribe_streaming(&segment.samples, segment.sample_rate, on_partial)
                        .await
                    {
                        Ok(text) if !text.is_empty() => {
                            let role = classify_segment(
                                #[cfg(feature = "diarization")]
                                diarizer_for_stt.as_ref(),
                                pipeline_for_stt.as_ref(),
                                &segment.samples,
                                segment.sample_rate,
                            );
                            let _ = event_tx_for_stt
                                .send(VoiceEvent::Transcript {
                                    text,
                                    duration_ms: segment.duration_ms,
                                    role,
                                })
                                .await;
                        }
                        Ok(_) => tracing::debug!("[voice] empty apple-speech transcript"),
                        Err(e) => {
                            tracing::error!("[voice] apple-speech streaming failed: {e}");
                            let _ = event_tx_for_stt
                                .send(VoiceEvent::Transcript {
                                    text: format!("[stt-error] {e}"),
                                    duration_ms: segment.duration_ms,
                                    role: TranscriptRole::Unknown,
                                })
                                .await;
                        }
                    }
                    continue;
                }
                // Batch path — single Transcript per finalized segment.
                // We *always* emit a Transcript event, even for empty
                // text (whisper's no_speech_thold filtered everything,
                // or the chunk decoded to nothing). The earlier silent
                // drop made #120's "transcribe is/isn't being called"
                // diagnosis impossible — bot.js couldn't distinguish
                // "consumer never ran" from "consumer ran 21 times,
                // every result was empty." Emitting empty Transcript
                // events restores the signal.
                match stt_provider
                    .transcribe(&segment.samples, segment.sample_rate)
                    .await
                {
                    Ok(text) => {
                        // Skip diarization for empty text — it'd be
                        // expensive and the role can't be meaningful.
                        let role = if text.is_empty() {
                            TranscriptRole::Unknown
                        } else {
                            classify_segment(
                                #[cfg(feature = "diarization")]
                                diarizer_for_stt.as_ref(),
                                pipeline_for_stt.as_ref(),
                                &segment.samples,
                                segment.sample_rate,
                            )
                        };
                        let _ = event_tx_for_stt
                            .send(VoiceEvent::Transcript {
                                text,
                                duration_ms: segment.duration_ms,
                                role,
                            })
                            .await;
                    }
                    Err(e) => {
                        tracing::error!("[voice] STT failed: {e}");
                        let _ = event_tx_for_stt
                            .send(VoiceEvent::Transcript {
                                text: format!("[stt-error] {e}"),
                                duration_ms: segment.duration_ms,
                                role: TranscriptRole::Unknown,
                            })
                            .await;
                    }
                }
            }
        });
        self.stt_task = Some(stt_handle);

        // Capture thread — owns the cpal stream, runs VAD, dispatches events.
        let running = self.running.clone();
        let cfg = config.clone();
        let capture_handle = std::thread::spawn(move || {
            if let Err(e) = run_capture(running.clone(), cfg, event_tx, audio_tx) {
                tracing::error!("[voice] capture thread error: {e}");
                running.store(false, Ordering::SeqCst);
            }
        });
        self.capture_thread = Some(capture_handle);

        Ok(event_rx)
    }

    async fn stop(&mut self) -> Result<()> {
        if !self.running.swap(false, Ordering::SeqCst) {
            return Err(VoiceError::NotRunning);
        }
        // Abort the STT worker first so it stops draining audio segments.
        if let Some(task) = self.stt_task.take() {
            task.abort();
        }
        // Join the cpal capture thread on a blocking pool so we don't stall
        // the async runtime while it polls its 100ms shutdown loop.
        if let Some(handle) = self.capture_thread.take() {
            tokio::task::spawn_blocking(move || {
                if let Err(e) = handle.join() {
                    tracing::warn!("[voice] capture thread join failed: {:?}", e);
                }
            })
            .await
            .map_err(|e| VoiceError::Capture(format!("join task failed: {e}")))?;
        }
        Ok(())
    }

    fn is_running(&self) -> bool {
        self.running.load(Ordering::SeqCst)
    }
}

/// One captured speech segment, ready for STT.
struct AudioSegment {
    samples: Vec<f32>,
    sample_rate: u32,
    duration_ms: u64,
}

/// Downmix interleaved multi-channel f32 PCM to mono by averaging
/// across channels. `(L + R) / 2` (or `(L + R + …) / N`) — never
/// drops a channel, never clips, returns immediately for `channels
/// <= 1`. The trailing partial frame (if `interleaved.len()` isn't
/// a multiple of `channels`) is dropped — cpal callbacks should
/// always hand whole frames, but the chunks_exact path makes the
/// guarantee explicit.
fn downmix_to_mono(interleaved: &[f32], channels: u16) -> Vec<f32> {
    if channels <= 1 {
        return interleaved.to_vec();
    }
    let n = channels as usize;
    let inv = 1.0 / channels as f32;
    interleaved
        .chunks_exact(n)
        .map(|frame| frame.iter().sum::<f32>() * inv)
        .collect()
}

/// Cheap linear-interpolation sample-rate conversion. Good enough
/// for VAD (energy detection is robust to mild aliasing) and good
/// enough for whisper-rs at 16 kHz to ingest — the alternative is
/// sending 44.1/48 kHz audio whisper has no path for and watching
/// it produce garbage. If quality ever matters here we can swap to
/// rubato's polynomial resampler with a chunk buffer; for now the
/// realtime path stays allocation-light and dependency-free.
fn linear_resample(samples: &[f32], in_rate: u32, out_rate: u32) -> Vec<f32> {
    if in_rate == out_rate || samples.is_empty() {
        return samples.to_vec();
    }
    let in_len = samples.len();
    let ratio = out_rate as f64 / in_rate as f64;
    let out_len = (in_len as f64 * ratio).round() as usize;
    if out_len == 0 {
        return Vec::new();
    }
    let mut out = Vec::with_capacity(out_len);
    let step = in_rate as f64 / out_rate as f64;
    for i in 0..out_len {
        let pos = i as f64 * step;
        let idx = pos.floor() as usize;
        if idx + 1 >= in_len {
            out.push(samples[in_len - 1]);
        } else {
            let frac = (pos - idx as f64) as f32;
            out.push(samples[idx] * (1.0 - frac) + samples[idx + 1] * frac);
        }
    }
    out
}

fn run_capture(
    running: Arc<AtomicBool>,
    config: VoiceConfig,
    event_tx: mpsc::Sender<VoiceEvent>,
    audio_tx: mpsc::Sender<AudioSegment>,
) -> Result<()> {
    let host = cpal::default_host();
    let device = host
        .default_input_device()
        .ok_or_else(|| VoiceError::Device("no default input device".into()))?;

    let stream_config = device
        .default_input_config()
        .map_err(|e| VoiceError::Device(format!("default input config: {e}")))?;

    let device_sample_rate = stream_config.sample_rate().0;
    let device_channels = stream_config.channels();
    // Sample rate VAD and STT see — always 16 kHz after the
    // downmix+resample stage below.
    let sample_rate = TARGET_SAMPLE_RATE;
    tracing::info!(
        "[voice] capture from {} at {}Hz / {}ch (delivering {}Hz mono to VAD)",
        device.name().unwrap_or_default(),
        device_sample_rate,
        device_channels,
        sample_rate
    );

    // Hard cap on segment length. VAD has been observed to lock onto
    // background noise (calibrated low noise floor + continuous
    // ambient-above-threshold) and produce "speech" segments that
    // never naturally end — see #120 / @ian-siddiqi's meet-bot
    // canary. Mirrors the same safety valve already wired into the
    // macOS VPIO listener (`voice_processing_listener.rs`). Without
    // it, `transcribe_stream` on a real Linux PulseAudio source
    // emits exactly one `SpeechStart` and then nothing for the
    // lifetime of the session.
    let max_segment_ms = config.max_segment_ms;

    let recording: Arc<Mutex<Vec<f32>>> = Arc::new(Mutex::new(Vec::new()));
    let is_recording = Arc::new(AtomicBool::new(false));

    // VAD backend selection. Prefer Silero VAD v6 through MLX on Apple
    // Silicon, otherwise the bundled Silero V5 ONNX path, with the
    // energy VAD as a last-resort fallback. Mirrors the wiring in
    // `voice_processing_listener.rs` so the Linux/Docker capture path
    // (used by meet-bot) gets the same neural speech detection that the
    // macOS VPIO path already uses. See parslee-knowledge `quip` for
    // the meet-bot canary that surfaced VAD lock-up under the energy
    // backend on continuous Meet ambience.
    //
    // The Silero crate (`voice_activity_detector 0.2.1`) currently
    // unwraps an `ort` call internally that hits `NotImplemented:
    // GetElementType` on `ort 2.0.0-rc.10`. Wrap construction in
    // `catch_unwind` so the fallback branch actually runs instead of
    // killing the capture thread silently.
    #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
    let mlx_silero_attempt = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        MlxSileroVadBackend::new(sample_rate, &config)
    }));
    #[cfg(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx)))]
    let mlx_silero: Option<Box<dyn VadBackend>> = match mlx_silero_attempt {
        Ok(Ok(silero)) => {
            tracing::info!(
                "[voice] using Silero VAD v6 MLX (neural, sample_rate={} Hz)",
                sample_rate
            );
            Some(Box::new(silero))
        }
        Ok(Err(e)) => {
            tracing::warn!("[voice] Silero VAD v6 MLX load failed ({e}); trying Silero V5 ONNX");
            None
        }
        Err(panic) => {
            let msg = if let Some(s) = panic.downcast_ref::<&str>() {
                s.to_string()
            } else if let Some(s) = panic.downcast_ref::<String>() {
                s.clone()
            } else {
                "unknown panic".to_string()
            };
            tracing::warn!(
                "[voice] Silero VAD v6 MLX panicked during init ({msg}); trying Silero V5 ONNX"
            );
            None
        }
    };
    #[cfg(not(all(target_os = "macos", target_arch = "aarch64", not(car_skip_mlx))))]
    let mlx_silero: Option<Box<dyn VadBackend>> = None;

    let onnx_silero_attempt = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        SileroVadBackend::new(sample_rate, &config)
    }));
    let vad_backend: Box<dyn VadBackend> = if let Some(vad) = mlx_silero {
        vad
    } else {
        match onnx_silero_attempt {
            Ok(Ok(silero)) => {
                tracing::info!(
                    "[voice] using Silero V5 VAD (neural, sample_rate={} Hz)",
                    sample_rate
                );
                Box::new(silero)
            }
            Ok(Err(e)) => {
                tracing::warn!("[voice] Silero VAD load failed ({e}); falling back to energy VAD");
                Box::new(EnergyVadBackend::from_config(sample_rate, &config))
            }
            Err(panic) => {
                let msg = if let Some(s) = panic.downcast_ref::<&str>() {
                    s.to_string()
                } else if let Some(s) = panic.downcast_ref::<String>() {
                    s.clone()
                } else {
                    "unknown panic".to_string()
                };
                tracing::warn!(
                    "[voice] Silero VAD panicked during init ({msg}); falling back to energy VAD"
                );
                Box::new(EnergyVadBackend::from_config(sample_rate, &config))
            }
        }
    };
    let vad: Arc<Mutex<Box<dyn VadBackend>>> = Arc::new(Mutex::new(vad_backend));

    let rec = recording.clone();
    let rec_flag = is_recording.clone();
    let v = vad.clone();
    let event_tx_cb = event_tx.clone();
    let audio_tx_cb = audio_tx.clone();
    let running_cb = running.clone();

    let stream = device
        .build_input_stream(
            &stream_config.into(),
            move |data: &[f32], _: &cpal::InputCallbackInfo| {
                if !running_cb.load(Ordering::Relaxed) {
                    return;
                }

                // Bring whatever cpal hands us (interleaved-stereo at
                // 44.1/48k under PulseAudio, mono 16k from the
                // bot-shaped raw monitor, etc.) onto the canonical
                // 16 kHz mono shape VAD + Whisper expect. Without
                // this, real-world Linux audio sources flow through
                // VAD as 2-channel 44.1 kHz interleaved, the noise
                // floor calibrates against a different per-callback
                // length than the smoothing constants assume, and
                // whisper-rs receives garbage even when a segment
                // does emit.
                let mono = downmix_to_mono(data, device_channels);
                let resampled =
                    linear_resample(&mono, device_sample_rate, TARGET_SAMPLE_RATE);
                let data: &[f32] = &resampled;

                let mut vad_lock = v.lock().unwrap();
                vad_lock.process_samples(data);
                let speech = vad_lock.is_speech_active();
                let recording_now = rec_flag.load(Ordering::Relaxed);

                if speech && !recording_now {
                    tracing::debug!("[voice] speech detected, recording...");
                    rec_flag.store(true, Ordering::Relaxed);
                    let mut r = rec.lock().unwrap();
                    r.clear();
                    r.extend_from_slice(data);
                    if event_tx_cb.try_send(VoiceEvent::SpeechStart).is_err() {
                        tracing::warn!("[voice] dropped SpeechStart event (consumer slow)");
                    }
                } else if recording_now {
                    rec.lock().unwrap().extend_from_slice(data);

                    // Force-end the segment if it has grown past the
                    // hard cap — see the `max_segment_ms` comment in
                    // run_capture above. Without this branch, a VAD
                    // that has locked on continuous baseline audio
                    // never reaches `turn_ended()` and the consumer
                    // never sees a SpeechEnd / Whisper transcript.
                    let current_ms = if sample_rate > 0 {
                        (rec.lock().unwrap().len() as u64 * 1000) / sample_rate as u64
                    } else {
                        0
                    };
                    let force_end = current_ms >= max_segment_ms;
                    if force_end {
                        tracing::warn!(
                            "[voice] forcing segment end at {}ms (VAD lock-up)",
                            current_ms
                        );
                    }

                    if (!speech && vad_lock.turn_ended()) || force_end {
                        rec_flag.store(false, Ordering::Relaxed);
                        let audio: Vec<f32> = std::mem::take(&mut *rec.lock().unwrap());
                        if force_end {
                            // Reset VAD speech-state so we don't
                            // immediately re-enter the same lockup.
                            // Floor / smoothed RMS are left in place;
                            // the next speech-start re-evaluates from
                            // current ambient.
                            vad_lock.reset_speech_state();
                        }
                        drop(vad_lock);

                        let duration_ms = if sample_rate > 0 {
                            (audio.len() as u64 * 1000) / sample_rate as u64
                        } else {
                            0
                        };
                        tracing::debug!("[voice] turn ended, {}ms", duration_ms);
                        if event_tx_cb.try_send(VoiceEvent::SpeechEnd).is_err() {
                            tracing::warn!("[voice] dropped SpeechEnd event (consumer slow)");
                        }

                        if duration_ms >= MIN_SEGMENT_MS {
                            tracing::trace!(
                                duration_ms,
                                samples_len = audio.len(),
                                channel_capacity = audio_tx_cb.capacity(),
                                "[voice] try_send AudioSegment"
                            );
                            // Losing a finalized segment means losing a
                            // transcription — log loudly so we notice.
                            // The most common cause is the consumer
                            // running slower than the capture loop
                            // (e.g. whisper under x86 Rosetta on
                            // Apple Silicon dev hosts, #120). With the
                            // bumped channel capacity we shouldn't see
                            // this except under sustained STT slowness.
                            if let Err(e) = audio_tx_cb.try_send(AudioSegment {
                                samples: audio,
                                sample_rate,
                                duration_ms,
                            }) {
                                tracing::warn!(
                                    duration_ms,
                                    "[voice] dropped speech segment (STT backlog full, consumer can't keep up): {}",
                                    e
                                );
                            }
                        }
                    }
                }
            },
            |err| {
                tracing::error!("[voice] capture error: {err}");
            },
            None,
        )
        .map_err(|e| VoiceError::Capture(format!("build stream: {e}")))?;

    stream
        .play()
        .map_err(|e| VoiceError::Capture(format!("play stream: {e}")))?;
    tracing::info!("[voice] capture started");

    while running.load(Ordering::Relaxed) {
        std::thread::sleep(std::time::Duration::from_millis(100));
    }

    drop(stream);
    tracing::info!("[voice] capture stopped");
    Ok(())
}

/// Classify a finalized speech segment.
///
/// Layered: SpeakerPipeline (enrollment-based) wins when it can ID
/// the speaker as the enrolled user; otherwise the diarizer
/// (clustering-based, requires `diarization` feature) provides the
/// `OtherSpeaker { local_id: "speaker_N" }` tag; otherwise Unknown.
///
/// Defined as a free function so the cfg-gated diarizer reference
/// stays out of the spawned task's argument list — feature-gated
/// closure captures get noisy fast.
fn classify_segment(
    #[cfg(feature = "diarization")] diarizer: Option<&crate::diarization::SharedDiarizer>,
    pipeline: Option<&std::sync::Arc<crate::enrollment::SpeakerPipeline>>,
    samples: &[f32],
    sample_rate: u32,
) -> TranscriptRole {
    // Cast once if anyone needs i16.
    let pcm_i16: Option<Vec<i16>> = if pipeline.is_some() || cfg!(feature = "diarization") {
        Some(
            samples
                .iter()
                .map(|s| (s.clamp(-1.0, 1.0) * i16::MAX as f32) as i16)
                .collect(),
        )
    } else {
        None
    };
    if let (Some(pipe), Some(pcm)) = (pipeline, pcm_i16.as_ref()) {
        let role = pipe.classify(pcm);
        if matches!(role, TranscriptRole::EnrolledUser) {
            return role;
        }
        // If the pipeline says OtherSpeaker, fall through to the
        // diarizer for a more granular cluster id when one's
        // available; otherwise return what the pipeline said.
    }
    #[cfg(feature = "diarization")]
    {
        if let (Some(diarizer), Some(pcm)) = (diarizer, pcm_i16.as_ref()) {
            match diarizer.classify(pcm, sample_rate) {
                Ok(role) => return role,
                Err(e) => {
                    tracing::warn!("[voice] diarizer failed: {} — defaulting to Unknown", e);
                }
            }
        }
    }
    // Last fallback: pipeline's verdict (may be Unknown or OtherSpeaker).
    if let (Some(pipe), Some(pcm)) = (pipeline, pcm_i16.as_ref()) {
        return pipe.classify(pcm);
    }
    let _ = (samples, sample_rate);
    TranscriptRole::Unknown
}

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

    #[test]
    fn downmix_passes_mono_through() {
        let mono = vec![0.1, -0.2, 0.3, -0.4];
        assert_eq!(downmix_to_mono(&mono, 1), mono);
    }

    #[test]
    fn downmix_averages_stereo_frames() {
        // [L0, R0, L1, R1] → [(L0+R0)/2, (L1+R1)/2]
        let stereo = vec![1.0, -1.0, 0.5, 0.5];
        assert_eq!(downmix_to_mono(&stereo, 2), vec![0.0, 0.5]);
    }

    #[test]
    fn downmix_drops_partial_trailing_frame() {
        // 5 samples in a 2-channel stream → 2 complete frames, last
        // sample dropped. cpal is contractually supposed to never do
        // this; the test pins the behavior so a future caller
        // doesn't get a panic from chunks_exact's silent skip.
        let bad = vec![1.0, 1.0, 0.5, 0.5, 9.0];
        assert_eq!(downmix_to_mono(&bad, 2), vec![1.0, 0.5]);
    }

    #[test]
    fn linear_resample_no_op_on_matching_rates() {
        let samples = vec![0.1, 0.2, 0.3, 0.4];
        assert_eq!(linear_resample(&samples, 16_000, 16_000), samples);
    }

    #[test]
    fn linear_resample_downsamples_to_expected_length() {
        // 44_100 → 16_000 ratio ≈ 0.36281…
        let in_len = 4410;
        let samples = vec![0.0_f32; in_len];
        let out = linear_resample(&samples, 44_100, 16_000);
        // ratio * in_len = 1600.000…  → round to 1600.
        assert_eq!(out.len(), 1600);
    }

    #[test]
    fn linear_resample_preserves_dc_signal() {
        // Constant input should produce a constant output of the
        // same value (linear interp between equal endpoints is the
        // value itself).
        let samples = vec![0.42_f32; 4410];
        let out = linear_resample(&samples, 44_100, 16_000);
        for s in &out {
            assert!((s - 0.42).abs() < 1e-6, "expected 0.42, got {s}");
        }
    }
}
