//! macOS `VoiceProcessingIO` listener — mic capture with hardware AEC.
//!
//! `cpal` opens the default `RemoteIO` audio unit which gives raw mic
//! samples with no processing. That's a problem when Tokhn's bed and
//! TTS are playing through the speakers: the mic picks up the speaker
//! audio, the user's voice is mixed with it, and STT transcribes the
//! mixture instead of just the user.
//!
//! `VoiceProcessingIO` is the same audio unit FaceTime / Zoom / Discord
//! use. It provides:
//!
//! - Acoustic Echo Cancellation (AEC) — subtracts the system's known
//!   speaker output from the mic input
//! - Noise Suppression
//! - Automatic Gain Control
//!
//! With VPIO active, Tokhn can listen to the user *while the bed is
//! still playing through the speakers* — which is the cinematic premise.
//! No muting, no ducking-to-zero, no headphone requirement.
//!
//! This module is macOS-only. Other platforms keep `CpalListener` as
//! the default.
//!
//! ## Architecture
//!
//! The audio unit's input render callback is a sync C function called
//! from a CoreAudio realtime thread. We can't do anything async or
//! allocate-heavy from inside it. The pattern:
//!
//!   CoreAudio realtime thread
//!     → render callback fills a Float32 mono buffer
//!     → bounded crossbeam_channel push (try_send, drop on full)
//!   Worker std::thread
//!     → drains the channel, runs VAD, finalizes segments
//!     → pushes audio segments into a tokio mpsc
//!   Tokio STT worker
//!     → calls configured STT provider, forwards transcripts via VoiceEvent

#![cfg(target_os = "macos")]

use crate::enrollment::{EnrollmentOutcome, SpeakerPipeline, TaggedTurn, TranscriptRole, TurnRole};
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::voice_audio_mixer::VoiceAudioMixer;
use crate::{Listener, Result, VoiceConfig, VoiceError, VoiceEvent};
use async_trait::async_trait;
use std::os::raw::c_void;
use std::ptr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc;

use coreaudio_sys::{
    kAUVoiceIOProperty_BypassVoiceProcessing, kAUVoiceIOProperty_VoiceProcessingEnableAGC,
    kAudioOutputUnitProperty_EnableIO, kAudioOutputUnitProperty_SetInputCallback,
    kAudioUnitManufacturer_Apple, kAudioUnitProperty_SetRenderCallback,
    kAudioUnitProperty_StreamFormat, kAudioUnitScope_Global, kAudioUnitScope_Input,
    kAudioUnitScope_Output, kAudioUnitSubType_VoiceProcessingIO, kAudioUnitType_Output,
    AURenderCallbackStruct, AudioBuffer, AudioBufferList, AudioComponent,
    AudioComponentDescription, AudioComponentFindNext, AudioComponentInstance,
    AudioComponentInstanceDispose, AudioComponentInstanceNew, AudioOutputUnitStart,
    AudioOutputUnitStop, AudioStreamBasicDescription, AudioTimeStamp, AudioUnitGetProperty,
    AudioUnitInitialize, AudioUnitRender, AudioUnitRenderActionFlags, AudioUnitSetProperty,
    AudioUnitUninitialize, OSStatus,
};

// VPIO is fussy about formats: setting an unsupported sample rate
// fails AudioUnitInitialize with kAudioUnitErr_FormatNotSupported
// (-10875). We don't set a format on either bus — VPIO picks its
// hardware default — and we query both back after init so the worker
// knows the input rate and the output mixer knows the rate/channels
// to render at.
const INPUT_BUS: u32 = 1;
const OUTPUT_BUS: u32 = 0;

const EVENT_CHANNEL_CAPACITY: usize = 64;
const AUDIO_CHANNEL_CAPACITY: usize = 8;
const SAMPLES_CHANNEL_CAPACITY: usize = 64; // ~one second of 20ms chunks
const MIN_SEGMENT_MS: u64 = 300;
// Other tunables (max segment length, SNR floor, barge-in boost,
// post-TTS suppression tail) live on `VoiceConfig` so they can be
// overridden per environment without rebuilding.

/// Default macOS listener using `VoiceProcessingIO` for hardware AEC.
///
/// VPIO routes audio in two directions:
/// - **Input bus 1**: mic samples (with AEC + NS + AGC applied)
/// - **Output bus 0**: speaker output that AEC uses as the reference
///   signal — the audio it subtracts from the mic input
///
/// For AEC to actually cancel Tokhn's bed and TTS, those have to flow
/// through bus 0. The listener takes a [`VoiceAudioMixer`] (built by
/// the caller, knows how to load + mix bed stems and TTS) and the
/// VPIO output render callback pulls samples from it each tick.
///
/// **Build the mixer with `mixer_factory` set** to opt into routing.
/// If you don't, the listener still works (input only) but AEC has
/// nothing to cancel — the bed will bleed through.
pub struct VoiceProcessingListener {
    running: Arc<AtomicBool>,
    capture_thread: Option<std::thread::JoinHandle<()>>,
    stt_task: Option<tokio::task::JoinHandle<()>>,
    mixer_factory: Option<MixerFactory>,
    /// Optional shared flag the capture worker uses to suppress mic
    /// processing while Tokhn is speaking. Cloned from
    /// [`crate::VoiceMixerHandle::speaking_flag`] by the caller.
    speaking_flag: Option<Arc<AtomicBool>>,
    /// Optional speaker-attribution pipeline. When set, each finalized
    /// segment gets classified (or, if enrollment is armed, consumed
    /// as a voiceprint) before STT sees it.
    speaker_pipeline: Option<Arc<SpeakerPipeline>>,
    /// Optional Apple-Speech provider for streaming partials. When set,
    /// the STT worker bypasses the configured `SttProvider` and uses
    /// `transcribe_streaming` to emit `VoiceEvent::Partial` snapshots
    /// per finalized segment, plus the canonical `Transcript`.
    apple_speech_streaming: Option<Arc<crate::apple_speech_stt::AppleSpeechSttProvider>>,
}

/// Closure that builds a [`VoiceAudioMixer`] given the actual sample
/// rate / channel count VPIO chose for its output bus. Called inside
/// `start()` after the format has been queried.
type MixerFactory = Box<dyn FnOnce(u32, u16) -> Result<VoiceAudioMixer> + Send>;

impl VoiceProcessingListener {
    pub fn new() -> Self {
        Self {
            running: Arc::new(AtomicBool::new(false)),
            capture_thread: None,
            stt_task: None,
            mixer_factory: None,
            speaking_flag: None,
            speaker_pipeline: None,
            apple_speech_streaming: None,
        }
    }

    /// Set the audio mixer factory. The listener calls this once it
    /// knows VPIO's output sample rate + channel count, then routes the
    /// resulting mixer through bus 0 so AEC actually has a reference.
    pub fn with_mixer_factory(mut self, factory: MixerFactory) -> Self {
        self.mixer_factory = Some(factory);
        self
    }

    /// Provide the mixer's `speaking_now` flag so the capture worker
    /// can suppress mic-to-STT processing while Tokhn is speaking.
    /// **This is the architectural fix that replaces the post-hoc
    /// echo-stripping heuristics in the GUI gate**: if the listener
    /// never emits a segment captured during own playback, there's
    /// nothing for downstream layers to filter out.
    pub fn with_speaking_flag(mut self, flag: Arc<AtomicBool>) -> Self {
        self.speaking_flag = Some(flag);
        self
    }

    /// Attach a [`SpeakerPipeline`]. The listener routes every
    /// finalized segment through it: if enrollment is armed, the
    /// segment is consumed as a voiceprint (emitting
    /// [`VoiceEvent::EnrollmentCaptured`] / [`VoiceEvent::EnrollmentFailed`])
    /// and skipped for STT; otherwise the pipeline classifies it and
    /// the resulting [`TranscriptRole`] rides along on the
    /// [`VoiceEvent::Transcript`] the STT worker emits.
    pub fn with_speaker_pipeline(mut self, pipeline: Arc<SpeakerPipeline>) -> Self {
        self.speaker_pipeline = Some(pipeline);
        self
    }

    /// Use Apple's `SFSpeechRecognizer` for streaming transcription on
    /// macOS. Per finalized segment the STT worker emits
    /// `VoiceEvent::Partial` per snapshot plus a final
    /// `VoiceEvent::Transcript`. Bypasses the configured `SttProvider`
    /// for transcription; speaker-attribution still runs.
    pub fn with_apple_speech_streaming(
        mut self,
        provider: Arc<crate::apple_speech_stt::AppleSpeechSttProvider>,
    ) -> Self {
        self.apple_speech_streaming = Some(provider);
        self
    }
}

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

impl Drop for VoiceProcessingListener {
    fn drop(&mut self) {
        if self.running.load(Ordering::SeqCst) {
            tracing::warn!("[voice] VPIO listener dropped while running; stopping");
            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 VoiceProcessingListener {
    async fn start(&mut self, config: VoiceConfig) -> Result<mpsc::Receiver<VoiceEvent>> {
        if self.running.load(Ordering::SeqCst) {
            return Err(VoiceError::AlreadyRunning);
        }

        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. Handles all pipeline interactions on this side
        // of the capture/STT boundary so the VPIO capture thread stays
        // lean: enrollment capture (skip STT, emit Captured/Failed),
        // then per-turn diarization so multi-speaker segments emit one
        // Transcript per turn with the correct role instead of a single
        // misattributed event.
        let event_tx_for_stt = event_tx.clone();
        let pipeline_for_stt = self.speaker_pipeline.clone();
        let apple_speech_for_stt = self.apple_speech_streaming.clone();
        // Long-lived runtime for the consumer; see the matching note
        // in `cpal_listener.rs` and #120's root cause.
        let stt_handle = crate::voice_runtime_handle().spawn(async move {
            tracing::info!("[voice] VPIO STT consumer task started");
            while let Some(segment) = audio_rx.recv().await {
                // 1. Enrollment-capture path. If pending, consume the
                //    whole segment as a voiceprint and skip STT — a
                //    legit enrollment utterance is not a command.
                if let Some(pipeline) = pipeline_for_stt.as_ref() {
                    if pipeline.has_pending_enrollment() {
                        let pcm_i16 = f32_to_i16(&segment.samples);
                        if let Some(outcome) = pipeline.capture_enrollment(&pcm_i16) {
                            emit_enrollment_outcome(&event_tx_for_stt, outcome).await;
                            continue;
                        }
                    }
                }

                // 2. Normal path. Classify into per-speaker turns
                //    when an enrollment is available; otherwise treat
                //    the full segment as one Unknown turn so the
                //    pre-enrollment flow keeps working unchanged.
                let turns = build_stt_turns(pipeline_for_stt.as_ref(), &segment);
                for turn in turns {
                    if turn.audio.is_empty() {
                        continue;
                    }
                    // Emit a Transcript event for every transcribe
                    // result, including empty — see the matching note
                    // in `cpal_listener.rs`. Without this, callers
                    // can't distinguish "consumer never ran" from
                    // "consumer ran but whisper filtered every
                    // segment as no-speech."
                    // Apple-Speech streaming path takes precedence when
                    // configured — emit Partial per snapshot, then a
                    // canonical Transcript at end-of-segment. Falls
                    // through to the configured batch SttProvider if
                    // not set or if streaming returns an error.
                    if let Some(apple) = apple_speech_for_stt.as_ref() {
                        let event_tx_partials = event_tx_for_stt.clone();
                        let duration_ms = turn.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(&turn.audio, segment.sample_rate, on_partial)
                            .await
                        {
                            // Drop empty results on the streaming path. The
                            // Partial events already prove the consumer ran,
                            // so empty Transcripts add UI noise without
                            // diagnostic value (matches CpalListener's
                            // streaming arm; differs from VPIO's batch path
                            // which emits empty Transcripts intentionally —
                            // see #120 root cause).
                            Ok(text) if !text.is_empty() => {
                                let _ = event_tx_for_stt
                                    .send(VoiceEvent::Transcript {
                                        text,
                                        duration_ms: turn.duration_ms,
                                        role: turn.role.clone(),
                                    })
                                    .await;
                                continue;
                            }
                            Ok(_) => {
                                tracing::debug!("[voice] empty apple-speech transcript");
                                continue;
                            }
                            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: turn.duration_ms,
                                        role: turn.role.clone(),
                                    })
                                    .await;
                                continue;
                            }
                        }
                    }
                    match stt_provider
                        .transcribe(&turn.audio, segment.sample_rate)
                        .await
                    {
                        Ok(text) => {
                            let _ = event_tx_for_stt
                                .send(VoiceEvent::Transcript {
                                    text,
                                    duration_ms: turn.duration_ms,
                                    role: turn.role.clone(),
                                })
                                .await;
                        }
                        Err(e) => {
                            tracing::error!("[voice] STT failed: {e}");
                            let _ = event_tx_for_stt
                                .send(VoiceEvent::Transcript {
                                    text: format!("[stt-error] {e}"),
                                    duration_ms: turn.duration_ms,
                                    role: turn.role.clone(),
                                })
                                .await;
                        }
                    }
                }
            }
        });
        self.stt_task = Some(stt_handle);

        // Capture thread — owns the VPIO audio unit and the worker
        // loop that turns raw samples into VAD events + segments.
        // Errors that happen inside the audio unit setup are propagated
        // back to the consumer via the [stt-error] hack on a Transcript
        // event so the badge surfaces them.
        let running = self.running.clone();
        let cfg = config.clone();
        let event_tx_for_errors = event_tx.clone();
        let mixer_factory = self.mixer_factory.take();
        let speaking_flag = self.speaking_flag.clone();
        let capture_handle = std::thread::Builder::new()
            .name("vpio-capture".into())
            .spawn(move || {
                if let Err(e) = run_vpio_capture(
                    running.clone(),
                    cfg,
                    event_tx,
                    audio_tx,
                    mixer_factory,
                    speaking_flag,
                ) {
                    tracing::error!("[voice] VPIO capture thread error: {e}");
                    let _ = event_tx_for_errors.blocking_send(VoiceEvent::Transcript {
                        text: format!("[stt-error] VPIO: {e}"),
                        duration_ms: 0,
                        role: TranscriptRole::Unknown,
                    });
                    running.store(false, Ordering::SeqCst);
                }
            })
            .map_err(|e| VoiceError::Capture(format!("spawn vpio thread: {e}")))?;
        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);
        }
        if let Some(task) = self.stt_task.take() {
            task.abort();
        }
        if let Some(handle) = self.capture_thread.take() {
            tokio::task::spawn_blocking(move || {
                if let Err(e) = handle.join() {
                    tracing::warn!("[voice] VPIO 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)
    }
}

struct AudioSegment {
    samples: Vec<f32>,
    sample_rate: u32,
    duration_ms: u64,
}

/// A segment or per-speaker turn ready for STT. Built by
/// [`build_stt_turns`] from an `AudioSegment` plus (optionally) a
/// diarized classification — same shape for both the "no pipeline"
/// fast path and the "diarize then STT each turn" path so the STT
/// worker loop stays uniform.
struct SttTurn {
    audio: Vec<f32>,
    duration_ms: u64,
    role: TranscriptRole,
}

/// Emit the right `VoiceEvent` for an enrollment outcome. Separated
/// so the STT worker's main loop stays readable.
async fn emit_enrollment_outcome(tx: &mpsc::Sender<VoiceEvent>, outcome: EnrollmentOutcome) {
    match outcome {
        EnrollmentOutcome::Captured { label, save_path } => {
            tracing::info!(
                "[voice] enrollment captured for '{label}' -> {}",
                save_path.display()
            );
            let _ = tx
                .send(VoiceEvent::EnrollmentCaptured { label, save_path })
                .await;
        }
        EnrollmentOutcome::Failed { reason } => {
            tracing::warn!("[voice] enrollment failed: {reason}");
            let _ = tx.send(VoiceEvent::EnrollmentFailed { reason }).await;
        }
    }
}

/// Convert a finalized segment into the list of STT-ready turns.
///
/// * No pipeline or no enrollment → one turn covering the full segment
///   tagged `TranscriptRole::Unknown`. Preserves the pre-enrollment
///   flow (bootstrapping, reflex commands, everything before the user
///   captures a voiceprint) bit-for-bit.
/// * Pipeline with enrollment → one turn per diarized speaker, each
///   tagged with the per-turn role so multi-speaker segments emit
///   multiple Transcript events instead of one misattributed blob.
fn build_stt_turns(
    pipeline: Option<&Arc<SpeakerPipeline>>,
    segment: &AudioSegment,
) -> Vec<SttTurn> {
    let fallback = || {
        vec![SttTurn {
            audio: segment.samples.clone(),
            duration_ms: segment.duration_ms,
            role: TranscriptRole::Unknown,
        }]
    };
    let Some(pipeline) = pipeline else {
        return fallback();
    };
    match pipeline.classify_turns(&segment.samples, segment.sample_rate) {
        Ok(turns) if !turns.is_empty() => turns
            .into_iter()
            .map(
                |TaggedTurn {
                     start_ms,
                     end_ms,
                     role,
                     audio,
                     ..
                 }| {
                    let duration_ms = end_ms.saturating_sub(start_ms);
                    let role = match role {
                        TurnRole::EnrolledUser => TranscriptRole::EnrolledUser,
                        TurnRole::OtherSpeaker(id) => TranscriptRole::OtherSpeaker { local_id: id },
                        TurnRole::Unknown => TranscriptRole::Unknown,
                    };
                    SttTurn {
                        audio,
                        duration_ms,
                        role,
                    }
                },
            )
            .collect(),
        Ok(_) => fallback(),
        Err(e) => {
            tracing::warn!("[voice] diarization failed, falling back to whole-segment: {e}");
            fallback()
        }
    }
}

/// Convert normalized `[-1.0, 1.0]` f32 samples to 16-bit PCM so they
/// can be fed to a [`SpeakerEmbedder`] whose contract is `&[i16]`.
/// Clamps out-of-range values (rare — AGC keeps things below unity).
fn f32_to_i16(samples: &[f32]) -> Vec<i16> {
    samples
        .iter()
        .map(|&s| (s.clamp(-1.0, 1.0) * 32767.0).round() as i16)
        .collect()
}

/// Compute the segment's average RMS in dB and verify it sits at
/// least `min_snr_db` above the calibrated noise floor. Cheap
/// last-mile filter — STT only sees segments with plausible
/// speaker energy.
fn segment_passes_snr_floor(samples: &[f32], noise_floor_db: f32, min_snr_db: f32) -> bool {
    if samples.is_empty() {
        return false;
    }
    let mut sum_sq: f32 = 0.0;
    for &s in samples {
        let scaled = s * 32768.0;
        sum_sq += scaled * scaled;
    }
    let rms = (sum_sq / samples.len() as f32).sqrt();
    let segment_db = if rms < 1.0 {
        -96.0
    } else {
        20.0 * (rms / 32768.0).log10()
    };
    let snr = segment_db - noise_floor_db;
    if snr < min_snr_db {
        tracing::debug!(
            "[voice] segment SNR {:.1}dB (segment={:.1}dB floor={:.1}dB)",
            snr,
            segment_db,
            noise_floor_db
        );
        return false;
    }
    true
}

// ─────────────────────────────────────────────────────────────────────
// Realtime callback context
// ─────────────────────────────────────────────────────────────────────

/// Shared state passed to the audio render callback. Lives for the
/// duration of the capture session. The pointer is stable because we
/// box it and never move it.
struct CallbackContext {
    audio_unit: AudioComponentInstance,
    samples_tx: crossbeam_channel::Sender<Vec<f32>>,
    running: Arc<AtomicBool>,
    /// Number of channels VPIO is delivering on the input bus. We
    /// pre-mix to mono before pushing samples to the worker.
    input_channels: u32,
    /// The audio mixer that provides samples for the OUTPUT render
    /// callback. `None` means we don't render output (AEC won't have a
    /// reference, but the unit still captures input).
    ///
    /// **No Mutex.** The mixer is touched by exactly one thread —
    /// the CoreAudio realtime callback — so wrapping it in a lock
    /// would only add a chance of stalling that thread, which is
    /// the worst possible thing for AEC reference signal continuity.
    /// Bevy talks to the mixer through atomics + a crossbeam channel
    /// inside [`crate::VoiceMixerHandle`], not directly.
    output_mixer: Option<VoiceAudioMixer>,
    /// Output channel count from VPIO's negotiated format. The output
    /// callback fills `frames * output_channels` interleaved samples.
    output_channels: u32,
    /// Reusable scratch buffer for the output callback so we never
    /// allocate on the realtime thread. Resized lazily on first call
    /// to match the actual frame size CoreAudio asks for, then reused
    /// for the lifetime of the audio unit.
    output_scratch: Vec<f32>,
    /// Reusable VPIO input buffer. VPIO writes interleaved mic samples
    /// into this on every render callback. Resized lazily on the first
    /// callback and reused for the audio unit's lifetime, so the input
    /// callback doesn't allocate on the realtime thread. (The Vec we
    /// send to the worker is still a fresh allocation because it moves
    /// ownership across threads — see the copy path in
    /// [`input_callback`].)
    input_scratch: Vec<f32>,
}

// ─────────────────────────────────────────────────────────────────────
// Capture loop
// ─────────────────────────────────────────────────────────────────────

fn run_vpio_capture(
    running: Arc<AtomicBool>,
    config: VoiceConfig,
    event_tx: mpsc::Sender<VoiceEvent>,
    audio_tx: mpsc::Sender<AudioSegment>,
    mixer_factory: Option<MixerFactory>,
    speaking_flag: Option<Arc<AtomicBool>>,
) -> Result<()> {
    // Sample channel: realtime callback → worker thread.
    let (samples_tx, samples_rx) = crossbeam_channel::bounded::<Vec<f32>>(SAMPLES_CHANNEL_CAPACITY);

    // Build the VPIO audio unit. Boxed so the callback context pointer
    // is stable across the audio unit's lifetime.
    let mut ctx = Box::new(CallbackContext {
        audio_unit: ptr::null_mut(),
        samples_tx,
        running: running.clone(),
        input_channels: 1,
        output_mixer: None,
        output_channels: 1,
        output_scratch: Vec::new(),
        input_scratch: Vec::new(),
    });

    let (input_format, output_format) =
        unsafe { open_voice_processing_unit(&mut *ctx, mixer_factory)? };
    let sample_rate = input_format.mSampleRate as u32;
    ctx.input_channels = input_format.mChannelsPerFrame;
    ctx.output_channels = output_format.mChannelsPerFrame;

    tracing::info!(
        "[voice] VPIO capture started (in: {} Hz {} ch, out: {} Hz {} ch)",
        sample_rate,
        input_format.mChannelsPerFrame,
        output_format.mSampleRate as u32,
        output_format.mChannelsPerFrame
    );

    // The audio unit is now pumping samples into samples_rx via the
    // realtime callback. The worker loop drains samples, runs VAD,
    // and dispatches events / segments through the same channels
    // CpalListener uses.
    //
    // **VAD backend selection.** Prefer Silero VAD v6 through MLX on
    // Apple Silicon. If that model or runtime is unavailable, fall
    // back to the bundled Silero V5 ONNX path and then to the original
    // RMS energy VAD so the listener still works in degraded mode.
    //
    // The Silero crate (`voice_activity_detector 0.2.1`) currently
    // unwraps an `ort` call internally (`vad.rs:69`) that hits
    // `NotImplemented: GetElementType` on the bundled `ort 2.0.0-rc.10`
    // — an upstream version mismatch we don't control. The unwrap
    // would kill this whole `vpio-capture` thread without surfacing
    // an Err. Wrap the construction in `catch_unwind` so the fallback
    // branch actually runs instead of the listener silently dying.
    #[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 = 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) as Box<dyn VadBackend>)
        }
        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 mut vad: 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 mut recording = Vec::with_capacity(sample_rate as usize * 4);
    let mut is_recording = false;

    // Barge-in friendly TTS gating: while Tokhn is actively speaking
    // we *raise the VAD threshold* by ~18 dB instead of dropping
    // chunks. That way only user voice substantially louder than the
    // bed/TTS leakage will register as "speech" — exactly the
    // signature of someone trying to interrupt — while passive room
    // sound and echo of Tokhn's own voice stay below the bar.
    //
    // When the elevated VAD does fire during playback, we treat it
    // as a barge-in: stop the active TTS via `mixer.stop_tts()`
    // (which the realtime callback honors next pull), drop the
    // threshold boost, and let the resulting segment proceed to STT
    // normally so the user's interrupt becomes their next command.
    //
    // The tail covers the few hundred ms after `speaking_now` goes
    // false during which the speaker is still settling and the room
    // reverb has audible energy.
    let barge_in_boost_db = config.barge_in_boost_db;
    let boost_tail_ms = config.boost_tail_ms as u128;
    let max_segment_ms = config.max_segment_ms;
    let segment_min_snr_db = config.segment_min_snr_db;
    let mut tts_ended_at: Option<std::time::Instant> = None;
    let mut boost_applied = false;
    let mut barge_in_pending = false;

    while running.load(Ordering::Relaxed) {
        // Block briefly so we don't spin.
        let chunk = match samples_rx.recv_timeout(std::time::Duration::from_millis(50)) {
            Ok(c) => c,
            Err(crossbeam_channel::RecvTimeoutError::Timeout) => continue,
            Err(crossbeam_channel::RecvTimeoutError::Disconnected) => break,
        };

        // ─── Barge-in threshold management ──────────────────────
        let is_speaking = speaking_flag
            .as_ref()
            .map(|f| f.load(Ordering::Relaxed))
            .unwrap_or(false);

        if is_speaking {
            tts_ended_at = Some(std::time::Instant::now());
            if !boost_applied {
                vad.set_threshold_boost(barge_in_boost_db);
                boost_applied = true;
                tracing::info!(
                    "[voice] vad threshold boosted +{:.0}dB (Tokhn speaking)",
                    barge_in_boost_db
                );
            }
        } else if let Some(end) = tts_ended_at {
            if end.elapsed().as_millis() < boost_tail_ms {
                // Still inside the tail window — keep the boost.
            } else {
                tts_ended_at = None;
                if boost_applied {
                    vad.set_threshold_boost(0.0);
                    boost_applied = false;
                    tracing::info!("[voice] vad threshold restored");
                }
            }
        }

        // Hard-suppression window: even if the VAD-with-boost still
        // fires (residual speaker tail, room reverb, or an unlucky
        // transient that clips the boosted threshold), do NOT start a
        // NEW recording while Tokhn is speaking or in the post-TTS
        // tail. An already-in-progress recording keeps going — that's
        // the legit barge-in path. This is the gate that prevents
        // "Tokhn's own intro gets transcribed as a user utterance" at
        // the listener layer; the transcript echo filter downstream is
        // the second line of defense.
        let in_tail = tts_ended_at
            .map(|end| end.elapsed().as_millis() < boost_tail_ms)
            .unwrap_or(false);
        let suppress_new_recording = is_speaking || in_tail;
        // ────────────────────────────────────────────────────────

        vad.process_samples(&chunk);
        let speech = vad.is_speech_active();

        // Detect barge-in: VAD fired AND Tokhn is speaking.
        if speech && is_speaking && !barge_in_pending {
            barge_in_pending = true;
            tracing::info!("[voice] barge-in detected — halting TTS");
            // Tell whoever owns the mixer to halt the active clip.
            // We can't reach the mixer directly from this thread, but
            // the mixer's `stop_tts_flag` is wired through the same
            // `speaking_flag` Arc — see VoiceMixerHandle::stop_tts.
            // The capture loop just emits a BargeIn voice event and
            // the bridge calls stop_tts() on the handle.
            if event_tx.try_send(VoiceEvent::BargeIn).is_err() {
                tracing::warn!("[voice] dropped BargeIn (consumer slow)");
            }
        }
        if !is_speaking {
            barge_in_pending = false;
        }

        if speech && !is_recording && !suppress_new_recording {
            tracing::debug!("[voice] speech detected (VPIO), recording...");
            is_recording = true;
            recording.clear();
            recording.extend_from_slice(&chunk);
            if event_tx.try_send(VoiceEvent::SpeechStart).is_err() {
                tracing::warn!("[voice] dropped SpeechStart (consumer slow)");
            }
        } else if is_recording {
            recording.extend_from_slice(&chunk);

            // Force-end the segment if it has grown past the hard cap.
            // VAD occasionally locks onto background noise and produces
            // 17 second "speech" segments that never naturally end.
            let current_ms = (recording.len() as u64 * 1000) / sample_rate as u64;
            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.turn_ended()) || force_end {
                is_recording = false;
                let audio: Vec<f32> = std::mem::take(&mut recording);
                let duration_ms = (audio.len() as u64 * 1000) / sample_rate as u64;
                tracing::debug!("[voice] turn ended (VPIO), {}ms", duration_ms);
                if event_tx.try_send(VoiceEvent::SpeechEnd).is_err() {
                    tracing::warn!("[voice] dropped SpeechEnd (consumer slow)");
                }
                if force_end {
                    // Reset VAD so we don't immediately start
                    // recording the same noise lock-up again.
                    vad.reset_speech_state();
                }
                if duration_ms >= MIN_SEGMENT_MS
                    && segment_passes_snr_floor(&audio, vad.noise_floor_db(), segment_min_snr_db)
                {
                    // Send raw segment to the STT worker; it owns all
                    // pipeline interaction (enrollment capture + per-
                    // turn diarization) so this thread stays purely
                    // audio-in, segment-out.
                    if let Err(e) = audio_tx.try_send(AudioSegment {
                        samples: audio,
                        sample_rate,
                        duration_ms,
                    }) {
                        tracing::error!(
                            "[voice] dropped {}ms VPIO segment (STT backlog full): {}",
                            duration_ms,
                            e
                        );
                    }
                } else if duration_ms >= MIN_SEGMENT_MS {
                    tracing::info!(
                        "[voice] dropped {}ms segment (below SNR floor: noise={:.1}dB)",
                        duration_ms,
                        vad.noise_floor_db()
                    );
                }
            }
        }
    }

    // Tear down the audio unit cleanly. The Box keeps `ctx` alive for
    // the duration of close_voice_processing_unit so the C callback
    // can't see freed memory.
    unsafe {
        close_voice_processing_unit(ctx.audio_unit);
    }
    drop(ctx);
    tracing::info!("[voice] VPIO capture stopped");
    Ok(())
}

// ─────────────────────────────────────────────────────────────────────
// CoreAudio FFI
// ─────────────────────────────────────────────────────────────────────

unsafe fn open_voice_processing_unit(
    ctx: &mut CallbackContext,
    mixer_factory: Option<MixerFactory>,
) -> Result<(AudioStreamBasicDescription, AudioStreamBasicDescription)> {
    // 1. Find the VPIO component.
    let desc = AudioComponentDescription {
        componentType: kAudioUnitType_Output,
        componentSubType: kAudioUnitSubType_VoiceProcessingIO,
        componentManufacturer: kAudioUnitManufacturer_Apple,
        componentFlags: 0,
        componentFlagsMask: 0,
    };
    let component: AudioComponent = AudioComponentFindNext(ptr::null_mut(), &desc);
    if component.is_null() {
        return Err(VoiceError::Device("VPIO component not found".into()));
    }

    // 2. Instantiate it.
    let mut unit: AudioComponentInstance = ptr::null_mut();
    let status = AudioComponentInstanceNew(component, &mut unit);
    osstatus(status, "AudioComponentInstanceNew")?;
    ctx.audio_unit = unit;

    let want_output = mixer_factory.is_some();

    // 3. Enable input on bus 1.
    let enable: u32 = 1;
    let status = AudioUnitSetProperty(
        unit,
        kAudioOutputUnitProperty_EnableIO,
        kAudioUnitScope_Input,
        INPUT_BUS,
        &enable as *const _ as *const c_void,
        std::mem::size_of::<u32>() as u32,
    );
    osstatus(status, "EnableIO input")?;

    // 4. Enable or disable output on bus 0. We enable it whenever a
    // mixer factory was provided so AEC has a reference signal.
    let output_enable: u32 = if want_output { 1 } else { 0 };
    let status = AudioUnitSetProperty(
        unit,
        kAudioOutputUnitProperty_EnableIO,
        kAudioUnitScope_Output,
        OUTPUT_BUS,
        &output_enable as *const _ as *const c_void,
        std::mem::size_of::<u32>() as u32,
    );
    osstatus(
        status,
        if want_output {
            "EnableIO output (enable)"
        } else {
            "EnableIO output (disable)"
        },
    )?;

    // 5. Install the input render callback.
    let in_cb = AURenderCallbackStruct {
        inputProc: Some(input_callback),
        inputProcRefCon: ctx as *mut _ as *mut c_void,
    };
    let status = AudioUnitSetProperty(
        unit,
        kAudioOutputUnitProperty_SetInputCallback,
        kAudioUnitScope_Global,
        0,
        &in_cb as *const _ as *const c_void,
        std::mem::size_of::<AURenderCallbackStruct>() as u32,
    );
    osstatus(status, "SetInputCallback")?;

    // 6. Install the output render callback (for the input scope of
    // the output bus — that's the data source side).
    if want_output {
        let out_cb = AURenderCallbackStruct {
            inputProc: Some(output_callback),
            inputProcRefCon: ctx as *mut _ as *mut c_void,
        };
        let status = AudioUnitSetProperty(
            unit,
            kAudioUnitProperty_SetRenderCallback,
            kAudioUnitScope_Input,
            OUTPUT_BUS,
            &out_cb as *const _ as *const c_void,
            std::mem::size_of::<AURenderCallbackStruct>() as u32,
        );
        osstatus(status, "SetRenderCallback (output)")?;
    }

    // 7. Explicitly opt INTO voice processing. On some Macs the
    // default for `kAUVoiceIOProperty_BypassVoiceProcessing` is 1
    // (bypass enabled, no AEC) — we have to set it to 0 to actually
    // get echo cancellation. Same with AGC. Both must be set BEFORE
    // AudioUnitInitialize.
    if want_output {
        let bypass: u32 = 0;
        let status = AudioUnitSetProperty(
            unit,
            kAUVoiceIOProperty_BypassVoiceProcessing,
            kAudioUnitScope_Global,
            0,
            &bypass as *const _ as *const c_void,
            std::mem::size_of::<u32>() as u32,
        );
        if status != 0 {
            tracing::warn!(
                "[voice] BypassVoiceProcessing=0 failed: OSStatus={status} (continuing)"
            );
        } else {
            tracing::info!("[voice] BypassVoiceProcessing=0 (AEC enabled)");
        }

        let enable_agc: u32 = 1;
        let status = AudioUnitSetProperty(
            unit,
            kAUVoiceIOProperty_VoiceProcessingEnableAGC,
            kAudioUnitScope_Global,
            0,
            &enable_agc as *const _ as *const c_void,
            std::mem::size_of::<u32>() as u32,
        );
        if status != 0 {
            tracing::warn!(
                "[voice] VoiceProcessingEnableAGC=1 failed: OSStatus={status} (continuing)"
            );
        } else {
            tracing::info!("[voice] VoiceProcessingEnableAGC=1");
        }
    }

    // 8. Initialize. VPIO is fussy about formats — we don't try to
    // override anything. Let it pick its hardware defaults and we'll
    // query them back so the mixer can adapt.
    let status = AudioUnitInitialize(unit);
    osstatus(status, "AudioUnitInitialize")?;

    // 8. Query both formats so we know what VPIO chose.
    let input_format = query_format(unit, kAudioUnitScope_Output, INPUT_BUS)?;
    let output_format = if want_output {
        query_format(unit, kAudioUnitScope_Input, OUTPUT_BUS)?
    } else {
        input_format
    };

    // 9. Log the format flags so we can see exactly what VPIO is
    // expecting on each side. Helps diagnose silent-output issues.
    if want_output {
        tracing::info!(
            "[voice] VPIO output (bus 0 input scope): {} Hz, {} ch, {} bits, flags=0x{:x}, bytes/frame={}, format_id=0x{:x}",
            output_format.mSampleRate as u32,
            output_format.mChannelsPerFrame,
            output_format.mBitsPerChannel,
            output_format.mFormatFlags,
            output_format.mBytesPerFrame,
            output_format.mFormatID,
        );
    }

    // 10. Now that we know the output format, build the mixer.
    if let Some(factory) = mixer_factory {
        let mixer = factory(
            output_format.mSampleRate as u32,
            output_format.mChannelsPerFrame as u16,
        )?;
        ctx.output_mixer = Some(mixer);
    }

    // 11. Start.
    let status = AudioOutputUnitStart(unit);
    osstatus(status, "AudioOutputUnitStart")?;

    Ok((input_format, output_format))
}

unsafe fn query_format(
    unit: AudioComponentInstance,
    scope: u32,
    bus: u32,
) -> Result<AudioStreamBasicDescription> {
    let mut actual = std::mem::zeroed::<AudioStreamBasicDescription>();
    let mut size = std::mem::size_of::<AudioStreamBasicDescription>() as u32;
    let status = AudioUnitGetProperty(
        unit,
        kAudioUnitProperty_StreamFormat,
        scope,
        bus,
        &mut actual as *mut _ as *mut c_void,
        &mut size,
    );
    osstatus(status, "GetProperty StreamFormat")?;
    Ok(actual)
}

unsafe fn close_voice_processing_unit(unit: AudioComponentInstance) {
    if unit.is_null() {
        return;
    }
    let _ = AudioOutputUnitStop(unit);
    let _ = AudioUnitUninitialize(unit);
    let _ = AudioComponentInstanceDispose(unit);
}

fn osstatus(status: OSStatus, op: &str) -> Result<()> {
    if status == 0 {
        Ok(())
    } else {
        Err(VoiceError::Device(format!(
            "{op} failed: OSStatus={status}"
        )))
    }
}

// ─────────────────────────────────────────────────────────────────────
// Realtime input callback
// ─────────────────────────────────────────────────────────────────────

unsafe extern "C" fn input_callback(
    in_ref_con: *mut c_void,
    io_action_flags: *mut AudioUnitRenderActionFlags,
    in_time_stamp: *const AudioTimeStamp,
    in_bus_number: u32,
    in_number_frames: u32,
    _io_data: *mut AudioBufferList,
) -> OSStatus {
    let ctx = &mut *(in_ref_con as *mut CallbackContext);

    let frames = in_number_frames as usize;
    let channels = ctx.input_channels.max(1) as usize;
    let total_samples = frames * channels;
    let bytes = total_samples * std::mem::size_of::<f32>();

    // Reuse the persistent input scratch buffer so VPIO writes into
    // it without allocating on the realtime thread. `resize` only
    // allocates when the frame count grows — normally just once, on
    // the first callback.
    if ctx.input_scratch.len() < total_samples {
        ctx.input_scratch.resize(total_samples, 0.0);
    }
    let interleaved = &mut ctx.input_scratch[..total_samples];

    let mut buffer_list = AudioBufferList {
        mNumberBuffers: 1,
        mBuffers: [AudioBuffer {
            mNumberChannels: ctx.input_channels,
            mDataByteSize: bytes as u32,
            mData: interleaved.as_mut_ptr() as *mut c_void,
        }],
    };

    let status = AudioUnitRender(
        ctx.audio_unit,
        io_action_flags,
        in_time_stamp,
        in_bus_number,
        in_number_frames,
        &mut buffer_list,
    );
    if status != 0 {
        return status;
    }

    // Copy + downmix into a fresh mono Vec. The fresh allocation here
    // is unavoidable — the Vec moves across threads into the capture
    // worker via the channel — but the VPIO buffer allocation (the
    // bigger of the two, and the one sized by channel count) is now
    // reused via `input_scratch`. Net: one heap allocation per
    // callback instead of two, with a small memcpy in its place.
    let mono: Vec<f32> = if channels == 1 {
        let mut out = Vec::with_capacity(frames);
        out.extend_from_slice(interleaved);
        out
    } else {
        let mut out = Vec::with_capacity(frames);
        for frame in interleaved.chunks_exact(channels) {
            let sum: f32 = frame.iter().sum();
            out.push(sum / channels as f32);
        }
        out
    };

    if ctx.running.load(Ordering::Relaxed) {
        let _ = ctx.samples_tx.try_send(mono);
    }

    0
}

/// Output render callback. CoreAudio calls this when it needs more
/// samples to play out of bus 0. We pull from the [`VoiceAudioMixer`]
/// the listener was constructed with — the mixer mixes the bed stems
/// + any active TTS clip and writes interleaved samples into a
/// pre-allocated scratch buffer that we then copy into CoreAudio's
/// destination. AEC sees a clean, gap-free reference signal so it can
/// cancel Tokhn's own voice from the mic input.
///
/// **Realtime contract:**
/// - No allocation. The scratch buffer is grown lazily on the first
///   call to match CoreAudio's frame request and then reused.
/// - No locking. The mixer is owned exclusively by this thread.
/// - No I/O. Stems are pre-decoded into memory by the mixer's
///   constructor; TTS bytes are decoded by the Bevy thread before
///   being sent over the channel.
/// - Always fills every requested sample. Silence is `0.0`, never
///   "no value" — gaps in the AEC reference signal would corrupt the
///   echo path estimator.
unsafe extern "C" fn output_callback(
    in_ref_con: *mut c_void,
    _io_action_flags: *mut AudioUnitRenderActionFlags,
    _in_time_stamp: *const AudioTimeStamp,
    _in_bus_number: u32,
    in_number_frames: u32,
    io_data: *mut AudioBufferList,
) -> OSStatus {
    if io_data.is_null() {
        return 0;
    }
    let ctx = &mut *(in_ref_con as *mut CallbackContext);
    let buffer_list = &mut *io_data;

    let channels = ctx.output_channels.max(1) as usize;
    let frames = in_number_frames as usize;
    let total = frames * channels;

    // Grow the scratch buffer once if needed (typically on the first
    // callback). After that it's reused for the lifetime of the unit.
    if ctx.output_scratch.len() < total {
        ctx.output_scratch.resize(total, 0.0);
    }
    let scratch = &mut ctx.output_scratch[..total];

    // Pull mixed samples directly from the mixer into the scratch
    // buffer. No lock — single-threaded ownership of `output_mixer`.
    if let Some(mixer) = ctx.output_mixer.as_mut() {
        mixer.pull(scratch);
    } else {
        for s in scratch.iter_mut() {
            *s = 0.0;
        }
    }

    // CoreAudio AudioBufferList may be interleaved (1 buffer with N
    // channels) or non-interleaved (N buffers with 1 channel each).
    // For VPIO output the typical layout is interleaved single buffer.
    if buffer_list.mNumberBuffers == 1 {
        let buffer = &mut buffer_list.mBuffers[0];
        let dst = buffer.mData as *mut f32;
        if !dst.is_null() {
            std::ptr::copy_nonoverlapping(scratch.as_ptr(), dst, total);
        }
    } else {
        // Non-interleaved: each buffer holds one channel of `frames` samples.
        for ch in 0..buffer_list.mNumberBuffers as usize {
            // The Rust binding only exposes mBuffers as a single-entry
            // array; we index past it manually for non-interleaved
            // layouts. CoreAudio guarantees the layout is contiguous.
            let buf_ptr = (&mut buffer_list.mBuffers[0] as *mut AudioBuffer).add(ch);
            let buf = &mut *buf_ptr;
            let dst = buf.mData as *mut f32;
            if dst.is_null() {
                continue;
            }
            for f in 0..frames {
                let src_idx = f * channels + ch.min(channels - 1);
                *dst.add(f) = scratch[src_idx];
            }
        }
    }

    0
}
