audio-io-bsd 0.2.0

//! [`OssBackend`] — a FreeBSD-native direct OSS [`AudioBackend`].
//!
//! Reaches the kernel audio layer through OSS (`/dev/dsp`, `sound(4)` /
//! `dsp(4)`) via libc raw `ioctl` calls — **no cpal, no ALSA indirection**. On FreeBSD OSS
//! is the native modern audio API; the `alsa-lib` port is only a Linux-software
//! compatibility shim. This backend is gated to `target_os = "freebsd"` and the
//! `oss` feature.
//!
//! # Real-time safety model
//!
//! Identical to the cpal backend: [`OssSink::write`] only **pushes** interleaved
//! `f32` samples into a lock-free [`rtrb`] ring (wait-free, alloc-free). A
//! dedicated audio thread owns the `/dev/dsp` file descriptor, drains the ring,
//! converts `f32` to the negotiated PCM format on a reused scratch buffer, and
//! performs the blocking `write(2)` to the device. The RT thread never touches
//! the file descriptor or performs I/O.
//!
//! # Format negotiation
//!
//! FreeBSD drivers vary in their support for `AFMT_FLOAT`. [`OssBackend`]
//! probes the device with `SNDCTL_DSP_GETFMTS` and negotiates
//! `AFMT_FLOAT → AFMT_S32_LE → AFMT_S16_LE`, confirming each with a readback.
//! The `f32` ↔ integer conversion is performed by [`crate::sample_conv`].
//!
//! > **Hardware note:** the acceptance criteria (sustained playback with xrun 0)
//! > require a `/dev/dsp` device. The pure negotiation/fragment logic is
//! > unit-tested without hardware; live playback is verified on a node with an
//! > audio device.

#![cfg(all(target_os = "freebsd", feature = "oss"))]

use std::ffi::CString;
use std::io;
use std::sync::mpsc;
use std::time::Duration;

use audio_core_bsd::AudioFrame;

use crate::backend::{AudioBackend, OutputSink};
use crate::device::{DeviceDirection, DeviceInfo, StreamParams};
use crate::error::{IoError, Result};
use crate::sample_conv;

// =============================================================================
// OSS ioctl constants — extracted from FreeBSD `<sys/soundcard.h>` (BSD encoding)
// =============================================================================
// These differ from Linux OSS numbers (FreeBSD uses the BSD _IOC encoding:
// `(inout | (len << 16) | (group << 8) | num)`), which is why this backend uses
// raw constants rather than the oss-sys crate (which targets Linux semantics).
#[allow(dead_code)]
const SNDCTL_DSP_RESET: libc::c_ulong = 0x2000_5000;
const SNDCTL_DSP_SPEED: libc::c_ulong = 0xc004_5002;
const SNDCTL_DSP_SETFMT: libc::c_ulong = 0xc004_5005;
const SNDCTL_DSP_CHANNELS: libc::c_ulong = 0xc004_5006;
const SNDCTL_DSP_GETFMTS: libc::c_ulong = 0x4004_500b;
const SNDCTL_DSP_SETFRAGMENT: libc::c_ulong = 0xc004_500a;
#[allow(dead_code)]
const SNDCTL_DSP_GETOSPACE: libc::c_ulong = 0x4010_500c;

/// `AFMT_FLOAT` — 32-bit IEEE float (preferred; not all drivers support it).
const AFMT_FLOAT: i32 = 0x1000_0000;
/// `AFMT_S32_LE` — signed 32-bit little-endian (first integer fallback).
const AFMT_S32_LE: i32 = 0x0000_1000;
/// `AFMT_S16_LE` — signed 16-bit little-endian (final fallback).
const AFMT_S16_LE: i32 = 0x0000_0010;

/// OSS buffer-space query result (`struct audio_buf_info`, 16 bytes, 4×i32).
///
/// Reserved for the `GETOSPACE`-based backpressure path (a future optimisation
/// over the current blocking-write loop); kept here so the constant and struct
/// stay in lockstep with the extracted FreeBSD ABI.
#[allow(dead_code)]
#[repr(C)]
#[derive(Default, Clone, Copy, Debug)]
struct AudioBufInfo {
    /// Number of free fragments.
    fragments: i32,
    /// Total fragments allocated.
    fragstotal: i32,
    /// Size of one fragment in bytes.
    fragsize: i32,
    /// Free bytes available (≥ fragments × fragsize; includes partial frags).
    bytes: i32,
}

/// The negotiated device sample format, paired with its byte width.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum OssFormat {
    /// Native float (no conversion needed).
    Float,
    /// Signed 32-bit little-endian (4 bytes/sample).
    S32Le,
    /// Signed 16-bit little-endian (2 bytes/sample).
    S16Le,
}

impl OssFormat {
    /// The OSS `AFMT_*` constant the driver was told to use.
    const fn afmt(self) -> i32 {
        match self {
            OssFormat::Float => AFMT_FLOAT,
            OssFormat::S32Le => AFMT_S32_LE,
            OssFormat::S16Le => AFMT_S16_LE,
        }
    }

    /// Bytes per sample in the negotiated device format.
    const fn bytes_per_sample(self) -> usize {
        match self {
            OssFormat::Float | OssFormat::S32Le => 4,
            OssFormat::S16Le => 2,
        }
    }
}

// =============================================================================
// Pure logic (unit-testable without /dev/dsp)
// =============================================================================

/// Picks the best device format from a `GETFMTS` support mask, preferring
/// float, then s32le, then s16le. Returns `None` if none are supported.
#[must_use]
fn negotiate_format(supported_mask: i32) -> Option<OssFormat> {
    for (bit, fmt) in [
        (AFMT_FLOAT, OssFormat::Float),
        (AFMT_S32_LE, OssFormat::S32Le),
        (AFMT_S16_LE, OssFormat::S16Le),
    ] {
        if supported_mask & bit == bit {
            return Some(fmt);
        }
    }
    None
}

/// Encodes a `SNDCTL_DSP_SETFRAGMENT` argument: high 16 bits = fragment count,
/// low 16 bits = fragment size as log2(bytes). `frag_log2` must be ≥ 4.
#[must_use]
fn encode_fragment(num_frags: i32, frag_log2: i32) -> i32 {
    let n = num_frags.clamp(1, 0x7FFF);
    let s = frag_log2.clamp(4, 0xFFFF);
    (n << 0x10) | (s & 0xFFFF)
}

/// Estimates the round-trip buffer latency in milliseconds for a fragment
/// configuration. `bytes_per_sample` is the negotiated device format width.
///
/// Useful for mapping a requested `StreamParams.buffer_size` to the OSS fragment
/// parameters (ROADMAP §4.6).
// The `as f64` casts on `usize` are sound: channel/byte counts are tiny.
#[allow(clippy::cast_precision_loss)]
#[must_use]
pub fn fragment_latency_ms(
    num_frags: i32,
    frag_log2: i32,
    channels: u16,
    bytes_per_sample: usize,
    sample_rate: u32,
) -> f64 {
    let ch = channels.max(1) as usize;
    let bps = bytes_per_sample.max(1);
    let bytes_per_frag = f64::from(1_u32 << u32::try_from(frag_log2.max(0)).unwrap_or(0));
    let frames_per_frag = bytes_per_frag / (ch * bps) as f64;
    let total_frames = f64::from(num_frags) * frames_per_frag;
    total_frames / f64::from(sample_rate) * 1000.0
}

/// Issue an ioctl with a pointer argument, mapping OS errors to [`IoError`].
unsafe fn dsp_ioctl_ptr<T>(fd: libc::c_int, req: libc::c_ulong, arg: *mut T) -> Result<()> {
    let rc = libc::ioctl(fd, req, arg.cast::<libc::c_void>());
    if rc < 0 {
        Err(io::Error::last_os_error().into())
    } else {
        Ok(())
    }
}

/// Issue an ioctl with an `int` argument (read-write), returning the readback
/// value the driver wrote (used to confirm format/rate acceptance).
unsafe fn dsp_ioctl_int(fd: libc::c_int, req: libc::c_ulong, mut value: i32) -> Result<i32> {
    let rc = libc::ioctl(fd, req, core::ptr::addr_of_mut!(value));
    if rc < 0 {
        Err(io::Error::last_os_error().into())
    } else {
        Ok(value)
    }
}

// =============================================================================
// OssBackend
// =============================================================================

/// A FreeBSD-native [`AudioBackend`] that talks to OSS `/dev/dsp` directly.
///
/// Construct with [`OssBackend::new`]. Device enumeration scans `/dev/dsp` and
/// any `/dev/dspN` units; `/dev/dsp` (unit 0) is the default output.
pub struct OssBackend;

impl OssBackend {
    /// Creates a new OSS backend.
    #[must_use]
    pub const fn new() -> Self {
        Self
    }

    /// The default OSS device path.
    const DEFAULT_DEV: &'static str = "/dev/dsp";
}

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

impl AudioBackend for OssBackend {
    fn enumerate_devices(&self) -> Vec<DeviceInfo> {
        let mut devs = Vec::new();
        // The default device.
        devs.push(DeviceInfo::new(
            Self::DEFAULT_DEV,
            DeviceDirection::Duplex,
            2,
            vec![44_100, 48_000, 96_000],
            true,
        ));
        // Additional units /dev/dsp1..N (best-effort scan; absent units skipped).
        for unit in 1..=8 {
            let path = format!("/dev/dsp{unit}");
            if std::path::Path::new(&path).exists() {
                devs.push(DeviceInfo::new(
                    path,
                    DeviceDirection::Duplex,
                    2,
                    vec![44_100, 48_000, 96_000],
                    false,
                ));
            }
        }
        devs
    }

    fn default_output(&self) -> Option<DeviceInfo> {
        Some(DeviceInfo::new(
            Self::DEFAULT_DEV,
            DeviceDirection::Output,
            2,
            vec![44_100, 48_000, 96_000],
            true,
        ))
    }

    fn default_input(&self) -> Option<DeviceInfo> {
        // Input is stubbed for this release (see open_input).
        None
    }

    fn open_output(&self, dev: &str, params: StreamParams) -> Result<Box<dyn OutputSink>> {
        params.validate()?;

        // Open /dev/dsp for writing.
        let cdev = CString::new(dev).map_err(|e| IoError::DeviceNotFound(format!("{dev}: {e}")))?;
        // SAFETY: cdev is a valid NUL-terminated C string; O_WRONLY | O_CLOEXEC.
        let fd = unsafe { libc::open(cdev.as_ptr(), libc::O_WRONLY | libc::O_CLOEXEC) };
        if fd < 0 {
            let e = io::Error::last_os_error();
            // ENOENT/ENXIO → DeviceNotFound; EBUSY → Backend (busy).
            return match e.raw_os_error() {
                Some(libc::ENOENT | libc::ENXIO) => Err(IoError::DeviceNotFound(dev.into())),
                _ => Err(IoError::Backend(format!("open {dev}: {e}"))),
            };
        }

        // Negotiate the device format: probe GETFMTS, then SETFMT readback.
        let fmt = configure_output(fd, params)?;

        // SPSC ring of interleaved f32 samples shared with the audio thread.
        let (producer, consumer) = rtrb::RingBuffer::<f32>::new(RING_CAPACITY);
        let (drop_tx, drop_rx) = mpsc::channel::<()>();
        let (ready_tx, ready_rx) = mpsc::channel::<Result<()>>();

        let channels = params.channels;
        let sample_rate = params.sample_rate;

        std::thread::Builder::new()
            .name("audio-io-oss-output".into())
            .spawn(move || {
                run_output_thread(fd, consumer, channels, sample_rate, fmt, ready_tx, drop_rx);
            })
            .map_err(|e| IoError::Backend(format!("spawn oss output thread: {e}")))?;

        ready_rx
            .recv()
            .map_err(|e| IoError::StreamSetup(format!("oss output thread panicked: {e}")))??;

        Ok(Box::new(OssSink {
            producer,
            scratch: Vec::with_capacity(4 * channels as usize),
            channels,
            drop_tx: Some(drop_tx),
        }))
    }

    fn open_input(
        &self,
        _dev: &str,
        _params: StreamParams,
    ) -> Result<Box<dyn crate::backend::InputSource>> {
        // Capture is deferred to a follow-up; output is the MVP path.
        Err(IoError::UnsupportedConfig(
            "OSS capture (InputSource) is not yet implemented".into(),
        ))
    }
}

/// Ring capacity (interleaved f32 samples) — ~0.68 s @ 48 kHz stereo.
const RING_CAPACITY: usize = 1 << 16;

/// Probe and configure the device: format negotiation, channels, sample rate,
/// and fragment size. Returns the negotiated [`OssFormat`].
fn configure_output(fd: libc::c_int, params: StreamParams) -> Result<OssFormat> {
    // 1. Query supported formats.
    let mut fmts = 0_i32;
    // SAFETY: GETFMTS writes one int into &fmts.
    unsafe { dsp_ioctl_ptr(fd, SNDCTL_DSP_GETFMTS, core::ptr::addr_of_mut!(fmts)) }?;

    let chosen = negotiate_format(fmts)
        .ok_or_else(|| IoError::UnsupportedConfig("no supported OSS sample format".into()))?;

    // 2. SETFMT with readback confirmation.
    // SAFETY: SETFMT reads/writes one int.
    let readback = unsafe { dsp_ioctl_int(fd, SNDCTL_DSP_SETFMT, chosen.afmt()) }?;
    if readback != chosen.afmt() {
        return Err(IoError::UnsupportedConfig(format!(
            "driver rejected format {chosen:?} (readback 0x{readback:x})"
        )));
    }

    // 3. Channels.
    let mut chans = i32::from(params.channels);
    // SAFETY: CHANNELS reads/writes one int.
    let got_chans = unsafe { dsp_ioctl_int(fd, SNDCTL_DSP_CHANNELS, chans) }?;
    if got_chans != chans {
        // Some drivers round; accept if it matches a sane value.
        if !(1..=8).contains(&got_chans) {
            return Err(IoError::UnsupportedConfig(format!(
                "driver returned {got_chans} channels (requested {chans})"
            )));
        }
        chans = got_chans;
    }

    // 4. Sample rate.
    let rate = i32::try_from(params.sample_rate).unwrap_or(i32::MAX);
    // SAFETY: SPEED reads/writes one int.
    let got_rate = unsafe { dsp_ioctl_int(fd, SNDCTL_DSP_SPEED, rate) }?;
    if got_rate > 0 && (got_rate - rate).abs() > rate / 100 {
        return Err(IoError::UnsupportedConfig(format!(
            "driver returned {got_rate} Hz (requested {rate})"
        )));
    }

    // 5. Fragment size — target ~10 ms latency (§4.6).
    let ch_us = usize::try_from(chans).unwrap_or(0);
    let frag_log2 = compute_fragment_log2(ch_us, chosen.bytes_per_sample(), &params);
    let frag_arg = encode_fragment(4, frag_log2);
    // SAFETY: SETFRAGMENT reads/writes one int.
    let _ = unsafe { dsp_ioctl_int(fd, SNDCTL_DSP_SETFRAGMENT, frag_arg) };

    Ok(chosen)
}

/// Picks a fragment size (log2 bytes) targeting a small latency window.
#[allow(clippy::needless_pass_by_value)]
fn compute_fragment_log2(channels: usize, bytes_per_sample: usize, params: &StreamParams) -> i32 {
    // Target one fragment ≈ 256 frames: bytes = 256 * channels * bps.
    let target_frames = match params.buffer_size {
        crate::device::BufferSize::Fixed(n) => n.max(64),
        crate::device::BufferSize::Default => 256,
    };
    let target_bytes = target_frames * channels.max(1) * bytes_per_sample.max(1);
    let mut log2 = 4_i32;
    while (1 << log2) < target_bytes && log2 < 16 {
        log2 += 1;
    }
    log2
}

/// The dedicated audio-thread loop: drain the ring, convert f32→PCM, write.
// `ready_tx`/`drop_rx` are owned for the thread's lifetime (used by reference).
#[allow(clippy::needless_pass_by_value)]
fn run_output_thread(
    fd: libc::c_int,
    mut consumer: rtrb::Consumer<f32>,
    channels: u16,
    _sample_rate: u32,
    fmt: OssFormat,
    ready_tx: mpsc::Sender<Result<()>>,
    drop_rx: mpsc::Receiver<()>,
) {
    // Signal readiness (the device is configured and the loop is about to run).
    let _ = ready_tx.send(Ok(()));

    let ch = channels.max(1) as usize;
    let bps = fmt.bytes_per_sample();
    // Reused scratch buffers (allocated once on the audio thread).
    let mut f32_buf: Vec<f32> = Vec::with_capacity(1024);
    let mut pcm_bytes: Vec<u8> = Vec::with_capacity(4096);

    // Run while the sink is alive: `try_recv` is `Empty` while the sink holds
    // its sender, and `Disconnected` once `Drop for OssSink` drops it → the
    // `while let` exits and the thread closes the fd. (An explicit `Ok` signal,
    // currently unused, also exits.)
    while let Err(mpsc::TryRecvError::Empty) = drop_rx.try_recv() {
        // Drain whatever is available in whole-frame chunks.
        f32_buf.clear();
        while let Ok(s) = consumer.pop() {
            f32_buf.push(s);
        }
        if f32_buf.is_empty() {
            // No data yet — brief yield to avoid a busy spin.
            std::thread::sleep(Duration::from_millis(1));
            continue;
        }
        // Truncate to a whole number of frames.
        let n = (f32_buf.len() / ch) * ch;
        if n == 0 {
            continue;
        }

        // Convert f32 → PCM bytes according to the negotiated format.
        let needed = n * bps;
        if pcm_bytes.len() < needed {
            pcm_bytes.resize(needed, 0);
        }
        match fmt {
            OssFormat::Float => {
                // f32 → little-endian bytes directly.
                for (i, &v) in f32_buf.iter().take(n).enumerate() {
                    pcm_bytes[i * 4..i * 4 + 4].copy_from_slice(&v.to_le_bytes());
                }
            }
            OssFormat::S32Le => {
                let mut tmp = vec![0_i32; n];
                sample_conv::f32_interleaved_to_s32(&f32_buf[..n], &mut tmp);
                sample_conv::s32_interleaved_to_le_bytes(&tmp, &mut pcm_bytes[..needed]);
            }
            OssFormat::S16Le => {
                let mut tmp = vec![0_i16; n];
                sample_conv::f32_interleaved_to_s16(&f32_buf[..n], &mut tmp);
                sample_conv::s16_interleaved_to_le_bytes(&tmp, &mut pcm_bytes[..needed]);
            }
        }

        // Blocking write to the device (audio thread — not RT).
        let mut off = 0;
        while off < needed {
            // SAFETY: writing pcm_bytes[off..] to a valid fd.
            let wr = unsafe {
                libc::write(
                    fd,
                    pcm_bytes[off..].as_ptr().cast::<libc::c_void>(),
                    needed - off,
                )
            };
            if wr < 0 {
                let e = io::Error::last_os_error();
                if e.kind() == io::ErrorKind::WouldBlock {
                    break;
                }
                // Transient errors: stop this chunk but keep the loop alive.
                break;
            }
            off += usize::try_from(wr).unwrap_or(0);
            if wr == 0 {
                break;
            }
        }
    }
    // Thread exit: close the fd.
    // SAFETY: fd is a valid open descriptor owned solely by this thread.
    unsafe { libc::close(fd) };
}

/// Output sink backed by a `/dev/dsp` descriptor (owned on a dedicated thread)
/// and a lock-free ring — structurally identical to the cpal backend's sink.
pub struct OssSink {
    producer: rtrb::Producer<f32>,
    scratch: Vec<f32>,
    channels: u16,
    drop_tx: Option<mpsc::Sender<()>>,
}

impl OutputSink for OssSink {
    fn write(&mut self, frame: &AudioFrame) -> Result<()> {
        let n = frame.num_frames();
        let ch = (self.channels.min(frame.channels)) as usize;
        if ch == 0 {
            return Ok(());
        }
        // Interleave planar → interleaved f32 (RT-safe: bounded loop, reused buf).
        self.scratch.clear();
        self.scratch.reserve(n * ch);
        for i in 0..n {
            for c in 0..ch {
                let v = frame.channel_slice(c).get(i).copied().unwrap_or(0.0);
                self.scratch.push(v);
            }
        }
        for &s in &self.scratch {
            if self.producer.push(s).is_err() {
                return Err(IoError::StreamSetup("oss ring full (back-pressure)".into()));
            }
        }
        Ok(())
    }
}

impl Drop for OssSink {
    fn drop(&mut self) {
        // Signal the audio thread to drain and exit (closes the fd).
        self.drop_tx.take();
    }
}

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

    #[test]
    fn negotiate_prefers_float_then_s32_then_s16() {
        assert_eq!(
            negotiate_format(AFMT_FLOAT | AFMT_S32_LE | AFMT_S16_LE),
            Some(OssFormat::Float)
        );
        assert_eq!(
            negotiate_format(AFMT_S32_LE | AFMT_S16_LE),
            Some(OssFormat::S32Le)
        );
        assert_eq!(negotiate_format(AFMT_S16_LE), Some(OssFormat::S16Le));
        assert_eq!(negotiate_format(0), None);
    }

    #[test]
    fn format_afmt_and_width() {
        assert_eq!(OssFormat::Float.afmt(), AFMT_FLOAT);
        assert_eq!(OssFormat::S32Le.afmt(), AFMT_S32_LE);
        assert_eq!(OssFormat::S16Le.afmt(), AFMT_S16_LE);
        assert_eq!(OssFormat::S16Le.bytes_per_sample(), 2);
        assert_eq!(OssFormat::S32Le.bytes_per_sample(), 4);
    }

    #[test]
    fn encode_fragment_packs_count_and_log2() {
        // 4 fragments × 2^10 bytes.
        let arg = encode_fragment(4, 10);
        assert_eq!(arg, (4 << 0x10) | 0xA);
    }

    #[test]
    fn encode_fragment_clamps_extremes() {
        let arg = encode_fragment(0, 2);
        // num clamped to 1, log2 clamped to 4.
        assert_eq!(arg, (1 << 0x10) | 0x4);
    }

    #[test]
    fn fragment_latency_matches_known_case() {
        // Stereo s16le @ 48 kHz: S=10 (1024 B = 256 frame/frag), N=4 → 1024
        // frames ≈ 21.3 ms (ROADMAP §4.6 example).
        let ms = fragment_latency_ms(4, 10, 2, 2, 48_000);
        assert!((ms - 21.33).abs() < 0.1, "got {ms} ms");
    }

    #[test]
    fn fragment_latency_doubles_with_fragment_count() {
        let one = fragment_latency_ms(2, 10, 2, 2, 48_000);
        let two = fragment_latency_ms(4, 10, 2, 2, 48_000);
        assert!((two - 2.0 * one).abs() < 0.01);
    }

    #[test]
    fn oss_backend_enumerates_default_device() {
        let b = OssBackend::new();
        let devs = b.enumerate_devices();
        assert!(devs.iter().any(|d| d.name == "/dev/dsp" && d.is_default));
    }

    #[test]
    fn oss_backend_has_default_output() {
        let b = OssBackend::new();
        let dev = b.default_output().unwrap();
        assert_eq!(dev.name, "/dev/dsp");
    }

    #[test]
    fn oss_backend_open_output_unknown_device_is_not_found() {
        let b = OssBackend::new();
        // A path that cannot contain a NUL and does not exist.
        let err = b
            .open_output("/dev/dsp_nonexistent_unit", StreamParams::pcm_48k_stereo())
            .err()
            .unwrap();
        assert!(err.to_string().contains("device not found") || err.to_string().contains("open"));
    }

    #[test]
    fn open_input_reports_unsupported() {
        let b = OssBackend::new();
        let err = b
            .open_input("/dev/dsp", StreamParams::pcm_48k_mono())
            .err()
            .unwrap();
        assert!(err.to_string().contains("not yet implemented"));
    }

    #[test]
    fn open_output_validates_params() {
        let b = OssBackend::new();
        let err = b
            .open_output("/dev/dsp", StreamParams::pcm_48k_stereo().with_channels(0))
            .err()
            .unwrap();
        assert!(err.to_string().contains("invalid channel count"));
    }

    /// Verifies the audio-thread shutdown contract: while the sink is alive
    /// `try_recv` is `Empty`; after the sink's `drop_tx` is dropped (mirroring
    /// `Drop for OssSink`), `try_recv` becomes `Disconnected`, which the loop
    /// treats as shutdown. This is the lifecycle the `run_output_thread` loop
    /// depends on (the sink's `drop_tx` is dropped, never sent — so the loop
    /// keys off `Disconnected`, not a message).
    #[test]
    fn shutdown_signal_disconnects_on_drop() {
        let (drop_tx, drop_rx) = mpsc::channel::<()>();
        // Alive: Empty (loop continues).
        assert_eq!(drop_rx.try_recv(), Err(mpsc::TryRecvError::Empty));
        // Sink dropped → sender taken and dropped (never sends).
        drop(drop_tx);
        // Now Disconnected → loop breaks, thread exits, fd closes.
        assert_eq!(drop_rx.try_recv(), Err(mpsc::TryRecvError::Disconnected));
    }
}