punktfunk/crates/pf-capture/src/linux/pipewire.rs

//! The PipeWire consumer, confined to its own thread (the PW types are `!Send`).

use super::pw_cursor::{composite_cursor, update_cursor_meta, CursorState};
use super::pw_pods::{
    build_cursor_meta_param, build_default_format_obj, build_dmabuf_buffers, build_dmabuf_format,
    build_hdr_dmabuf_format, build_mappable_buffers, build_shm_only_buffers, serialize_pod,
};
use super::{CapturedFrame, DmabufFrame, FramePayload, PixelFormat, ZeroCopyPolicy};
use anyhow::{Context, Result};
use pipewire as pw;
use pw::{properties::properties, spa};
use std::os::fd::{FromRawFd, OwnedFd};
use std::sync::atomic::Ordering;
use std::sync::mpsc::SyncSender;
use std::time::{SystemTime, UNIX_EPOCH};

use spa::param::video::{VideoFormat, VideoInfoRaw};
use spa::pod::Pod;

/// Map a negotiated SPA video format to a layout the encoder can consume. Returns
/// `None` for formats we don't handle (the frame is then skipped).
fn map_format(f: VideoFormat) -> Option<PixelFormat> {
    Some(match f {
        VideoFormat::BGRx => PixelFormat::Bgrx,
        VideoFormat::RGBx => PixelFormat::Rgbx,
        VideoFormat::BGRA => PixelFormat::Bgra,
        VideoFormat::RGBA => PixelFormat::Rgba,
        VideoFormat::RGB => PixelFormat::Rgb,
        VideoFormat::BGR => PixelFormat::Bgr,
        VideoFormat::NV12 => PixelFormat::Nv12,
        // The GNOME 50+ HDR screencast formats (packed 2:10:10:10; only ever negotiated by
        // the `want_hdr` offer, whose MANDATORY colorimetry props pin them to PQ/BT.2020).
        VideoFormat::xRGB_210LE => PixelFormat::X2Rgb10,
        VideoFormat::xBGR_210LE => PixelFormat::X2Bgr10,
        _ => return None,
    })
}

struct UserData {
    info: VideoInfoRaw,
    /// Negotiated layout (`None` until param_changed, or if unsupported).
    format: Option<PixelFormat>,
    /// Negotiated DRM format modifier (for dmabuf import); 0 = LINEAR.
    modifier: u64,
    /// The one-deep frame mailbox (see [`super::FrameSlot`]) + its wakeup sender. Written
    /// through [`UserData::publish`], never directly.
    slot: super::FrameSlot,
    wake: SyncSender<()>,
    /// Everything this thread publishes to the capturer — see [`super::CaptureSignals`], which also
    /// documents each flag's contract (it used to be restated here, and drifted).
    signals: super::CaptureSignals,
    /// Consecutive tiled-import failures (reset on success); see [`IMPORT_FAIL_POISON`].
    import_fail_streak: u32,
    /// Present when zero-copy is enabled on NVIDIA: imports a dmabuf → CUDA device buffer,
    /// normally via the isolated worker process (`pf_zerocopy::Importer::Remote`).
    importer: Option<pf_zerocopy::Importer>,
    /// VAAPI zero-copy: hand the raw dmabuf to the encoder (which imports + GPU-CSCs it) instead
    /// of a CUDA import. Set when zero-copy is on, the EGL→CUDA importer is unavailable, and the
    /// encoder backend is VAAPI (AMD/Intel).
    vaapi_passthrough: bool,
    /// `PUNKTFUNK_NV12`: on the tiled EGL/GL zero-copy path, convert to NV12 on the GPU and feed
    /// NVENC native YUV (Tier 2A). Off ⇒ the BGRx path is unchanged.
    nv12: bool,
    /// 4:4:4 session: on the tiled EGL/GL zero-copy path, convert to planar YUV444 on the GPU
    /// (`ImportKind::Tiled444`) and feed NVENC native full-chroma YUV — takes precedence over
    /// `nv12` (a 4:4:4 session must never subsample).
    yuv444: bool,
    /// The LINEAR (gamescope) NV12 compute CSC failed once — RGB for the rest of the stream
    /// (T2.5b's per-stream fallback latch; cleared by the next stream's fresh `Ud`).
    linear_nv12_failed: bool,
    /// Rate-limit counter for the latest-frame-only diagnostic log (see `.process`).
    dbg_log_n: u64,
    /// Cursor-as-metadata state, composited into the CPU de-pad path (see `consume_frame`).
    cursor: CursorState,
    /// `Some((w, h))` while the producer's negotiated size is a sacrificial birth mode and a
    /// renegotiation to these dims is guaranteed (KWin virtual outputs — kwin.rs `create`):
    /// `.process` skips whole buffers until the negotiated size matches, then clears this
    /// (self-disarming — later legitimate resizes are unaffected). `None` = no gating.
    expect_dims: Option<(u32, u32)>,
    /// Buffers skipped by the `expect_dims` gate (rate-limits its log).
    gate_skips: u64,
    /// When the gate first held a buffer — after [`GATE_DEADLINE`] with no renegotiation the
    /// gate disarms and accepts what the producer serves (degraded dims beat a session wedged
    /// into the first-frame-timeout retry loop; the promised renegotiation normally lands
    /// within a frame or two).
    gate_since: Option<std::time::Instant>,
}

impl UserData {
    /// Hand `frame` to the consumer as THE latest, OVERWRITING any frame it has not taken yet
    /// (drop-oldest — see [`super::FrameSlot`]), then poke the wakeup edge.
    ///
    /// Never blocks the PipeWire loop, which is the hard constraint here: this runs inside
    /// `.process`, so blocking would stall the compositor's stream. Both operations are
    /// best-effort — a full wakeup channel means an edge is already pending (nothing lost, the
    /// slot is the truth), and a poisoned mutex is unreachable in practice (the only critical
    /// sections are a `take` and this store).
    fn publish(&self, frame: CapturedFrame) {
        if let Ok(mut slot) = self.slot.lock() {
            *slot = Some(frame);
        }
        let _ = self.wake.try_send(());
    }
}

/// Everything the zero-copy negotiation decision depends on, gathered at ONE point in time.
/// Split out from [`pipewire_thread`]'s prologue so the decision is a pure function of these
/// facts — testable, and consumable by both the thread and `spawn_pipewire` (see
/// [`NegotiationPlan`]).
#[derive(Debug, Clone, Copy)]
pub(super) struct NegotiationInputs {
    /// `allow_zerocopy && pf_zerocopy::enabled()`.
    pub zerocopy: bool,
    /// `PUNKTFUNK_FORCE_SHM` — the race-free download path.
    pub force_shm: bool,
    /// This session offers the 10-bit PQ/BT.2020 formats.
    pub want_hdr: bool,
    /// This session negotiated full-chroma 4:4:4.
    pub want_444: bool,
    /// [`ZeroCopyPolicy::backend_is_vaapi`].
    pub backend_is_vaapi: bool,
    /// [`ZeroCopyPolicy::pyrowave_session`].
    pub pyrowave_session: bool,
    /// [`ZeroCopyPolicy::native_nv12_session`].
    pub native_nv12_session: bool,
    /// `pf_zerocopy::raw_dmabuf_import_disabled()` — the scoped raw-passthrough latch.
    pub raw_dmabuf_import_disabled: bool,
    /// `pf_zerocopy::gpu_import_disabled()` — repeated import-worker deaths.
    pub gpu_import_disabled: bool,
    /// `PUNKTFUNK_PIPEWIRE_NV12` (default ON) — allow the producer-side NV12 preference.
    pub native_nv12_env_on: bool,
}

/// The resolved zero-copy negotiation decision — **one resolver, consumed by the PipeWire thread
/// AND by `spawn_pipewire`.**
///
/// `spawn_pipewire` used to hand-mirror `vaapi_passthrough` so the capturer's
/// negotiation-timeout branch could tell which offer had failed, and the copy drifted: it omitted
/// `raw_dmabuf_import_disabled`, so after that latch fired the mirror still said "raw
/// passthrough" while the thread had already fallen back — and the timeout branch then latched a
/// downgrade for an offer it had not made. Deriving both consumers from this struct is what makes
/// that class of drift unrepresentable rather than merely fixed (same shape as WP7.6's single
/// Linux encode-backend resolver).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct NegotiationPlan {
    /// Build the EGL→CUDA importer for this capture.
    pub build_importer: bool,
    /// Hand raw dmabufs straight to the encoder instead of importing to CUDA.
    pub vaapi_passthrough: bool,
    /// Offer gamescope's producer-side NV12 pod ahead of packed RGB.
    pub prefer_native_nv12: bool,
    /// Carried through so [`want_dmabuf`](Self::want_dmabuf) needs no second copy of it.
    pub force_shm: bool,
    /// Diagnostic: this capture WOULD have taken the raw passthrough, but its scoped latch is
    /// set (the encoder repeatedly failed to import, or a previous negotiation timed out).
    pub raw_dmabuf_latched: bool,
    /// Diagnostic: this capture WOULD have built the EGL→CUDA importer, but repeated
    /// import-worker deaths latched the GPU import off.
    pub gpu_import_latched: bool,
}

/// Resolve the negotiation plan. **Pure** — every environment read is already in `i`.
///
/// The four invariants this encodes were previously prose-only comments spread across the
/// prologue; `negotiation_plan_invariants` in the tests below pins each one:
///   1. HDR never builds the EGL→CUDA importer (its de-tile blit is 8-bit RGBA8 → silent depth
///      loss); the HDR consumers are the CPU mmap path and the raw passthrough.
///   2. 4:4:4 never prefers producer NV12 (a 4:4:4 session must not be subsampled).
///   3. Producer-native NV12 only on a `native_nv12_session` under an active raw passthrough
///      (libav VAAPI would misread the two-plane buffer; the CUDA importer expects packed RGB).
///   4. The raw passthrough is off whenever its own latch has fired.
pub(super) fn negotiation_plan(i: NegotiationInputs) -> NegotiationPlan {
    // The frames' consumer imports raw dmabufs itself: the VAAPI backend (libva import + GPU
    // CSC) or a PyroWave session (the wavelet encoder's own Vulkan device, any vendor).
    let raw_passthrough = i.backend_is_vaapi || i.pyrowave_session;
    // Building the EGL→CUDA importer would waste a CUDA probe under a raw passthrough — or
    // worse, succeed and produce CUDA payloads only NVENC can consume. HDR is excluded by
    // invariant 1. `gpu_import_disabled` is the repeated-worker-death latch (a wedged GPU stack
    // must not crash-loop).
    let build_importer = i.zerocopy && !raw_passthrough && !i.want_hdr && !i.gpu_import_disabled;
    // Note there is no `importer.is_none()` term, unlike the expression this replaces: it was
    // redundant (`build_importer` already excludes `raw_passthrough`, so the importer is
    // necessarily absent here) and it is what made the decision look impure — the reason
    // `spawn_pipewire` "had to" mirror it by hand.
    let vaapi_passthrough =
        i.zerocopy && !i.force_shm && raw_passthrough && !i.raw_dmabuf_import_disabled;
    let prefer_native_nv12 = i.native_nv12_env_on
        && i.native_nv12_session
        && i.backend_is_vaapi
        && vaapi_passthrough
        && !i.pyrowave_session
        && !i.want_444
        && !i.want_hdr;
    NegotiationPlan {
        build_importer,
        vaapi_passthrough,
        prefer_native_nv12,
        force_shm: i.force_shm,
        raw_dmabuf_latched: i.zerocopy
            && !i.force_shm
            && raw_passthrough
            && i.raw_dmabuf_import_disabled,
        gpu_import_latched: i.zerocopy && !raw_passthrough && !i.want_hdr && i.gpu_import_disabled,
    }
}

impl NegotiationPlan {
    /// Whether to request dmabuf buffers. Not part of the plan proper: it depends on whether the
    /// importer actually CONSTRUCTED (`have_importer` — a GPU/driver fact) and on the modifier
    /// list that construction yielded.
    pub(super) fn want_dmabuf(&self, have_importer: bool, modifiers: &[u64]) -> bool {
        (have_importer || self.vaapi_passthrough) && !modifiers.is_empty() && !self.force_shm
    }
}

/// Consecutive tiled-import failures (worker alive, e.g. a per-buffer `EGL_BAD_MATCH`) before
/// the stream is poisoned for rebuild. A tiled import failure must NEVER fall through to the
/// CPU mmap path — de-padding tiled bytes as linear produces a scrambled image — so after a
/// short streak of dropped frames the capturer fails loudly and the session renegotiates.
const IMPORT_FAIL_POISON: u32 = 3;

/// Log a frame-drop reason once per process (the process callback runs per frame; a stuck
/// pipeline must say why without flooding).
fn warn_once(msg: &'static str) {
    use std::sync::Mutex;
    static SEEN: Mutex<Vec<&'static str>> = Mutex::new(Vec::new());
    let mut seen = SEEN.lock().unwrap();
    if !seen.contains(&msg) {
        seen.push(msg);
        tracing::warn!("{msg}");
    }
}

/// A read-only mmap of a dmabuf fd, unmapped on drop. Used when MAP_BUFFERS didn't map the
/// buffer (producers don't always flag dmabufs mappable, e.g. gamescope's Vulkan exports).
struct DmabufMap {
    ptr: *mut std::ffi::c_void,
    len: usize,
}

impl DmabufMap {
    fn new(fd: i32, len: usize) -> Option<DmabufMap> {
        // SAFETY: a null `addr` lets the kernel choose the mapping address; `fd` is a caller-owned
        // dmabuf/MemFd fd, valid for the duration of this call, and `len` is the requested map length.
        // `mmap` reads no Rust memory — it installs a fresh PROT_READ/MAP_SHARED page mapping and
        // returns its base (or MAP_FAILED, checked below before `DmabufMap` adopts it). The returned
        // region is a brand-new VMA, so it aliases no live Rust object, and it keeps the underlying
        // object mapped independently of `fd` (which may be closed after this returns).
        let ptr = unsafe {
            libc::mmap(
                std::ptr::null_mut(),
                len,
                libc::PROT_READ,
                libc::MAP_SHARED,
                fd,
                0,
            )
        };
        (ptr != libc::MAP_FAILED).then_some(DmabufMap { ptr, len })
    }
}

impl Drop for DmabufMap {
    fn drop(&mut self) {
        // SAFETY: `self.ptr`/`self.len` are exactly the base+length of a successful `mmap` in
        // `DmabufMap::new` (constructed only when `ptr != MAP_FAILED`). This `DmabufMap` uniquely owns
        // that mapping and `drop` runs once, so `munmap` releases a live mapping exactly once — no
        // double-unmap. Every `&[u8]` derived from the mapping is bounded by this `DmabufMap`'s
        // lifetime, so no borrow outlives the unmap.
        unsafe {
            libc::munmap(self.ptr, self.len);
        }
    }
}

/// De-pad / import a single PipeWire buffer and push it to the encoder. Called from the
/// `.process` callback with the NEWEST drained buffer (latest-frame-only). `datas` is sourced
/// via the same transparent cast libspa's `Buffer::datas_mut` performs, so the safe `Data`
/// accessors (`.type_()`, `.chunk()`, `.data()`, `.fd()`, `.as_raw()`) keep working.
fn consume_frame(ud: &mut UserData, spa_buf: *mut spa::sys::spa_buffer) {
    // No active stream: release the buffer without the (expensive at 5K) de-pad.
    if !ud.signals.active.load(Ordering::Relaxed) {
        return;
    }
    // Poisoned (GPU import lost): the capturer is already surfacing an error to the encode
    // loop; skip per-frame work until the rebuild tears this stream down.
    if ud.signals.broken.load(Ordering::Relaxed) {
        return;
    }
    // SAFETY: `spa_buf` is the `*mut spa_buffer` of the PipeWire buffer we dequeued and still hold for
    // this `.process` callback (not requeued until after `consume_frame` returns), so it is live. The
    // block null-checks `spa_buf`, requires `n_datas != 0`, and null-checks the `datas` array pointer
    // before forming any slice. `(*spa_buf).datas` points to `n_datas` libspa `spa_data` structs, and
    // `pw::spa::buffer::Data` is `#[repr(transparent)]` over `spa_data` (the same cast
    // `Buffer::datas_mut` performs — see the function doc), so the pointer cast + length describe
    // exactly that array, in bounds. The PipeWire loop is single-threaded and owns the buffer here, so
    // this `&mut` slice is the only reference to it (no aliasing/data race).
    let datas: &mut [pw::spa::buffer::Data] = unsafe {
        if spa_buf.is_null() || (*spa_buf).n_datas == 0 || (*spa_buf).datas.is_null() {
            &mut []
        } else {
            std::slice::from_raw_parts_mut(
                (*spa_buf).datas as *mut pw::spa::buffer::Data,
                (*spa_buf).n_datas as usize,
            )
        }
    };
    if datas.is_empty() {
        return;
    }
    let sz = ud.info.size();
    let (w, h) = (sz.width as usize, sz.height as usize);
    if w == 0 || h == 0 {
        return; // format not negotiated yet
    }

    // Implicit-fence wait: Mutter renders into the dmabuf and hands it over at
    // GPU-submit time; with no producer explicit sync (Mutter+NVIDIA can't) we snapshot
    // the buffer's implicit fence and wait the producer's render before sampling —
    // closing the stale/old-frame race on NVIDIA. No-op for shm buffers or drivers that
    // attach no fence. Covers both the GPU import and the CPU mmap read below.
    if datas[0].type_() == pw::spa::buffer::DataType::DmaBuf {
        match pf_zerocopy::dmabuf_fence::wait_read_ready(datas[0].fd(), 100) {
            Ok(waited) => {
                static F1: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(true);
                if F1.swap(false, Ordering::Relaxed) {
                    tracing::info!(
                        waited,
                        "dmabuf implicit-fence sync active (waited=true → driver fences \
                         the render, race closed; false → no implicit fence, zero-copy \
                         may still show stale frames)"
                    );
                }
            }
            Err(e) => {
                static F2: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(true);
                if F2.swap(false, Ordering::Relaxed) {
                    tracing::warn!(
                        error = %e,
                        "dmabuf EXPORT_SYNC_FILE failed — no implicit-fence sync; NVIDIA \
                         zero-copy may show stale frames (no producer explicit sync)"
                    );
                }
            }
        }
    }

    // Raw DMA-BUF passthrough: packed RGB is imported for GPU CSC; producer-native NV12 can
    // be consumed by the Vulkan Video encoder without another color conversion.
    if ud.vaapi_passthrough {
        if let Some(fmt) = ud.format {
            if datas[0].type_() == pw::spa::buffer::DataType::DmaBuf {
                if let Some(fourcc) = pf_frame::drm_fourcc(fmt) {
                    let chunk = datas[0].chunk();
                    let offset = chunk.offset();
                    let stride = chunk.stride().max(0) as u32;
                    // Native NV12 usually arrives as a two-plane SPA buffer over ONE buffer
                    // object; plane 1's chunk carries the REAL UV offset/stride (compositors
                    // may align the Y plane before UV). Pass it through instead of assuming
                    // contiguity. Each spa_data holds its own (dup'd) fd, so BO identity is
                    // by inode, not fd number; a genuinely two-BO frame cannot travel through
                    // the single-fd import — drop it with a diagnosis instead of streaming
                    // garbage chroma.
                    let plane1 =
                        if fmt == PixelFormat::Nv12 && datas.len() >= 2 && datas[1].fd() > 0 {
                            // SAFETY: zeroed `libc::stat` is a valid POD initializer; both fds are
                            // owned by the live PipeWire buffer for this callback, and `fstat`
                            // only writes the out-param structs, whose fields are read only after
                            // the `== 0` success checks.
                            let same_bo = unsafe {
                                let mut s0: libc::stat = std::mem::zeroed();
                                let mut s1: libc::stat = std::mem::zeroed();
                                libc::fstat(datas[0].fd() as i32, &mut s0) == 0
                                    && libc::fstat(datas[1].fd() as i32, &mut s1) == 0
                                    && (s0.st_dev, s0.st_ino) == (s1.st_dev, s1.st_ino)
                            };
                            if !same_bo {
                                warn_once(
                                    "NV12 planes live in different buffer objects — frames \
                                 dropped (single-fd import only)",
                                );
                                return;
                            }
                            let c1 = datas[1].chunk();
                            Some((c1.offset(), c1.stride().max(0) as u32))
                        } else {
                            None
                        };
                    // dup the fd so it survives the SPA buffer recycle — the encode thread
                    // imports it. Content stability across the brief import/encode window relies
                    // on the compositor's buffer-pool depth, like any zero-copy capture.
                    // SAFETY: `datas[0].fd()` is the dmabuf fd owned by the live PipeWire buffer (valid
                    // for this callback). `fcntl(fd, F_DUPFD_CLOEXEC, 0)` reads only the integer fd,
                    // touches no Rust memory, and returns a fresh independent CLOEXEC duplicate (or -1).
                    // The original stays owned by PipeWire; the dup is a new fd we own (checked >= 0).
                    let dup =
                        unsafe { libc::fcntl(datas[0].fd() as i32, libc::F_DUPFD_CLOEXEC, 0) };
                    if dup >= 0 {
                        let pts_ns = SystemTime::now()
                            .duration_since(UNIX_EPOCH)
                            .map(|d| d.as_nanos() as u64)
                            .unwrap_or(0);
                        ud.publish(CapturedFrame {
                            width: w as u32,
                            height: h as u32,
                            pts_ns,
                            format: fmt,
                            payload: FramePayload::Dmabuf(DmabufFrame {
                                // SAFETY: `dup` is the fresh fd `fcntl(F_DUPFD_CLOEXEC)` just returned
                                // (checked `dup >= 0`); nothing else owns it, so `OwnedFd` takes sole
                                // ownership and closes it exactly once on drop — no alias, no
                                // double-close.
                                fd: unsafe { OwnedFd::from_raw_fd(dup) },
                                fourcc,
                                modifier: ud.modifier,
                                offset,
                                stride,
                                plane1,
                            }),
                            // Cursor-as-metadata is blended only by RGB→NV12 backends. Gamescope
                            // embeds its pointer in the produced pixels, so native NV12 has none.
                            cursor: ud.cursor.overlay(),
                        });
                        static ONCE: std::sync::atomic::AtomicBool =
                            std::sync::atomic::AtomicBool::new(true);
                        if ONCE.swap(false, Ordering::Relaxed) {
                            tracing::info!(
                                w,
                                h,
                                modifier = ud.modifier,
                                fourcc = format_args!("{:#010x}", fourcc),
                                source = if fmt == PixelFormat::Nv12 {
                                    "producer-native NV12"
                                } else {
                                    "packed RGB (encoder GPU CSC)"
                                },
                                "zero-copy: handing the raw DMA-BUF to the encoder"
                            );
                        }
                        return;
                    }
                }
            }
        }
        // Not a dmabuf (or unmappable format) — fall through to the CPU de-pad path.
    }

    // Zero-copy path: if the buffer is a dmabuf and we have an importer, import it
    // into a CUDA device buffer (no CPU touch) and deliver that. Otherwise fall
    // through to the shm de-pad copy below.
    let mut gpu_import_broken = false;
    if let (Some(importer), Some(fmt)) = (ud.importer.as_mut(), ud.format) {
        // Defense-in-depth: the 10-bit PQ formats must never enter the EGL→CUDA import (its
        // de-tile blit is 8-bit RGBA8 — silent depth loss). An HDR offer never builds the
        // importer, so this gate only matters if those invariants ever drift apart.
        if datas[0].type_() == pw::spa::buffer::DataType::DmaBuf && !fmt.is_hdr_rgb10() {
            let plane = pf_zerocopy::DmabufPlane {
                fd: datas[0].fd(),
                offset: datas[0].chunk().offset(),
                stride: datas[0].chunk().stride().max(0) as u32,
            };
            // Tiled modifier → EGL/GL de-tile import; LINEAR (0/unset, e.g.
            // gamescope) → direct CUDA external-memory import (NVIDIA EGL can't
            // sample LINEAR).
            let modifier = (ud.modifier != 0).then_some(ud.modifier);
            if let Some(fourcc) = pf_frame::drm_fourcc(fmt) {
                // GPU converts: a 4:4:4 session gets the planar-YUV444 convert on the tiled
                // EGL/GL path (full chroma, takes precedence over NV12 — 4:4:4 must never
                // subsample), otherwise `PUNKTFUNK_NV12` gets NV12 — tiled via the EGL/GL
                // blit, LINEAR/gamescope via the Vulkan bridge's compute CSC (latency plan
                // T2.5b) — so NVENC encodes native YUV and skips its internal RGB→YUV CSC on
                // the contended SM. A 4:4:4 session on LINEAR frames has no convert and
                // stays RGB, falling to the encoder's clear-error path (`want_444` with an
                // RGB CUDA payload) rather than silently subsampling. A LINEAR NV12 convert
                // failure latches RGB for the stream (mid-frame fallback, no drop).
                let yuv444 = ud.yuv444 && modifier.is_some();
                let mut nv12 = ud.nv12 && !ud.yuv444;
                let imported = if let Some(m) = modifier {
                    if yuv444 {
                        importer.import_yuv444(&plane, w as u32, h as u32, fourcc, Some(m))
                    } else if nv12 {
                        importer.import_nv12(&plane, w as u32, h as u32, fourcc, Some(m))
                    } else {
                        importer.import(&plane, w as u32, h as u32, fourcc, Some(m))
                    }
                } else if nv12 && !ud.linear_nv12_failed {
                    match importer.import_linear_nv12(&plane, w as u32, h as u32) {
                        Ok(buf) => Ok(buf),
                        Err(e) => {
                            ud.linear_nv12_failed = true;
                            nv12 = false;
                            tracing::warn!(error = %format!("{e:#}"),
                                "LINEAR NV12 compute CSC failed — RGB for the rest of this \
                                 stream (NVENC does the CSC internally)");
                            importer.import_linear(&plane, w as u32, h as u32)
                        }
                    }
                } else {
                    nv12 = false;
                    importer.import_linear(&plane, w as u32, h as u32)
                };
                match imported {
                    Ok(devbuf) => {
                        ud.import_fail_streak = 0;
                        pf_zerocopy::note_gpu_import_ok();
                        static ONCE: std::sync::atomic::AtomicBool =
                            std::sync::atomic::AtomicBool::new(true);
                        if ONCE.swap(false, Ordering::Relaxed) {
                            tracing::info!(
                                w,
                                h,
                                modifier = ud.modifier,
                                nv12,
                                yuv444,
                                "zero-copy: dmabuf imported to CUDA (no CPU copy)"
                            );
                        }
                        let pts_ns = SystemTime::now()
                            .duration_since(UNIX_EPOCH)
                            .map(|d| d.as_nanos() as u64)
                            .unwrap_or(0);
                        ud.publish(CapturedFrame {
                            width: w as u32,
                            height: h as u32,
                            pts_ns,
                            format: if yuv444 {
                                PixelFormat::Yuv444
                            } else if nv12 {
                                PixelFormat::Nv12
                            } else {
                                fmt
                            },
                            payload: FramePayload::Cuda(devbuf),
                            // Cursor-as-metadata: blended by the CUDA encoder into its owned
                            // device surface. (RGB LINEAR-import case; YUV sessions blend planes.)
                            cursor: ud.cursor.overlay(),
                        });
                        return;
                    }
                    Err(e) => {
                        let dead = importer.dead();
                        if dead {
                            pf_zerocopy::note_gpu_import_death();
                        }
                        if modifier.is_some() {
                            // Tiled buffer: the CPU fallback below would mmap TILED bytes
                            // and de-pad them as linear — a scrambled image, worse than no
                            // frame. Drop the frame instead; on a dead worker (it absorbed a
                            // driver fault) or a short failure streak, poison the stream so
                            // the session's capture-loss rebuild renegotiates cleanly.
                            ud.import_fail_streak += 1;
                            if dead || ud.import_fail_streak >= IMPORT_FAIL_POISON {
                                tracing::error!(error = %format!("{e:#}"), dead,
                                    "tiled GPU import lost — failing this capture for rebuild");
                                ud.signals.broken.store(true, Ordering::Relaxed);
                            } else {
                                tracing::warn!(error = %format!("{e:#}"),
                                    streak = ud.import_fail_streak,
                                    "tiled dmabuf GPU import failed — frame dropped");
                            }
                            return;
                        }
                        // LINEAR dmabuf: CPU-mappable, so disable the importer and fall
                        // through to the CPU mmap path — degraded, not dead.
                        tracing::warn!(error = %format!("{e:#}"),
                            "LINEAR dmabuf GPU import failed — falling back to the CPU copy path");
                        gpu_import_broken = true;
                    }
                }
            } else {
                return; // format has no DRM fourcc mapping — skip the frame
            }
        }
    }
    if gpu_import_broken {
        ud.importer = None;
    }

    let d = &mut datas[0];
    // CPU path may also receive LINEAR dmabufs (gamescope offers only those once its
    // modifier-bearing format pod wins); capture the fd before `data()` borrows `d`.
    let data_type = d.type_();
    // fd-backed buffer (MemFd SHM, or DmaBuf)? Capture the fd before `data()` borrows `d`.
    let raw_fd = d.fd();
    // `mapoffset` is where THIS spa_data's region begins inside the fd — non-zero for a pooled
    // producer that carves every buffer out of one shared fd. PipeWire's own MAP_BUFFERS slice
    // already starts there, but our self-mmap below maps the fd from 0, so that path must add it
    // (see `region_off`). Reading it without it made the self-mmap path index the WRONG buffer,
    // and the `needed > avail` guard could not catch that: `avail` came from the whole-fd
    // mapping, so it was large enough for any single buffer's span.
    let map_off = d.as_raw().mapoffset as usize;
    let (size, chunk_off, stride) = {
        let c = d.chunk();
        (
            c.size() as usize,
            c.offset() as usize,
            c.stride().max(0) as usize,
        )
    };
    let Some(fmt) = ud.format else { return }; // unsupported/not negotiated
                                               // The de-pad below assumes ONE packed plane of `bytes_per_pixel` bytes. `bytes_per_pixel`'s
                                               // catch-all answers 4 for NV12, so a producer-native NV12 buffer (stride ≈ w, two planes)
                                               // computed `row = 4w` and always tripped the `stride < row` guard below — blaming the
                                               // PRODUCER's stride for a host limitation. NV12 cannot be de-padded on this path at all
                                               // (the second plane is not even in `datas[0]`'s span), so say so honestly. It only arrives
                                               // here if a native-NV12 negotiation happens without the zero-copy passthrough that is
                                               // supposed to consume it, which is a host bug, not a producer one.
    if matches!(fmt, PixelFormat::Nv12) {
        warn_once(
            "negotiated producer-native NV12 but this capture fell back to the CPU de-pad path, \
             which handles single-plane packed formats only — frames dropped (the NV12 offer is \
             only valid under the raw-dmabuf passthrough that imports it directly)",
        );
        return;
    }
    let bpp = fmt.bytes_per_pixel();
    let row = w * bpp;
    let stride = if stride == 0 { row } else { stride };
    if stride < row {
        warn_once("chunk stride < row — frames dropped");
        return;
    }
    let needed = stride * (h - 1) + row;
    // dmabuf chunks commonly report size 0; fall back to the computed span.
    let size = if size == 0 { needed } else { size };
    // For fd-backed buffers (MemFd SHM, DmaBuf) mmap the fd OURSELVES, sized to the fd's real
    // length (fstat), rather than trusting PipeWire's MAP_BUFFERS slice: xdg-desktop-portal-wlr
    // hands MemFd buffers whose reported `data.maxsize` exceeds the bytes actually mapped into
    // our process, so reading to maxsize segfaults (it also covers the original case — MAP_BUFFERS
    // not mapping Vulkan dmabufs, e.g. gamescope). The `needed > avail` guard below then drops
    // cleanly if the real buffer is genuinely too small. MemPtr buffers (no fd) are same-process —
    // trust `d.data()`.
    let fd_len = if raw_fd > 0 {
        // SAFETY: `libc::stat` is a C plain-old-data struct for which all-zero is a valid value, so
        // `mem::zeroed()` is a sound initializer. `raw_fd` is the buffer's fd (`> 0` checked here) and
        // valid for this callback; `fstat` writes metadata into `&mut st`, a live, aligned,
        // correctly-sized stack `stat` that outlives the synchronous call. `st.st_size` is read only
        // after the return value is confirmed `== 0`. `st` is a fresh local, so nothing aliases it.
        unsafe {
            let mut st: libc::stat = std::mem::zeroed();
            (libc::fstat(raw_fd as i32, &mut st) == 0 && st.st_size > 0)
                .then_some(st.st_size as usize)
        }
    } else {
        None
    };
    let _mapping; // keeps a manual mmap alive for the copy below
                  // Prefer our own fstat-sized mmap of the fd; fall back to PipeWire's MAP_BUFFERS slice
                  // (and finally drop) so an fd PipeWire could map but we can't never silently over-reads.
                  //
                  // `fd_len` is REQUIRED, not preferred: it used to fall back to
                  // `offset + needed`, i.e. a length invented from producer-controlled geometry.
                  // That defeated the whole point of the fstat (the "buffer smaller than the
                  // frame span" guard compares against exactly the number the producer just
                  // supplied) and could map — and then read — past the end of the object, which
                  // is a SIGBUS, not an `Err`. Without a real length we decline to self-map and
                  // let PipeWire's own slice serve, which is bounded by construction.
    let self_mapped: Option<&[u8]> = if raw_fd > 0 {
        match fd_len.and_then(|map_len| DmabufMap::new(raw_fd as i32, map_len)) {
            Some(m) => {
                _mapping = m;
                // SAFETY: `_mapping` is the `DmabufMap` just stored; its `ptr`/`len` come from a
                // successful `mmap` of `map_len` PROT_READ bytes, so `ptr` is non-null, page-aligned,
                // and the VMA is one allocated object of `len` bytes valid for reads. In the common
                // path `map_len == fd_len` (the fd's real size from `fstat`), so the mapping spans the
                // whole object; the de-pad copy below is further bounded by the `offset <= buf.len()`
                // and `needed > avail` guards. The `&[u8]` borrows `_mapping`, which lives to the end
                // of `consume_frame`, so the slice never outlives the mapping, and the memory is only
                // read here, so there is no aliasing/mutation.
                Some(unsafe { std::slice::from_raw_parts(_mapping.ptr as *const u8, _mapping.len) })
            }
            None => None,
        }
    } else {
        None
    };
    // Which base `chunk.offset` is relative to differs by path: our self-mmap starts at fd
    // offset 0, so this spa_data's region begins at `mapoffset`; PipeWire's MAP_BUFFERS slice
    // already begins there. Checked add — both halves are producer-controlled.
    let (buf, region_off): (&[u8], usize) = if let Some(b) = self_mapped {
        match map_off.checked_add(chunk_off) {
            Some(off) => (b, off),
            None => {
                warn_once("mapoffset + chunk offset overflows — frames dropped");
                return;
            }
        }
    } else if let Some(data) = d.data() {
        (data, chunk_off)
    } else {
        warn_once("buffer has no mappable data — frames dropped");
        return;
    };
    // Need stride*(h-1)+row valid bytes within [region_off, region_off+size).
    if region_off > buf.len() {
        return;
    }
    let avail = buf.len() - region_off;
    {
        // One-time geometry dump — makes a new compositor/GPU's buffer layout visible in the
        // logs (the kind of mismatch that crashed xdpw MemFd capture before the self-mmap fix).
        use std::sync::atomic::{AtomicBool, Ordering};
        static ONCE: AtomicBool = AtomicBool::new(true);
        if ONCE.swap(false, Ordering::Relaxed) {
            tracing::info!(
                stride, size, chunk_off, map_off, region_off, buf_len = buf.len(), needed,
                data_type = ?data_type, fd_len = ?fd_len, self_mapped = self_mapped.is_some(),
                "capture CPU de-pad geometry (first frame)"
            );
        }
    }
    if needed > avail || needed > size {
        warn_once("buffer smaller than frame span — frames dropped");
        return;
    }
    let region = &buf[region_off..region_off + size.min(avail)];
    // De-pad into a tightly-packed buffer (chunk stride may exceed w*bpp).
    let mut tight = vec![0u8; row * h];
    for y in 0..h {
        tight[y * row..y * row + row].copy_from_slice(&region[y * stride..y * stride + row]);
    }
    // Cursor-as-metadata: blit the latched pointer into the frame (no-op when hidden or when
    // the layout isn't packed RGB). This is the CPU path's counterpart to the producer's
    // hardware cursor plane, which stays out of the captured buffer.
    composite_cursor(&mut tight, w, h, fmt, &ud.cursor);
    let pts_ns = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64)
        .unwrap_or(0);
    let frame = CapturedFrame {
        width: w as u32,
        height: h as u32,
        pts_ns,
        format: fmt,
        payload: FramePayload::Cpu(tight),
        // Already composited inline into `tight` above — nothing for the encoder to blend.
        cursor: None,
    };
    // Overwrite whatever the consumer has not taken yet — never block the pipewire loop.
    ud.publish(frame);
}

#[allow(clippy::too_many_arguments)]
pub fn pipewire_thread(
    fd: Option<OwnedFd>,
    node_id: u32,
    // The frame mailbox + its wakeup sender (see `super::FrameSlot`): publishing OVERWRITES,
    // so a stalled consumer costs the intermediate frames, never the freshest one.
    slot: super::FrameSlot,
    wake: SyncSender<()>,
    // Every flag/slot this thread PUBLISHES, in one struct (see `super::CaptureSignals`). It used to
    // be seven separate parameters here, seven `Arc` clones at the call site, and the same seven
    // fields restated on `UserData`.
    signals: super::CaptureSignals,
    // THE zero-copy negotiation decision, resolved once by `spawn_pipewire` (which consumes the
    // same struct for the capturer's timeout diagnosis) — never re-derived here.
    plan: NegotiationPlan,
    // The session's capture options (see `super::CaptureOpts`) — `want_444`/`want_hdr` pick the pod
    // family, `expect_exact_dims` arms the sacrificial-mode gate. `allow_zerocopy` is already folded
    // into `plan`.
    opts: super::CaptureOpts,
    preferred: Option<(u32, u32, u32)>,
    quit_rx: pw::channel::Receiver<()>,
    // Encode-backend facts resolved by the facade (never re-derived here) — the one-way
    // capture→encode edge (plan §W6).
    policy: ZeroCopyPolicy,
) -> Result<()> {
    let super::CaptureOpts {
        want_444,
        want_hdr,
        expect_exact_dims,
        ..
    } = opts;
    crate::pwinit::ensure_init();

    let mainloop = pw::main_loop::MainLoopRc::new(None).context("pw MainLoop")?;
    // A quit signal (capturer `Drop`) lands here on the loop thread and stops `run()` so the
    // thread unwinds instead of blocking to process exit. Hold the attachment for the loop's
    // life; the cloned loop handle is the one the callback quits.
    let quit_loop = mainloop.clone();
    let _quit_attach = quit_rx.attach(mainloop.loop_(), move |()| {
        tracing::debug!("pipewire: quit signal received — stopping capture loop");
        quit_loop.quit();
    });
    let context = pw::context::ContextRc::new(&mainloop, None).context("pw Context")?;
    // A portal source hands us an fd to a (sandboxed) PipeWire remote; the KWin
    // virtual-output source has no fd — its node lives on the user's default daemon.
    let core = match fd {
        Some(fd) => context
            .connect_fd_rc(fd, None)
            .context("pw connect_fd (portal remote)")?,
        None => context
            .connect_rc(None)
            .context("pw connect (default daemon)")?,
    };

    // The negotiation decision arrives pre-resolved in `plan` (see `negotiation_plan`): what to
    // build, which offer to make, and whether to prefer producer NV12. Nothing here re-derives
    // any of it — that duplication is what F1/L3 was.
    let backend_is_vaapi = policy.backend_is_vaapi;
    let force_shm = plan.force_shm;
    let vaapi_passthrough = plan.vaapi_passthrough;
    let prefer_native_nv12 = plan.prefer_native_nv12;
    // Build the GPU importer — normally the ISOLATED worker process
    // (design/zerocopy-worker-isolation.md), so a driver fault on a dying compositor's dmabuf
    // kills the worker, not this host. If construction fails, log and fall back to the CPU path
    // (we simply won't request dmabuf below); `plan.build_importer` already encodes WHEN to try
    // at all (not under a raw passthrough, not for HDR, not once repeated worker deaths latched
    // the import off — see `negotiation_plan`).
    if plan.gpu_import_latched {
        tracing::warn!(
            "zero-copy GPU import disabled after repeated import-worker deaths — using CPU path"
        );
    }
    let mut importer = if plan.build_importer {
        match pf_zerocopy::Importer::new_for_capture() {
            Ok(i) => Some(i),
            Err(e) => {
                tracing::warn!(error = %format!("{e:#}"), "zero-copy import unavailable — using CPU path");
                None
            }
        }
    } else {
        None
    };
    if prefer_native_nv12 {
        tracing::info!(
            "zero-copy: preferring gamescope producer-side NV12 LINEAR DMA-BUF (no host \
             RGB CSC; PUNKTFUNK_PIPEWIRE_NV12=0 restores the packed-RGB negotiation)"
        );
    }
    // Modifiers our import stack handles for BGRx: the EGL-importable (tiled) set, plus LINEAR
    // (0) — NVIDIA's EGL won't list it, but LINEAR dmabufs (gamescope's only offer) import via
    // CUDA external memory instead. For the VAAPI passthrough path we advertise LINEAR only:
    // radeonsi/iHD import it and any compositor can allocate it.
    let mut modifiers = importer
        .as_mut()
        .map(|i| i.supported_modifiers(pf_frame::drm_fourcc(PixelFormat::Bgrx).unwrap()))
        .unwrap_or_default();
    if (importer.is_some() || vaapi_passthrough) && !modifiers.contains(&0) {
        modifiers.push(0); // DRM_FORMAT_MOD_LINEAR
    }
    // PyroWave passthrough: the encoder imports through Vulkan, not libva — extend the
    // advertisement with every modifier its device samples from, so compositors that
    // never allocate LINEAR (Mutter+NVIDIA) still negotiate zero-copy dmabufs. The modifiers
    // were resolved by the facade (`ZeroCopyPolicy::pyrowave_modifiers`) — non-empty only when
    // the host's `pyrowave` feature is on AND the session (or the global encoder pref) is
    // PyroWave — so capture never calls back into `encode` and needs no feature gate of its
    // own (the emptiness check gates it).
    if vaapi_passthrough && !policy.pyrowave_modifiers.is_empty() {
        for &m in &policy.pyrowave_modifiers {
            if !modifiers.contains(&m) {
                modifiers.push(m);
            }
        }
        tracing::info!(
            count = modifiers.len(),
            "zero-copy: advertising the PyroWave device's Vulkan-importable dmabuf modifiers"
        );
    }
    // The one runtime-dependent half of the decision (see `NegotiationPlan::want_dmabuf`): it
    // needs the modifier list the importer's construction actually yielded.
    let want_dmabuf = plan.want_dmabuf(importer.is_some(), &modifiers);
    if force_shm {
        tracing::info!(
            "capture: PUNKTFUNK_FORCE_SHM — race-free SHM download path (no dmabuf, no zero-copy)"
        );
    } else if plan.raw_dmabuf_latched {
        tracing::warn!(
            "zero-copy raw-dmabuf passthrough disabled for this host process (repeated encoder \
             import failures, or a previous dmabuf negotiation timeout) — capturing CPU frames \
             instead"
        );
    } else if !want_dmabuf && (plan.build_importer || plan.vaapi_passthrough) {
        tracing::warn!("zero-copy: no importable dmabuf modifiers — using CPU path");
    } else if vaapi_passthrough {
        // The raw-passthrough advertisement. Covers the PyroWave case too: its extra
        // Vulkan-importable modifiers were appended (and logged) just above, so this arm must
        // NOT be gated on `pyrowave_modifiers.is_empty()` — that gate is what dropped a fully
        // zero-copy PyroWave session through to the CPU-path warning below (L11).
        tracing::info!(
            native_nv12_preferred = prefer_native_nv12,
            modifier_count = modifiers.len(),
            pyrowave_extended = !policy.pyrowave_modifiers.is_empty(),
            "zero-copy: advertising DMA-BUF modifiers for direct encoder import (LINEAR \
             always; native NV12 first when enabled, packed RGB fallback)"
        );
    } else if want_dmabuf {
        tracing::info!(
            count = modifiers.len(),
            sample = ?&modifiers[..modifiers.len().min(6)],
            "zero-copy: advertising EGL-importable dmabuf modifiers"
        );
    } else if backend_is_vaapi && policy.backend_is_gpu {
        // Reached only when no dmabuf is advertised at all (every arm above rules out a
        // zero-copy path), so this genuinely IS the CPU capture path: a VAAPI session then pays
        // three full-frame CPU touches (mmap de-pad + swscale RGB→NV12 + surface upload) —
        // make the silent fallback visible.
        // The `raw_dmabuf_latched` arm above catches the latched downgrade, so by here zero-copy
        // is off at the source: the env var, or the session's own output format.
        tracing::warn!(
            "VAAPI encode with the CPU capture path (per-frame de-pad + swscale CSC + \
             upload) — zero-copy is off for this capture ({}); clear PUNKTFUNK_ZEROCOPY to \
             restore the dmabuf default",
            if std::env::var_os("PUNKTFUNK_ZEROCOPY").is_some() {
                "PUNKTFUNK_ZEROCOPY is set falsy"
            } else {
                "this session's output format asked for CPU frames"
            }
        );
    }
    if want_dmabuf && !vaapi_passthrough && want_444 {
        tracing::info!(
            "4:4:4 zero-copy: tiled dmabufs convert to planar YUV444 (BT.709) on the GPU — \
             NVENC fed native full-chroma YUV, no CPU pixel path"
        );
    } else if want_dmabuf && !vaapi_passthrough && pf_zerocopy::nv12_enabled() {
        tracing::info!(
            "PUNKTFUNK_NV12: tiled dmabufs convert to NV12 (BT.709 limited) on the GPU — NVENC \
             fed native YUV (no internal RGB→YUV CSC)"
        );
    }

    let data = UserData {
        info: VideoInfoRaw::default(),
        format: None,
        modifier: 0,
        slot,
        wake,
        signals,
        import_fail_streak: 0,
        importer,
        vaapi_passthrough,
        nv12: pf_zerocopy::nv12_enabled(),
        yuv444: want_444,
        linear_nv12_failed: false,
        dbg_log_n: 0,
        cursor: CursorState::default(),
        expect_dims: if expect_exact_dims {
            preferred.map(|(w, h, _)| (w, h))
        } else {
            None
        },
        gate_skips: 0,
        gate_since: None,
    };

    let stream = pw::stream::StreamBox::new(
        &core,
        "punktfunk-screencast",
        properties! {
            *pw::keys::MEDIA_TYPE     => "Video",
            *pw::keys::MEDIA_CATEGORY => "Capture",
            *pw::keys::MEDIA_ROLE     => "Screen",
            // Never let the session manager re-target this stream to a different node when
            // its target goes away: an orphaned stream auto-linked to a fresh Video/Source
            // wedges that node — and a stuck link head-blocks the PipeWire daemon's shared
            // work queue, stalling ALL new link negotiation system-wide.
            "node.dont-reconnect"     => "true",
        },
    )
    .context("pw Stream")?;

    let _listener = stream
        .add_local_listener_with_user_data(data)
        .state_changed(|_stream, ud, old, new| {
            tracing::info!(?old, ?new, "pipewire stream state");
            // Track whether the node is actively producing. A live source sits in `Streaming`
            // (a static desktop just sends no buffers); anything else — `Paused`/`Unconnected`/
            // `Error` — means the source went away (compositor died, virtual output removed on a
            // Gaming↔Desktop switch). `try_latest` turns a sustained non-Streaming state into a
            // capture-loss so the encode loop rebuilds instead of freezing on the last frame.
            ud.signals.streaming.store(
                matches!(new, pw::stream::StreamState::Streaming),
                Ordering::Relaxed,
            );
        })
        .param_changed(|_stream, ud, id, param| {
            let Some(param) = param else { return };
            if id != pw::spa::param::ParamType::Format.as_raw() {
                return;
            }
            let Ok((media_type, media_subtype)) =
                pw::spa::param::format_utils::parse_format(param)
            else {
                return;
            };
            if media_type != pw::spa::param::format::MediaType::Video
                || media_subtype != pw::spa::param::format::MediaSubtype::Raw
            {
                return;
            }
            if ud.info.parse(param).is_ok() {
                ud.signals.negotiated.store(true, Ordering::Relaxed);
                // A (re)negotiation replaces the buffer pool: every cached per-buffer import
                // (stored fds in the worker, the Vulkan bridge's per-fd sources) keys on
                // buffers that no longer exist — and a recycled fd number/inode must never
                // resolve to a stale import. No-op on the first negotiation (empty caches).
                if let Some(imp) = ud.importer.as_mut() {
                    imp.clear_cache();
                }
                let sz = ud.info.size();
                // Publish the negotiated size for the gamescope cursor source's root→frame
                // scaling (`xfixes_cursor::scale_to_frame`); a renegotiation updates it.
                ud.signals.frame_size.store(
                    (u64::from(sz.width) << 32) | u64::from(sz.height),
                    Ordering::Relaxed,
                );
                ud.format = map_format(ud.info.format());
                ud.modifier = ud.info.modifier();
                // HDR: the 10-bit PQ formats are only ever offered with MANDATORY BT.2020/PQ
                // colorimetry props, so a 10-bit negotiation IS an HDR negotiation — but log
                // what the producer actually fixated for diagnosis.
                let hdr = ud.format.is_some_and(|f| f.is_hdr_rgb10());
                ud.signals.hdr_negotiated.store(hdr, Ordering::Relaxed);
                tracing::info!(
                    width = sz.width,
                    height = sz.height,
                    spa_format = ?ud.info.format(),
                    mapped = ?ud.format,
                    modifier = ud.modifier,
                    hdr,
                    transfer_function = ud.info.transfer_function(),
                    color_primaries = ud.info.color_primaries(),
                    "pipewire format negotiated"
                );
                if ud.format.is_none() {
                    tracing::error!(
                        spa_format = ?ud.info.format(),
                        "negotiated a pixel format the encoder cannot consume — frames will be skipped"
                    );
                }
            }
        })
        .process(|stream, ud| {
            // Latest-frame-only (OBS pattern): Mutter delivers buffers in bursts and recycles its
            // pool; an older queued buffer carries a STALE frame. Drain all queued buffers, requeue
            // the older ones, keep only the newest. This dequeue/requeue runs OUTSIDE the
            // `catch_unwind` below — they are non-panicking C FFI pointer ops, and `newest` is
            // requeued exactly once AFTER the panic-containing region. Previously the whole thing was
            // inside the catch, so a caught panic (in `update_cursor_meta`/`consume_frame`) stranded
            // `newest` forever, permanently shrinking the stream's fixed pool until capture wedged.
            // SAFETY: `stream` is the live stream PipeWire passes into this `.process` callback on the
            // loop thread; `dequeue_raw_buffer` returns a stream-owned `*mut pw_buffer` or null
            // (null-checked), single-threaded so no concurrent access.
            let mut newest = unsafe { stream.dequeue_raw_buffer() };
            if newest.is_null() {
                return;
            }
            let mut drained = 1u32;
            loop {
                // SAFETY: same stream/loop-thread contract; returns the next stream-owned buffer or null.
                let next = unsafe { stream.dequeue_raw_buffer() };
                if next.is_null() {
                    break;
                }
                // SAFETY: `newest` was dequeued from this stream and not yet requeued; we immediately
                // overwrite it, so the requeued pointer is never touched again.
                unsafe { stream.queue_raw_buffer(newest) };
                newest = next;
                drained += 1;
            }
            // Sacrificial-mode gate (kwin.rs `create`): until the producer renegotiates to the
            // expected dims, every buffer — frame AND cursor meta, whose positions are in the
            // doomed mode's space — belongs to the birth mode; consuming one would build the
            // pipeline at the wrong size. Self-disarms on the first matching negotiation, or
            // after `GATE_DEADLINE` without one — degraded dims beat wedging the session into
            // the first-frame-timeout retry loop when the promised renegotiation never comes.
            if let Some((ew, eh)) = ud.expect_dims {
                /// The renegotiation normally lands within a frame or two of recording; well
                /// past that, the producer is not going to deliver it (the on-glass case: the
                /// real mode never actually applied) — stop starving the pipeline.
                const GATE_DEADLINE: std::time::Duration = std::time::Duration::from_secs(3);
                let sz = ud.info.size();
                if sz.width == ew && sz.height == eh {
                    tracing::info!(
                        skipped = ud.gate_skips,
                        width = ew,
                        height = eh,
                        "producer renegotiated to the expected mode — frames flow"
                    );
                    ud.expect_dims = None;
                } else if ud
                    .gate_since
                    .get_or_insert_with(std::time::Instant::now)
                    .elapsed()
                    > GATE_DEADLINE
                {
                    tracing::warn!(
                        negotiated_w = sz.width,
                        negotiated_h = sz.height,
                        expected_w = ew,
                        expected_h = eh,
                        skipped = ud.gate_skips,
                        "producer never renegotiated to the expected mode — accepting its \
                         dims (session runs degraded rather than wedged)"
                    );
                    ud.expect_dims = None;
                } else {
                    ud.gate_skips += 1;
                    if ud.gate_skips == 1 || ud.gate_skips.is_power_of_two() {
                        tracing::info!(
                            negotiated_w = sz.width,
                            negotiated_h = sz.height,
                            expected_w = ew,
                            expected_h = eh,
                            n = ud.gate_skips,
                            "holding frames until the producer renegotiates to the expected mode"
                        );
                    }
                    // SAFETY: `newest` was dequeued from this stream and not yet requeued;
                    // requeued exactly once here, then never touched (mirrors the null path).
                    unsafe { stream.queue_raw_buffer(newest) };
                    return;
                }
            }
            // PipeWire dispatches from a C trampoline with no catch_unwind; a panic crossing that FFI
            // boundary would abort the whole host. Contain the inspect/consume work — the only Rust
            // code here that can panic — and requeue `newest` unconditionally after it.
            let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                // SAFETY: `newest` is the non-null buffer we still own (dequeued, not requeued);
                // `.buffer` is a `*mut spa_buffer` field libpipewire populated. This is a single field
                // load through a valid pointer — no mutation or aliasing.
                let spa_buf = unsafe { (*newest).buffer };

                // Refresh cursor-as-metadata BEFORE the stale-frame skip below: Mutter delivers
                // pointer-only movements as metadata-only "corrupted" buffers we drop for their
                // frame, but their cursor meta is fresh and must still move our overlay.
                update_cursor_meta(&mut ud.cursor, spa_buf);
                // Publish the LIVE overlay (frames or not) so the encode loop's forwarder
                // tracks pointer-only motion on a static desktop — the frame-attached overlay
                // alone stales between damage frames. ONLY when we actually have one: a
                // gamescope node carries no `SPA_META_Cursor`, so `overlay()` is always `None`
                // here, and writing that would clobber — at frame rate — the `Some` the
                // attached XFixes source publishes into this SAME slot, strobing the
                // composited pointer on/off. Portal cursors are `None` only before the first
                // bitmap (nothing to drop), and a HIDDEN pointer is still `Some(visible:false)`.
                if let Some(overlay) = ud.cursor.overlay() {
                    if let Ok(mut slot) = ud.signals.cursor_live.lock() {
                        *slot = Some(overlay);
                    }
                }

                // Inspect the newest buffer's header + first chunk for the diagnostic and the
                // CORRUPTED skip. SPA_META_Header is optional — `hdr` may be null.
                // SAFETY: `spa_buf` is the `*mut spa_buffer` of the buffer we still hold.
                // `spa_buffer_find_meta_data` scans that buffer's metadata array for a `SPA_META_Header`
                // of at least `size_of::<spa_meta_header>()` bytes and returns a pointer into the held
                // buffer's metadata (or null). The size argument matches the struct the result is cast
                // to, and the pointer stays valid as long as the buffer is held (until requeue). Null is
                // handled below.
                let hdr = unsafe {
                    spa::sys::spa_buffer_find_meta_data(
                        spa_buf,
                        spa::sys::SPA_META_Header,
                        std::mem::size_of::<spa::sys::spa_meta_header>(),
                    ) as *const spa::sys::spa_meta_header
                };
                let hdr_flags = if hdr.is_null() {
                    0u32
                } else {
                    // SAFETY: reached only when `hdr` is non-null; it points to a `spa_meta_header`
                    // inside the live buffer's metadata (returned for a size >=
                    // `size_of::<spa_meta_header>()`, so `.flags` is in bounds). A single field read
                    // while the buffer is still held.
                    unsafe { (*hdr).flags }
                };
                // First data chunk's size + flags (used for the diagnostic + CORRUPTED check)
                // and its data type (a dmabuf legitimately reports chunk size 0, so the size-0
                // stale skip only applies to mappable SHM buffers).
                // SAFETY: every dereference is guarded in order before any field read — `spa_buf`
                // non-null, `n_datas > 0`, the `datas` (`*mut spa_data`) array non-null, and the first
                // element's `chunk` (`*mut spa_chunk`) non-null. `d0` is that first `spa_data` and `c`
                // its chunk; reading `(*d0).type_`, `(*c).size`, `(*c).flags` are in-bounds field loads
                // of libspa structs inside the buffer we still hold. Single-threaded loop, no mutation.
                let (chunk_size, chunk_flags, is_dmabuf) = unsafe {
                    if !spa_buf.is_null()
                        && (*spa_buf).n_datas > 0
                        && !(*spa_buf).datas.is_null()
                        && !(*(*spa_buf).datas).chunk.is_null()
                    {
                        let d0 = (*spa_buf).datas;
                        let c = (*d0).chunk;
                        let is_dmabuf =
                            (*d0).type_ == spa::sys::SPA_DATA_DmaBuf;
                        ((*c).size, (*c).flags, is_dmabuf)
                    } else {
                        (0u32, 0i32, false)
                    }
                };

                let corrupted = (hdr_flags & spa::sys::SPA_META_HEADER_FLAG_CORRUPTED) != 0
                    || (chunk_flags & spa::sys::SPA_CHUNK_FLAG_CORRUPTED as i32) != 0;

                // THE GNOME FLASH FIX: skip Mutter's CORRUPTED / size-0 cursor-update buffers.
                // When the pointer moves (e.g. dragging a window) Mutter sends metadata-only
                // buffers flagged CORRUPTED (chunk size 0) that still reference a RECYCLED old
                // frame; consuming them encodes "the window at its old position" — the flash.
                // Confirmed live on worker-3 (chunk_flags=CORRUPTED, size 0) for both the zero-copy
                // and SHM paths. The size-0 half is SHM-only (a real dmabuf legitimately reports
                // chunk size 0). `drained` is the latest-frame-only depth — a cheap extra defense
                // against bursty delivery, though here Mutter sends one buffer per callback.
                if corrupted || (chunk_size == 0 && !is_dmabuf) {
                    ud.dbg_log_n += 1;
                    if ud.dbg_log_n.is_power_of_two() {
                        tracing::debug!(
                            skipped = ud.dbg_log_n,
                            drained,
                            "capture: skipped a stale CORRUPTED/cursor buffer (GNOME)"
                        );
                    }
                    // Skip this stale/cursor buffer — `newest` is requeued unconditionally below.
                    return;
                }

                consume_frame(ud, spa_buf);
            }));
            // Hand `newest` back to the stream exactly once, on EVERY path — normal, corrupted-skip,
            // or a caught panic in the closure above. This single requeue is what keeps the fixed
            // buffer pool from draining.
            // SAFETY: all reads of `spa_buf`/`newest` (update_cursor_meta, consume_frame) completed
            // inside the closure above; `newest` was dequeued from this stream and not yet requeued.
            unsafe { stream.queue_raw_buffer(newest) };
            if outcome.is_err() {
                // In the per-frame `.process` callback: a deterministic panic (e.g. a bad
                // format) would fire this every frame, so power-of-two throttle it — enough to
                // surface the fault without evicting the whole log ring.
                static PANICS: std::sync::atomic::AtomicU64 =
                    std::sync::atomic::AtomicU64::new(0);
                let n = PANICS.fetch_add(1, Ordering::Relaxed) + 1;
                if n.is_power_of_two() {
                    tracing::error!(count = n, "panic in pipewire process callback — frame dropped");
                }
            }
        })
        .register()
        .context("register stream listener")?;

    // Debug knob: offer a single fixed format (PUNKTFUNK_PW_FIXED_POD="WxH") to bisect
    // negotiation failures against a producer's exact EnumFormat (e.g. gamescope).
    let fixed_pod: Option<(u32, u32)> = std::env::var("PUNKTFUNK_PW_FIXED_POD")
        .ok()
        .and_then(|v| v.split_once('x').map(|(w, h)| (w.parse(), h.parse())))
        .and_then(|(w, h)| Some((w.ok()?, h.ok()?)));

    // Request raw video in any encoder-mappable layout, any size/framerate.
    let obj = if let Some((fw, fh)) = fixed_pod {
        tracing::info!(
            fw,
            fh,
            "pipewire: offering a fixed BGRx format pod (PUNKTFUNK_PW_FIXED_POD)"
        );
        pw::spa::pod::object!(
            pw::spa::utils::SpaTypes::ObjectParamFormat,
            pw::spa::param::ParamType::EnumFormat,
            pw::spa::pod::property!(
                pw::spa::param::format::FormatProperties::MediaType,
                Id,
                pw::spa::param::format::MediaType::Video
            ),
            pw::spa::pod::property!(
                pw::spa::param::format::FormatProperties::MediaSubtype,
                Id,
                pw::spa::param::format::MediaSubtype::Raw
            ),
            pw::spa::pod::property!(
                pw::spa::param::format::FormatProperties::VideoFormat,
                Id,
                VideoFormat::BGRx
            ),
            pw::spa::pod::property!(
                pw::spa::param::format::FormatProperties::VideoSize,
                Rectangle,
                pw::spa::utils::Rectangle {
                    width: fw,
                    height: fh
                }
            ),
            pw::spa::pod::property!(
                pw::spa::param::format::FormatProperties::VideoFramerate,
                Fraction,
                pw::spa::utils::Fraction { num: 0, denom: 1 }
            ),
        )
    } else {
        build_default_format_obj(preferred)
    };

    // gamescope trap — the Steam overlay's presence in the stream is decided HERE by omission:
    // gamescope's `paint_pipewire()` composites the overlay (Shift+Tab / Quick Access Menu) into
    // the node it hands us ONLY when the consumer-negotiated `gamescope_focus_appid` is 0 — the
    // default, and the "mirror the focused window + overlay" branch (gamescope ≥ 3.16.23; see
    // `MIN_GAMESCOPE_OVERLAY`). None of the EnumFormat pods below advertise the
    // `SPA_FORMAT_VIDEO_gamescope_focus_appid` property, so gamescope reads 0 and paints the
    // overlay for us for free. DO NOT add a non-zero focus-appid (e.g. to "isolate the game" in a
    // dedicated session) — that flips gamescope into the Remote-Play branch that deliberately
    // drops the overlay (and all host chrome) back out of the capture. The cursor, external
    // overlay (MangoHUD), and notifications are excluded from the node on EVERY gamescope
    // version and are composited host-side instead (see `xfixes_cursor.rs`).
    //
    // When zero-copy is on, offer ONLY a BGRx dmabuf format with our EGL-importable modifiers
    // (offering shm too makes the compositor pick shm). The modifier list goes out as a plain
    // MANDATORY `ChoiceEnum::Enum` and the producer fixates one of the alternatives directly —
    // this is NOT the two-step DONT_FIXATE handshake (libspa 0.9's `ChoiceFlags` cannot express
    // `SPA_POD_PROP_FLAG_DONT_FIXATE`, and `param_changed` only READS the fixated format, it
    // re-emits nothing). Worth revisiting if a multi-modifier offer is ever seen to fail
    // negotiation on a compositor that needs the allocator round-trip. Otherwise offer the
    // multi-format shm pod and let MAP_BUFFERS map it. An HDR session replaces ALL of this with the two 10-bit
    // PQ pods (LINEAR dmabuf, MANDATORY colorimetry — see `build_hdr_dmabuf_format`): offering
    // SDR alongside would make the producer pick its earlier-listed SDR format, and the
    // negotiation-timeout path latches the process-wide SDR downgrade if nothing matches.
    let format_pods: Vec<Vec<u8>> = if want_hdr {
        tracing::info!(
            "HDR capture: offering xRGB_210LE/xBGR_210LE LINEAR dmabufs with MANDATORY \
             BT.2020 + SMPTE-2084 (PQ) colorimetry (GNOME 50+ monitor stream)"
        );
        vec![
            build_hdr_dmabuf_format(VideoFormat::xRGB_210LE, preferred)?,
            build_hdr_dmabuf_format(VideoFormat::xBGR_210LE, preferred)?,
        ]
    } else if want_dmabuf {
        let mut pods = Vec::with_capacity(if prefer_native_nv12 { 2 } else { 1 });
        if prefer_native_nv12 {
            // First compatible consumer pod wins. Gamescope advertises NV12 and BGRx; pinning
            // BT.709 limited here selects its RGB→NV12 shader with our bitstream colorimetry.
            pods.push(build_dmabuf_format(VideoFormat::NV12, &[0], preferred)?);
        }
        pods.push(build_dmabuf_format(
            VideoFormat::BGRx,
            &modifiers,
            preferred,
        )?);
        pods
    } else {
        vec![serialize_pod(obj)?]
    };
    let buffers_values = if want_hdr || want_dmabuf {
        // Dmabuf-only. For HDR this is load-bearing beyond zero-copy: Mutter's SHM record
        // path paints 8-bit ARGB32 regardless of the negotiated format, so a MemFd buffer
        // under a 10-bit format would carry mislabeled bytes.
        Some(build_dmabuf_buffers()?)
    } else if force_shm {
        // True SHM: exclude DmaBuf so Mutter MUST download (glReadPixels orders against render).
        Some(build_shm_only_buffers()?)
    } else {
        // CPU path still accepts mappable dmabufs (gamescope offers only those once its
        // modifier-bearing format pod wins the intersection).
        Some(build_mappable_buffers()?)
    };

    // Ask for cursor-as-metadata on every path (harmless if the producer can't supply it): the
    // pointer rides as SPA_META_Cursor rather than being burned into the frame, so the
    // compositor keeps its cheap hardware cursor plane (see `choose_cursor_mode`).
    let cursor_meta = build_cursor_meta_param()?;
    let mut byte_slices: Vec<&[u8]> = Vec::new();
    for pod in &format_pods {
        byte_slices.push(pod);
    }
    if let Some(b) = &buffers_values {
        byte_slices.push(b);
    }
    byte_slices.push(&cursor_meta);
    let mut params: Vec<&Pod> = byte_slices
        .iter()
        .map(|&b| Pod::from_bytes(b).context("pod from bytes"))
        .collect::<Result<_>>()?;

    stream
        .connect(
            spa::utils::Direction::Input,
            Some(node_id),
            pw::stream::StreamFlags::AUTOCONNECT | pw::stream::StreamFlags::MAP_BUFFERS,
            &mut params,
        )
        .context("pw stream connect")?;

    // Blocks this thread, pumping frame callbacks until the capturer's `Drop` fires the quit
    // channel attached above (`_quit_attach` → `quit_loop.quit()`), at which point `run()`
    // returns and the thread unwinds — releasing the importer / CUDA context deterministically.
    mainloop.run();
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{negotiation_plan, NegotiationInputs};

    /// A healthy NVENC session: zero-copy on, no latches, SDR 4:2:0, non-VAAPI backend.
    fn nvenc() -> NegotiationInputs {
        NegotiationInputs {
            zerocopy: true,
            force_shm: false,
            want_hdr: false,
            want_444: false,
            backend_is_vaapi: false,
            pyrowave_session: false,
            native_nv12_session: false,
            raw_dmabuf_import_disabled: false,
            gpu_import_disabled: false,
            native_nv12_env_on: true,
        }
    }

    /// A gamescope-style VAAPI session that CAN take producer-native NV12.
    fn vaapi_native_nv12() -> NegotiationInputs {
        NegotiationInputs {
            backend_is_vaapi: true,
            native_nv12_session: true,
            ..nvenc()
        }
    }

    /// The four invariants that were prose-only comments in `pipewire_thread`'s prologue.
    #[test]
    fn negotiation_plan_invariants() {
        // 1. HDR NEVER builds the EGL→CUDA importer — its de-tile blit is 8-bit RGBA8, so an
        //    importer here would silently crush the 10-bit depth.
        for want_444 in [false, true] {
            let p = negotiation_plan(NegotiationInputs {
                want_hdr: true,
                want_444,
                ..nvenc()
            });
            assert!(!p.build_importer, "HDR must not build the importer");
        }
        // …on every backend, including the ones that take the raw passthrough.
        assert!(
            !negotiation_plan(NegotiationInputs {
                want_hdr: true,
                ..vaapi_native_nv12()
            })
            .build_importer
        );

        // 2. 4:4:4 never prefers producer NV12 (a 4:4:4 session must not be subsampled).
        let p = negotiation_plan(NegotiationInputs {
            want_444: true,
            ..vaapi_native_nv12()
        });
        assert!(!p.prefer_native_nv12, "4:4:4 must not take NV12");
        // Nor does HDR (no 10-bit NV12 path).
        assert!(
            !negotiation_plan(NegotiationInputs {
                want_hdr: true,
                ..vaapi_native_nv12()
            })
            .prefer_native_nv12
        );

        // 3. Producer-native NV12 needs a `native_nv12_session` AND an active raw passthrough:
        //    libav VAAPI would misread the two-plane buffer, and the CUDA importer expects
        //    packed RGB.
        assert!(negotiation_plan(vaapi_native_nv12()).prefer_native_nv12);
        assert!(
            !negotiation_plan(NegotiationInputs {
                native_nv12_session: false,
                ..vaapi_native_nv12()
            })
            .prefer_native_nv12,
            "a session whose encoder can't ingest NV12 must never be offered it"
        );
        assert!(
            !negotiation_plan(NegotiationInputs {
                force_shm: true,
                ..vaapi_native_nv12()
            })
            .prefer_native_nv12,
            "no passthrough (force_shm) ⇒ no native NV12"
        );
        // A PyroWave session takes the passthrough but its CSC ingests packed RGB only.
        assert!(
            !negotiation_plan(NegotiationInputs {
                pyrowave_session: true,
                ..vaapi_native_nv12()
            })
            .prefer_native_nv12
        );

        // 4. The pyrowave-modifier extension (and the passthrough generally) is off whenever
        //    the raw-dmabuf latch has fired.
        let p = negotiation_plan(NegotiationInputs {
            raw_dmabuf_import_disabled: true,
            ..vaapi_native_nv12()
        });
        assert!(!p.vaapi_passthrough, "latched ⇒ no raw passthrough");
        assert!(!p.prefer_native_nv12);
        assert!(p.raw_dmabuf_latched, "…and the operator gets told why");
    }

    /// The drift F1 was about: `spawn_pipewire` derived `vaapi_passthrough` by hand and its copy
    /// omitted the raw-dmabuf latch, so after that latch fired the capturer believed it had made
    /// the passthrough offer while the thread had already fallen back — and a timeout then
    /// latched a downgrade for an offer nobody made. With one resolver the two cannot disagree,
    /// so this pins the property that made the mirror wrong: the latch MUST move the decision.
    #[test]
    fn the_raw_dmabuf_latch_moves_the_passthrough_decision() {
        for pyrowave in [false, true] {
            let base = NegotiationInputs {
                backend_is_vaapi: !pyrowave,
                pyrowave_session: pyrowave,
                ..nvenc()
            };
            assert!(negotiation_plan(base).vaapi_passthrough);
            assert!(
                !negotiation_plan(NegotiationInputs {
                    raw_dmabuf_import_disabled: true,
                    ..base
                })
                .vaapi_passthrough
            );
        }
    }

    /// A PyroWave session on an NVIDIA box takes the raw passthrough (the wavelet encoder's own
    /// Vulkan device imports dmabufs on any vendor) and therefore must NOT also build the
    /// EGL→CUDA importer — whose payloads only NVENC can consume.
    #[test]
    fn a_pyrowave_session_passes_through_without_a_cuda_importer() {
        let p = negotiation_plan(NegotiationInputs {
            pyrowave_session: true,
            ..nvenc()
        });
        assert!(p.vaapi_passthrough);
        assert!(!p.build_importer);
    }

    /// `force_shm` is the race-free download path: no passthrough, and `want_dmabuf` stays false
    /// even with an importer and a full modifier list.
    #[test]
    fn force_shm_wins_over_every_dmabuf_path() {
        let p = negotiation_plan(NegotiationInputs {
            force_shm: true,
            ..vaapi_native_nv12()
        });
        assert!(!p.vaapi_passthrough);
        assert!(!p.want_dmabuf(true, &[0, 1, 2]));
        // …and an SHM-forced NVENC session may still build the importer (it just won't be fed
        // dmabufs), which is why `want_dmabuf` — not `build_importer` — is the gate.
        let p = negotiation_plan(NegotiationInputs {
            force_shm: true,
            ..nvenc()
        });
        assert!(p.build_importer);
        assert!(!p.want_dmabuf(true, &[0]));
    }

    /// `want_dmabuf` needs a real modifier list: an importer that constructed but advertised
    /// nothing importable falls back to the CPU path.
    #[test]
    fn want_dmabuf_needs_both_a_consumer_and_a_modifier() {
        let p = negotiation_plan(nvenc());
        assert!(p.want_dmabuf(true, &[0]));
        assert!(!p.want_dmabuf(true, &[]), "no modifiers ⇒ CPU path");
        assert!(
            !p.want_dmabuf(false, &[0]),
            "importer failed to construct and no passthrough ⇒ CPU path"
        );
        // The passthrough needs no importer at all.
        let p = negotiation_plan(vaapi_native_nv12());
        assert!(p.want_dmabuf(false, &[0]));
    }

    /// The GPU-import death latch stops the importer being rebuilt, and says so.
    #[test]
    fn the_gpu_import_death_latch_skips_the_importer() {
        let p = negotiation_plan(NegotiationInputs {
            gpu_import_disabled: true,
            ..nvenc()
        });
        assert!(!p.build_importer);
        assert!(p.gpu_import_latched);
        // It is NOT reported for a capture that would never have built one anyway (HDR, or a
        // raw passthrough) — that would misdirect the operator.
        assert!(
            !negotiation_plan(NegotiationInputs {
                gpu_import_disabled: true,
                want_hdr: true,
                ..nvenc()
            })
            .gpu_import_latched
        );
        assert!(
            !negotiation_plan(NegotiationInputs {
                gpu_import_disabled: true,
                ..vaapi_native_nv12()
            })
            .gpu_import_latched
        );
    }

    /// Zero-copy off ⇒ nothing at all: no importer, no passthrough, no NV12 preference.
    #[test]
    fn zerocopy_off_disables_every_branch() {
        for i in [
            NegotiationInputs {
                zerocopy: false,
                ..nvenc()
            },
            NegotiationInputs {
                zerocopy: false,
                ..vaapi_native_nv12()
            },
        ] {
            let p = negotiation_plan(i);
            assert!(!p.build_importer);
            assert!(!p.vaapi_passthrough);
            assert!(!p.prefer_native_nv12);
            assert!(!p.want_dmabuf(false, &[0]));
        }
    }
}