//! 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,
    HDR_FORMAT_ORDER,
};
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,
    /// WP-A3/B3 — which clock the wire's `pts_ns` comes from, and how clean each one is. See
    /// [`crate::pts_provenance`]: the delivery stamp this loop used to take unconditionally is
    /// downstream of the compositor's delivery jitter, so it bakes that jitter into the
    /// timestamps the client plays back from.
    pts: crate::pts_provenance::PtsProvenance,
    /// When the provenance window opened.
    pts_reported: std::time::Instant,
    /// `CLOCK_REALTIME − CLOCK_MONOTONIC`, ns — what carries the compositor's monotonic stamp
    /// into the wire's realtime domain. Re-sampled each reporting window; the two clocks drift
    /// by µs over 30 s, so this is not a per-frame cost.
    rt_minus_mono_ns: i64,
    /// `PUNKTFUNK_CAPTURE_HDR_PTS=0` puts the wire back on the delivery stamp unconditionally.
    hdr_pts_enabled: bool,
    /// PW4 step 1: the producer-fence wait distribution, measured on this (the PipeWire loop)
    /// thread. Per-session, like the fall-through tally.
    fence_wait: FenceWaitStats,
    /// PW5 step 1: the negotiated buffer-pool depth, counted from `add_buffer`/`remove_buffer`.
    /// See [`PoolCensus`] — this is the number that decides whether a deeper encode pipeline is
    /// safe, and until now nobody had it.
    pool: PoolCensus,
    /// Raw-passthrough frames that silently fell through to the CPU de-pad path, by reason — see
    /// [`PassthroughFallbacks`]. Per-session: a fresh `UserData` is built per pipeline, so a
    /// compositor that starts serving dmabufs again after a rebuild gets a fresh log budget.
    passthrough_fallbacks: PassthroughFallbacks,
    /// 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>,
    /// Deferred requeue of raw-passthrough buffers (see [`DeferredRequeue`]): the encode thread
    /// reads the dmabuf long after `.process` returns, so the buffer must not rejoin the
    /// producer's pool until the frame's [`BufferHold`] drops.
    defer: std::sync::Arc<DeferredRequeue>,
}

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(());
    }

    /// Withhold the raw-passthrough buffer from the producer's pool until the returned hold
    /// drops — the deferred requeue that closes the rewrite-while-the-encoder-reads race.
    /// `None` (pool too shallow, or `PUNKTFUNK_ZEROCOPY_HOLD=0`) falls back to the immediate
    /// `.process`-epilogue requeue, i.e. the old racy contract; said once per session.
    fn try_defer(&mut self, pw_buf: *mut pw::sys::pw_buffer) -> Option<pf_frame::FrameHold> {
        if !zerocopy_hold_enabled() {
            return None;
        }
        let buf = pw_buf as usize;
        let pool_live = self.pool.live;
        let generation = self.defer.book.lock().ok()?.try_hold(buf, pool_live);
        let Some(generation) = generation else {
            if !self.defer.logged_shallow.swap(true, Ordering::Relaxed) {
                tracing::warn!(
                    pool_depth = pool_live,
                    reserve = HOLD_POOL_RESERVE,
                    "zero-copy: the producer's buffer pool cannot spare a buffer to hold across \
                     the encode — falling back to the immediate requeue, which the producer may \
                     rewrite mid-encode (torn/discolored frames under load); PUNKTFUNK_FORCE_SHM=1 \
                     trades CPU for a race-free capture if artifacts appear"
                );
            }
            return None;
        };
        if !self.defer.logged_active.swap(true, Ordering::Relaxed) {
            tracing::info!(
                pool_depth = pool_live,
                reserve = HOLD_POOL_RESERVE,
                "zero-copy: withholding each published buffer from the producer until the \
                 encoder releases it (deferred requeue — the producer can no longer rewrite a \
                 frame mid-encode); PUNKTFUNK_ZEROCOPY_HOLD=0 restores the immediate requeue"
            );
        }
        Some(std::sync::Arc::new(BufferHold {
            defer: self.defer.clone(),
            buf,
            generation,
        }))
    }
}

/// 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,
    /// `pf_zerocopy::gpu_dmabuf_negotiation_disabled()` — a previous EGL→CUDA dmabuf-only offer
    /// timed out (the compositor accepts none of the importer's modifiers).
    pub gpu_dmabuf_negotiation_failed: bool,
    /// `PUNKTFUNK_PIPEWIRE_NV12` (default ON) — allow the producer-side NV12 preference.
    pub native_nv12_env_on: bool,
    /// [`ZeroCopyPolicy::hdr_cuda_ok`] — the resolved encoder can ingest a packed 10-bit PQ CUDA
    /// payload. Only the direct-SDK NVENC backend can.
    pub hdr_cuda_ok: 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 a latch fired —
    /// repeated import-worker deaths, or a previous dmabuf-offer negotiation timeout.
    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 takes the TILED EGL de-tile blit (it renders into an 8-bit `GL_RGBA8` texture
///      → silent depth loss). It may still build the importer, because the HDR pod family
///      advertises LINEAR only ([`build_hdr_dmabuf_format`]) — so an HDR dmabuf necessarily
///      takes the Vulkan-bridge / CUDA-external-memory arm, which is byte-exact for any 4 Bpp
///      packed format. The per-frame gate in `.process` enforces the tiled half.
///   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. `gpu_import_disabled` is
    // the repeated-worker-death latch (a wedged GPU stack must not crash-loop);
    // `gpu_dmabuf_negotiation_failed` is the offer's own timeout latch (a compositor that accepts
    // none of the importer's modifiers refuses them identically on every retry, so the next
    // session negotiates the CPU path instead of re-paying the 10 s timeout).
    //
    // HDR is NOT excluded outright (invariant 1): its pods are LINEAR-only, so it lands on the
    // Vulkan-bridge arm, never the 8-bit de-tile blit. But it is excluded where the encoder cannot
    // take a packed 10-bit CUDA payload — the libav fallback's HDR route swscales into a P010
    // hardware frame, so it must keep getting CPU frames. Without that term a
    // `PUNKTFUNK_NVENC_DIRECT=0` host would stream garbage.
    let build_importer = i.zerocopy
        && !raw_passthrough
        && !i.gpu_import_disabled
        && !i.gpu_dmabuf_negotiation_failed
        && (!i.want_hdr || i.hdr_cuda_ok);
    // 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,
        // Every `build_importer` term EXCEPT the two latches, and then either latch — i.e. exactly
        // "this capture would have built the importer, but a latch stopped it".
        gpu_import_latched: i.zerocopy
            && !raw_passthrough
            && (!i.want_hdr || i.hdr_cuda_ok)
            && (i.gpu_import_disabled || i.gpu_dmabuf_negotiation_failed),
    }
}

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
    }
}

/// Which capture arm a negotiated pipeline actually resolved to.
///
/// The 2026-08-08 PyroWave triage had to reconstruct this from four files, because no single line
/// ever states it: the arm is the product of a policy, a latch, an importer that may or may not
/// have constructed, and a modifier list. A degraded host and a healthy one logged the same
/// thing. [`resolved_capture_arm`] plus the one INFO line at pipeline build is the whole fix.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum CaptureArm {
    /// Raw dmabufs handed straight to the encoder, which imports them itself — libva (VAAPI) or
    /// the PyroWave encoder's own Vulkan device. No host pixel touch.
    DmabufPassthrough,
    /// dmabufs imported to CUDA device buffers by the EGL→CUDA worker, for NVENC. No host pixel
    /// touch either, but a different failure surface (the worker, the modifier negotiation).
    CudaImport,
    /// CPU frames: an mmap de-pad of every frame, then whatever CSC + upload the encoder needs.
    /// The slow path — always a downgrade when the consumer could have taken a dmabuf.
    Cpu,
}

impl CaptureArm {
    pub(super) fn as_str(self) -> &'static str {
        match self {
            CaptureArm::DmabufPassthrough => "dmabuf-passthrough",
            CaptureArm::CudaImport => "cuda-import",
            CaptureArm::Cpu => "cpu",
        }
    }
}

/// Resolve the arm this pipeline ended up on. **Pure** — `have_importer` and `want_dmabuf` are the
/// two runtime facts `negotiation_plan` cannot know (whether the importer constructed, and what
/// modifier list that yielded); everything else is already in the plan.
pub(super) fn resolved_capture_arm(
    plan: &NegotiationPlan,
    have_importer: bool,
    want_dmabuf: bool,
) -> CaptureArm {
    if !want_dmabuf {
        // No dmabuf offer at all: SHM/CPU frames, whatever the plan wanted.
        CaptureArm::Cpu
    } else if plan.vaapi_passthrough {
        CaptureArm::DmabufPassthrough
    } else if have_importer {
        CaptureArm::CudaImport
    } else {
        // Unreachable via `want_dmabuf` (it requires `have_importer || vaapi_passthrough`), but
        // stated rather than `unreachable!()`: a logging helper must never be the thing that
        // panics a capture thread.
        CaptureArm::Cpu
    }
}

/// Who consumes the captured frames — the fact that decides whether a CPU arm is a *downgrade*
/// worth warning about, and what to call it in the log.
///
/// Derived from the resolved [`ZeroCopyPolicy`](crate::ZeroCopyPolicy), **not** from the encoder
/// pref: `pyrowave_session` is per-session (the negotiated codec), so a PyroWave session on an
/// otherwise-NVENC host reads as PyroWave here. Naming the pref instead is exactly how a PyroWave
/// session's CPU downgrade came to be reported as an NVENC one — or, on an NVIDIA host, not
/// reported at all.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum ConsumerKind {
    /// This session encodes PyroWave: the wavelet encoder's own Vulkan device imports dmabufs on
    /// any vendor, so a CPU arm costs it the passthrough it was designed around.
    PyroWave,
    /// The VAAPI backend (AMD/Intel): libva imports the dmabuf and CSCs on the GPU.
    Vaapi,
    /// NVENC, fed by the EGL→CUDA importer.
    Nvenc,
    /// The software encoder — CPU frames are its native input, so a CPU arm is no downgrade.
    Software,
}

impl ConsumerKind {
    pub(super) fn as_str(self) -> &'static str {
        match self {
            ConsumerKind::PyroWave => "pyrowave",
            ConsumerKind::Vaapi => "vaapi",
            ConsumerKind::Nvenc => "nvenc",
            ConsumerKind::Software => "software",
        }
    }

    /// Whether landing on [`CaptureArm::Cpu`] is a performance downgrade for this consumer (i.e.
    /// worth a `warn!`). True for every GPU consumer; false for the software encoder, which wants
    /// CPU frames anyway.
    pub(super) fn cpu_is_downgrade(self) -> bool {
        !matches!(self, ConsumerKind::Software)
    }
}

/// Classify the frames' consumer. **Pure.** `pyrowave_session` wins over `backend_is_vaapi`
/// because it is the per-session truth and the pref is host-global (a PyroWave session also flips
/// `backend_is_vaapi` on, via `linux_zero_copy_is_vaapi`'s `Pyrowave` arm — so testing vaapi first
/// would swallow every PyroWave session).
pub(super) fn consumer_kind(
    pyrowave_session: bool,
    backend_is_vaapi: bool,
    backend_is_gpu: bool,
) -> ConsumerKind {
    if pyrowave_session {
        ConsumerKind::PyroWave
    } else if !backend_is_gpu {
        ConsumerKind::Software
    } else if backend_is_vaapi {
        ConsumerKind::Vaapi
    } else {
        ConsumerKind::Nvenc
    }
}

/// Why a frame on the raw-dmabuf passthrough could not be handed to the encoder and fell through
/// to the CPU de-pad path instead.
///
/// Each variant is a *different* diagnosis with a different fix, and all four were silent: the
/// passthrough block simply fell out of its `if` and the frame took the slow path, so a session
/// that had negotiated zero-copy could pay CPU costs on every frame while logging a healthy
/// "advertising DMA-BUF modifiers" line at open.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum PassthroughFallback {
    /// No format negotiated yet — transient, and expected around a renegotiation.
    NoFormat,
    /// The producer delivered an SHM/MemFd buffer for this frame, not a dmabuf.
    NotDmabuf,
    /// The negotiated pixel format has no DRM fourcc, so it cannot be described to the encoder.
    NoFourcc,
    /// `F_DUPFD_CLOEXEC` failed — the fd could not be duplicated to outlive the buffer recycle
    /// (an fd-limit symptom, not a graphics one).
    DupFailed,
}

impl PassthroughFallback {
    fn bit(self) -> u8 {
        match self {
            PassthroughFallback::NoFormat => 1 << 0,
            PassthroughFallback::NotDmabuf => 1 << 1,
            PassthroughFallback::NoFourcc => 1 << 2,
            PassthroughFallback::DupFailed => 1 << 3,
        }
    }

    pub(super) fn as_str(self) -> &'static str {
        match self {
            PassthroughFallback::NoFormat => "no format negotiated yet",
            PassthroughFallback::NotDmabuf => "the producer delivered an SHM/MemFd buffer",
            PassthroughFallback::NoFourcc => "the negotiated format has no DRM fourcc",
            PassthroughFallback::DupFailed => "F_DUPFD_CLOEXEC failed on the dmabuf fd",
        }
    }

    /// What actually happens to the frame. Three of the four reasons downgrade it to the CPU
    /// de-pad path; `NoFormat` does not — the CPU path needs `ud.format` too and returns without
    /// de-padding, so that frame is DROPPED. Worth the distinction: "slower" and "gone" are
    /// different faults, and a diagnostic that conflates them is the thing PW2 exists to remove.
    pub(super) fn falls_back_to_cpu(self) -> bool {
        !matches!(self, PassthroughFallback::NoFormat)
    }

    /// What to do about it — the half a log line is useless without.
    pub(super) fn hint(self) -> &'static str {
        match self {
            PassthroughFallback::NoFormat => {
                "harmless if it stops: the first buffers can arrive before param_changed"
            }
            PassthroughFallback::NotDmabuf => {
                "the compositor accepted the dmabuf offer and is serving memory anyway — check \
                 PUNKTFUNK_FORCE_SHM and the compositor's allocator"
            }
            PassthroughFallback::NoFourcc => {
                "a capture format the encoder path cannot describe — file it, the negotiation \
                 should not have accepted it"
            }
            PassthroughFallback::DupFailed => "out of file descriptors — raise the host's NOFILE",
        }
    }
}

/// Upper bounds (µs) of the fence-wait histogram's buckets; the last bucket is everything above.
///
/// Deliberately coarse and log-ish. The question this instrument exists to answer is not "what is
/// the wait, to the microsecond" but "**is the tail ~0, or is it milliseconds?**" — the first
/// answer retires PW4 as a comment correction, the second justifies moving the wait off the
/// PipeWire loop thread. Bucket edges are placed so those two worlds cannot be confused: anything
/// at or below 100 µs is noise, anything past 1 ms is a real stall on a 60 Hz budget of 16.6 ms.
const FENCE_WAIT_BUCKETS_US: [u64; 6] = [100, 500, 1_000, 2_000, 5_000, 10_000];

/// PW4 step 1: the distribution of the producer's implicit-fence wait, measured **on the PipeWire
/// loop thread**, which is exactly where it is expensive — that thread is the compositor's
/// consumer, so time spent blocked here delays buffer recycling for the NEXT frame.
///
/// PW4 is pre-registered to be **abandoned on evidence**: if the p99 sits in the first bucket the
/// wait is already ~free and the package becomes a comment correction. Shipping the instrument
/// before the change is the whole point — the alternative is moving load-bearing synchronisation
/// off a thread on a hunch.
///
/// One caveat the reader needs: measure this AFTER the priority levers land. The win case for
/// moving the wait is a loaded GPU, which is the same scenario PW1 targets, so a measurement taken
/// before PW1 would hand PW4 credit for PW1's problem.
#[derive(Debug, Default, Clone, Copy)]
pub(super) struct FenceWaitStats {
    samples: u64,
    total_us: u64,
    max_us: u64,
    /// One more than the bucket bounds: the overflow bucket.
    buckets: [u64; FENCE_WAIT_BUCKETS_US.len() + 1],
    /// Outcome split — a `NoFence` majority means the wait is structurally free on this producer
    /// (nothing to wait for), which is a different finding from "the wait is short".
    signaled: u64,
    no_fence: u64,
    timed_out: u64,
    failed: u64,
}

impl FenceWaitStats {
    /// Record one wait. `bucket_of` is inlined here rather than exposed: the histogram is only
    /// ever read through [`summary`](Self::summary).
    pub(super) fn record(&mut self, us: u64) {
        self.samples += 1;
        self.total_us += us;
        self.max_us = self.max_us.max(us);
        let idx = FENCE_WAIT_BUCKETS_US
            .iter()
            .position(|&b| us <= b)
            .unwrap_or(FENCE_WAIT_BUCKETS_US.len());
        self.buckets[idx] += 1;
    }

    /// The bucket the `q`-quantile falls in, as its upper bound in µs — `None` for the overflow
    /// bucket (i.e. "worse than the last edge"). Counting up to the quantile rather than
    /// interpolating keeps this honest about what a histogram can actually say.
    pub(super) fn quantile_bucket_us(&self, q: f64) -> Option<Option<u64>> {
        if self.samples == 0 {
            return None;
        }
        // The index of the sample at `q`, 0-based, so q=1.0 picks the last sample.
        let target = ((self.samples as f64) * q).ceil().max(1.0) as u64;
        let mut seen = 0u64;
        for (i, &count) in self.buckets.iter().enumerate() {
            seen += count;
            if seen >= target {
                return Some(FENCE_WAIT_BUCKETS_US.get(i).copied());
            }
        }
        Some(None)
    }

    pub(super) fn mean_us(&self) -> u64 {
        self.total_us.checked_div(self.samples).unwrap_or(0)
    }

    /// Whether enough has been seen for the p99 to mean anything. 100 frames is under two seconds
    /// at 60 fps and is the point where one outlier stops dominating the answer.
    pub(super) fn is_meaningful(&self) -> bool {
        self.samples >= 100
    }
}

/// PW5 stage 1: how many buffers the producer actually allocated for this stream.
///
/// **Nothing in this codebase had ever counted them.** The zero-copy path used to hand the SPA
/// buffer straight back to the producer at `.process` return, leaving pool depth as the only
/// thing keeping capture untorn. The deferred requeue ([`DeferredRequeue`]) now withholds
/// published buffers until the consumer is done, but the depth still matters twice over: it is
/// the budget `HoldBook::try_hold` spends (a pool of ≤ [`HOLD_POOL_RESERVE`] cannot defer at
/// all and runs the old race), and for un-deferred frames it remains the race window.
///
/// `live` is maintained by the `add_buffer`/`remove_buffer` stream callbacks, which PipeWire fires
/// on the loop thread as the pool is allocated (and again, remove-then-add, on a renegotiation that
/// replaces it). There is no "pool complete" event, so the count is published from `.process`: by
/// the time the first buffer is dequeued the allocation has finished.
#[derive(Debug, Default, Clone, Copy)]
pub(super) struct PoolCensus {
    /// Buffers currently in the pool (adds minus removes).
    live: u32,
    /// Deepest `live` seen this session — the number a depth decision must key on, since a
    /// renegotiation can transiently shrink the pool to zero.
    high_water: u32,
    /// The `live` value already logged, so a stable pool logs exactly one line per distinct depth
    /// (a renegotiation that changes the depth is worth a second line; 240 frames a second of the
    /// same number is not).
    logged: Option<u32>,
}

impl PoolCensus {
    fn add(&mut self) {
        self.live += 1;
        self.high_water = self.high_water.max(self.live);
    }

    fn remove(&mut self) {
        self.live = self.live.saturating_sub(1);
    }

    /// Called from `.process`, once per buffer. Returns `Some(live)` the first time each distinct
    /// depth is seen — the caller logs then and only then.
    fn note_frame(&mut self) -> Option<u32> {
        (self.logged != Some(self.live)).then(|| {
            self.logged = Some(self.live);
            self.live
        })
    }
}

/// Per-session tally of raw-passthrough frames that fell through to the CPU path, with a one-line
/// budget per distinct reason.
///
/// Rate-limiting is what makes this shippable: `.process` runs per frame, so an unconditional log
/// would flood at the capture rate. Per *reason* rather than per session, because the four reasons
/// diagnose different faults and a transient `NoFormat` at open must not spend the budget a
/// persistent `NotDmabuf` needs.
#[derive(Debug, Default, Clone, Copy)]
pub(super) struct PassthroughFallbacks {
    frames: u64,
    logged: u8,
}

impl PassthroughFallbacks {
    /// Record one fall-through. Returns `Some(frames_so_far)` the FIRST time each distinct reason
    /// is seen this session — the caller logs then and only then, so at most four lines per
    /// session regardless of frame rate.
    pub(super) fn note(&mut self, reason: PassthroughFallback) -> Option<u64> {
        self.frames += 1;
        let bit = reason.bit();
        (self.logged & bit == 0).then(|| {
            self.logged |= bit;
            self.frames
        })
    }
}

/// 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;

/// Buffers the deferred requeue always leaves in the producer's pool. One for the frame the
/// producer is rendering right now, one in transit — withholding past that would make the
/// producer skip frames whenever our holds are at their worst (host frame + up to two encoder
/// ring slots), which is a pacing hiccup, not corruption, but there is no reason to court it.
const HOLD_POOL_RESERVE: u32 = 2;

/// `PUNKTFUNK_ZEROCOPY_HOLD=0` restores the immediate `.process`-return requeue (the racy
/// pre-hold behavior) — a field bisect lever, not a tuning knob. `env_on` grammar like every
/// other capture knob (a bare `== "0"` compare is the trap `PUNKTFUNK_FORCE_SHM` already fell in).
fn zerocopy_hold_enabled() -> bool {
    static ON: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
    *ON.get_or_init(|| pf_host_config::env_on("PUNKTFUNK_ZEROCOPY_HOLD").unwrap_or(true))
}

/// Pure bookkeeping for the deferred requeue: which buffers are currently withheld from the
/// producer, each under a per-hold generation so a pointer-value reuse across a pool
/// renegotiation can never satisfy a stale hold's release (see `complete`).
///
/// Threading contract (what makes the single-requeue invariant hold with no atomics): entries are
/// INSERTED (`try_hold`) and REMOVED (`complete` via the requeue channel's callback, `purge` via
/// `remove_buffer`) only on the PipeWire loop thread; a dropping [`BufferHold`] on any other
/// thread only *sends* the release message. So between a hold's creation and the loop servicing
/// its release, `contains` is stable — which is exactly what the `.process` epilogue relies on to
/// decide "requeue now" vs "the hold owns the requeue".
#[derive(Default)]
struct HoldBook {
    /// Withheld buffers: `*mut pw_buffer` as usize → the generation of the hold that owns it.
    out: std::collections::HashMap<usize, u64>,
    /// Last issued hold generation (monotonic per stream).
    last_gen: u64,
}

impl HoldBook {
    /// Withhold `buf` if the pool can spare it: at most `pool_live - HOLD_POOL_RESERVE` buffers
    /// out at once. Returns the generation to release with, or `None` (pool too shallow / buffer
    /// somehow already out — the caller falls back to the immediate requeue).
    fn try_hold(&mut self, buf: usize, pool_live: u32) -> Option<u64> {
        let cap = pool_live.saturating_sub(HOLD_POOL_RESERVE) as usize;
        if self.out.len() >= cap || self.out.contains_key(&buf) {
            return None;
        }
        self.last_gen += 1;
        self.out.insert(buf, self.last_gen);
        Some(self.last_gen)
    }

    /// A hold released: take `buf` out of the book iff this generation still owns it. `true` ⇒
    /// the caller must requeue the buffer; `false` ⇒ the entry was purged (pool renegotiated —
    /// the pointer may even be a NEW buffer under a reused address) and the buffer must NOT be
    /// touched.
    fn complete(&mut self, buf: usize, generation: u64) -> bool {
        match self.out.get(&buf) {
            Some(&g) if g == generation => {
                self.out.remove(&buf);
                true
            }
            _ => false,
        }
    }

    /// `remove_buffer`: the buffer is being freed under us (renegotiation/teardown) — forget it.
    /// Its hold's later release finds the generation gone and becomes a no-op.
    fn purge(&mut self, buf: usize) {
        self.out.remove(&buf);
    }

    fn contains(&self, buf: usize) -> bool {
        self.out.contains_key(&buf)
    }
}

/// Shared between the loop thread ([`HoldBook`] ops) and the [`BufferHold`] guards riding
/// published frames to the encode thread.
struct DeferredRequeue {
    book: std::sync::Mutex<HoldBook>,
    /// Wakes the loop to requeue `(buffer, generation)`. Send failure = the loop (and with it
    /// the stream and every buffer) is gone — nothing to release.
    tx: pw::channel::Sender<(usize, u64)>,
    /// One-per-session lines: the first successful defer, and the shallow-pool fallback.
    logged_active: std::sync::atomic::AtomicBool,
    logged_shallow: std::sync::atomic::AtomicBool,
}

/// The concrete [`pf_frame::FrameHold`]: releases its buffer back to the producer when the last
/// clone drops. Send-only from the dropping thread — the actual `pw_stream_queue_buffer` runs in
/// the requeue channel's loop-thread callback.
struct BufferHold {
    defer: std::sync::Arc<DeferredRequeue>,
    buf: usize,
    generation: u64,
}

impl Drop for BufferHold {
    fn drop(&mut self) {
        let _ = self.defer.tx.send((self.buf, self.generation));
    }
}

/// 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.
///
/// `pw_buf` is the buffer's `pw_buffer` handle (`spa_buf`'s owner), used only as the identity a
/// raw-passthrough publish withholds via [`UserData::try_defer`] — never dereferenced here.
/// How often the wire-pts provenance line is emitted (WP-A3). Matches the audio plane's stats
/// cadence so a field log reads as one timeline.
const PTS_REPORT_EVERY: std::time::Duration = std::time::Duration::from_secs(30);

/// `CLOCK_REALTIME − CLOCK_MONOTONIC`, ns.
///
/// PipeWire stamps `spa_meta_header.pts` in the graph's clock domain (`CLOCK_MONOTONIC`); the wire
/// — and the client's plausibility gate — speak realtime-since-epoch. Sampling the pair back to
/// back is what carries one into the other, and the residual is the few hundred nanoseconds
/// between the two reads against a 50 ms plausibility window. A failed read reports 0, which puts
/// every rebased stamp outside that window and falls the whole stream back to delivery stamps —
/// the safe direction.
fn realtime_minus_monotonic_ns() -> i64 {
    let rt = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos() as i64)
        .unwrap_or(0);
    let mut ts = libc::timespec {
        tv_sec: 0,
        tv_nsec: 0,
    };
    // SAFETY: `clock_gettime` writes one `timespec` through the pointer and touches nothing else;
    // `ts` is a live, properly aligned local. A non-zero return leaves it untouched.
    if unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) } != 0 {
        return 0;
    }
    rt - (ts.tv_sec * 1_000_000_000 + ts.tv_nsec)
}

fn consume_frame(
    ud: &mut UserData,
    spa_buf: *mut spa::sys::spa_buffer,
    pw_buf: *mut pw::sys::pw_buffer,
    hdr_pts_ns: Option<i64>,
) {
    // 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
    }

    // ONE stamp for this frame, whichever of the three paths below publishes it (WP-B3). Each
    // used to take its own `SystemTime::now()` at the moment it happened to reach the publish, so
    // a CPU de-pad's milliseconds landed INSIDE the timestamp and the three paths could drift
    // apart without anything saying so. Sampled here, before any of that work, the stamp
    // describes the frame's arrival rather than our handling of it.
    let delivery_ns = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64)
        .unwrap_or(0);
    let hdr_pts_ns = hdr_pts_ns.filter(|&p| p > 0);
    // Measured on EVERY capture, whether or not the compositor's stamp is the one we ship: the
    // question "which clock is cleaner on this host" is what decides that, and a diagnostic that
    // only runs once you already trust the answer cannot inform it.
    ud.pts.observe(hdr_pts_ns, delivery_ns);
    let stamp = crate::pts_provenance::wire_pts(
        ud.hdr_pts_enabled.then_some(hdr_pts_ns).flatten(),
        delivery_ns,
        ud.rt_minus_mono_ns,
    );
    let pts_ns = stamp.pts_ns;
    if ud.hdr_pts_enabled && hdr_pts_ns.is_some() && !stamp.from_header {
        ud.pts.implausible += 1;
    }
    if ud.pts_reported.elapsed() >= PTS_REPORT_EVERY {
        if let Some(r) = ud.pts.report() {
            tracing::info!(
                frames = r.frames,
                with_hdr = r.with_hdr,
                samples = r.samples,
                period_us = r.period_us,
                // THE pair this whole work package exists to compare. Materially tighter on the
                // left ⇒ the compositor's stamp is worth shipping; equally ragged ⇒ the producer
                // composes irregularly and no choice of stamp can fix it.
                hdr_mad_us = r.hdr_mad_us,
                delivery_mad_us = r.delivery_mad_us,
                offset_p50_ms = r.offset_p50_ms,
                implausible = r.implausible,
                hdr_pts_used = ud.hdr_pts_enabled,
                "capture wire-pts provenance"
            );
        }
        ud.pts.reset_window();
        ud.pts_reported = std::time::Instant::now();
        // The two clocks drift by µs over a window; re-pairing them here keeps the rebase honest
        // over a multi-hour session without costing anything per frame.
        ud.rt_minus_mono_ns = realtime_minus_monotonic_ns();
    }

    // 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.
    //
    // MEASURED 2026-08-08 (Wave-2 PW4, which proposed moving this off the loop thread and was
    // pre-registered to be abandoned if the wait was already free — it is):
    //   gamescope + NVIDIA (RTX 5070 Ti)   outcome=NoFence
    //   Mutter    + NVIDIA (RTX 5070 Ti)   outcome=NoFence
    //   gamescope + RADV   (Deck VANGOGH)  300 samples, ALL NoFence, mean 23us, max 48us,
    //                                      p50 and p99 both in the <=100us bucket
    //   KWin      + RADV   (Deck desktop)  no implicit fence either
    // That is EVERY compositor × vendor combination this fleet has, and not one of them attaches an
    // implicit fence — so this is one ioctl and a return, not a block: there is nothing to wait on,
    // and moving it to the consumer side buys nothing. The 100 ms budget stays as a guard for a
    // producer that DOES fence, which is a real thing even if nothing here does it. The histogram
    // below is how to re-check if that ever changes: run with PUNKTFUNK_PERF=1 and read the p99
    // bucket.
    if datas[0].type_() == pw::spa::buffer::DataType::DmaBuf {
        // PW4 step 1: time the wait. Two `Instant::now()` per frame on a path that is already
        // making a syscall — and this is the measurement that decides whether PW4 ships at all.
        let t0 = std::time::Instant::now();
        let waited = pf_zerocopy::dmabuf_fence::wait_read_ready(datas[0].fd(), 100);
        ud.fence_wait.record(t0.elapsed().as_micros() as u64);
        match waited {
            Ok(outcome) => {
                use pf_zerocopy::dmabuf_fence::WaitOutcome;
                match outcome {
                    WaitOutcome::Signaled => ud.fence_wait.signaled += 1,
                    WaitOutcome::NoFence => ud.fence_wait.no_fence += 1,
                    WaitOutcome::TimedOut => ud.fence_wait.timed_out += 1,
                }
                static F1: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(true);
                if F1.swap(false, Ordering::Relaxed) {
                    tracing::info!(
                        ?outcome,
                        "dmabuf implicit-fence sync active (Signaled → driver fences the \
                         render, race closed; NoFence → no implicit fence, zero-copy may \
                         still show stale frames; TimedOut → fence pending past 100ms, \
                         proceeded anyway)"
                    );
                }
            }
            Err(e) => {
                ud.fence_wait.failed += 1;
                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)"
                    );
                }
            }
        }
        // One line per ~5 s at 60 fps, under PUNKTFUNK_PERF only — the same gate and cadence the
        // encode backends use for their submit splits, so a perf run reads as one instrument.
        if pf_host_config::config().perf
            && ud.fence_wait.is_meaningful()
            && ud.fence_wait.samples % 300 == 0
        {
            let q = |p: f64| match ud.fence_wait.quantile_bucket_us(p) {
                Some(Some(us)) => format!("<={us}us"),
                Some(None) => format!(
                    ">{}us",
                    FENCE_WAIT_BUCKETS_US[FENCE_WAIT_BUCKETS_US.len() - 1]
                ),
                None => "n/a".to_string(),
            };
            tracing::info!(
                samples = ud.fence_wait.samples,
                mean_us = ud.fence_wait.mean_us(),
                max_us = ud.fence_wait.max_us,
                p50 = %q(0.50),
                p99 = %q(0.99),
                signaled = ud.fence_wait.signaled,
                no_fence = ud.fence_wait.no_fence,
                timed_out = ud.fence_wait.timed_out,
                failed = ud.fence_wait.failed,
                "dmabuf implicit-fence wait on the PipeWire loop thread (PW4: a p99 in the first \
                 bucket means this wait is already free and moving it off-thread buys nothing)"
            );
        }
    }

    // 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.
    //
    // The block below either publishes a dmabuf and RETURNS, or breaks with the reason it could
    // not — so every non-success exit is named and counted instead of silently falling out of
    // three nested `if`s into the CPU path, which is how a session that had negotiated zero-copy
    // could pay a full CPU pixel touch per frame while logging nothing but a healthy open.
    if ud.vaapi_passthrough {
        let reason = 'passthrough: {
            let Some(fmt) = ud.format else {
                break 'passthrough PassthroughFallback::NoFormat;
            };
            if datas[0].type_() != pw::spa::buffer::DataType::DmaBuf {
                break 'passthrough PassthroughFallback::NotDmabuf;
            }
            let Some(fourcc) = pf_frame::drm_fourcc(fmt) else {
                break 'passthrough PassthroughFallback::NoFourcc;
            };
            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)",
                    );
                    // Not a fall-through: this frame is DROPPED, not downgraded (de-padding it as
                    // linear would stream scrambled chroma), so it is not counted below.
                    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 read window comes from the deferred
            // requeue below (`try_defer` — the producer does not get this buffer back until
            // the frame's hold drops); with no hold (shallow pool / PUNKTFUNK_ZEROCOPY_HOLD=0)
            // it falls back to the compositor's pool depth outrunning the encode, the old
            // racy contract.
            // 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 {
                break 'passthrough PassthroughFallback::DupFailed;
            }
            let hold = ud.try_defer(pw_buf);
            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,
                    hold,
                }),
                // 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;
        };
        // The passthrough declined this frame. Say so ONCE per distinct reason (`.process` runs
        // per frame — see `PassthroughFallbacks`), carrying the running count so a persistent
        // downgrade is distinguishable from a one-frame hiccup at renegotiation.
        if let Some(frames) = ud.passthrough_fallbacks.note(reason) {
            tracing::warn!(
                frames,
                "zero-copy raw-dmabuf passthrough did not take this frame: {} — {} ({})",
                reason.as_str(),
                if reason.falls_back_to_cpu() {
                    "it falls back to the CPU capture path, costing a full-resolution mmap \
                     de-pad plus the encoder's own upload on every such frame"
                } else {
                    "the frame is DROPPED — the CPU de-pad path needs the negotiated format \
                     too, so nothing streams while this persists"
                },
                reason.hint()
            );
        }
    }

    // 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) {
        // Invariant 1's teeth: a 10-bit PQ frame may take the LINEAR (Vulkan-bridge → CUDA
        // external memory) arm, which moves 4 Bpp words verbatim, but must NEVER take the TILED
        // EGL de-tile blit — that renders into an 8-bit `GL_RGBA8` texture and would crush the
        // depth silently. The HDR pods advertise LINEAR only, so a tiled modifier here means the
        // producer ignored the offer; drop to the CPU path rather than trust it.
        let hdr_tiled = fmt.is_hdr_rgb10() && ud.modifier != 0;
        if hdr_tiled {
            warn_once(
                "HDR frame arrived with a tiled modifier — the GPU de-tile blit is 8-bit, so \
                 this stream falls back to the CPU path (the producer ignored our LINEAR-only \
                 HDR offer)",
            );
        }
        if datas[0].type_() == pw::spa::buffer::DataType::DmaBuf && !hdr_tiled {
            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).
                // A 10-bit frame takes NEITHER convert: both the GL and the Vulkan compute CSCs
                // write 8-bit planes, and NVENC ingests the packed 10-bit RGB natively
                // (`ARGB10`/`ABGR10`) with its own BT.2020 CSC. So HDR stays packed RGB all the
                // way to the encoder — no depth loss, no extra pass.
                let ten_bit = fmt.is_hdr_rgb10();
                let yuv444 = ud.yuv444 && modifier.is_some() && !ten_bit;
                let mut nv12 = ud.nv12 && !ud.yuv444 && !ten_bit;
                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)"
                            );
                        }
                        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 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,
        cursor_id0_hides,
        ..
    } = 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 once repeated worker deaths latched the import
    // off — see `negotiation_plan`).
    if plan.gpu_import_latched {
        tracing::warn!(
            "zero-copy GPU import disabled for this host process (repeated import-worker deaths, \
             or a previous dmabuf negotiation timeout) — 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, enumerated PER FOURCC. `XR24` (BGRx) and `AR24` (BGRA)
    // are asked separately on purpose: EGL/libva answer per format, and nothing entitles us to
    // assume a driver that imports one imports the other. Keeping them apart is also what makes
    // the BGRA pod below correct on AMD and Intel rather than an NVIDIA-shaped guess — each list
    // is whatever THIS GPU's stack actually said.
    //
    // To each list we add 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 there is
    // no importer at all, so the lists start empty and LINEAR is all we advertise: radeonsi/iHD
    // import it and any compositor can allocate it.
    let mut modifiers = Vec::new();
    let mut modifiers_bgra = Vec::new();
    if let Some(i) = importer.as_mut() {
        modifiers = i.supported_modifiers(pf_frame::drm_fourcc(PixelFormat::Bgrx).unwrap());
        modifiers_bgra = i.supported_modifiers(pf_frame::drm_fourcc(PixelFormat::Bgra).unwrap());
    }
    // 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).
    let extend_pyrowave = vaapi_passthrough && !policy.pyrowave_modifiers.is_empty();
    for list in [&mut modifiers, &mut modifiers_bgra] {
        if (importer.is_some() || vaapi_passthrough) && !list.contains(&0) {
            list.push(0); // DRM_FORMAT_MOD_LINEAR
        }
        if extend_pyrowave {
            for &m in &policy.pyrowave_modifiers {
                if !list.contains(&m) {
                    list.push(m);
                }
            }
        }
    }
    if extend_pyrowave {
        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);
    // Record whether THIS capture really advertises the EGL→CUDA dmabuf-only offer, for the
    // capturer's negotiation-timeout diagnosis (its latch must fire only for an offer that was
    // actually made — `plan.build_importer` alone can't know the importer constructed).
    signals.gpu_dmabuf_offer.store(
        want_dmabuf && !vaapi_passthrough && !want_hdr,
        Ordering::Relaxed,
    );
    // The ONE line that says which arm this session actually got, and who consumes it. Everything
    // below explains a particular arm; nothing below states the answer. Reconstructing it from the
    // detail lines cost the 2026-08-08 PyroWave triage four files, and on the one arm that matters
    // most — a PyroWave session downgraded to CPU on an NVIDIA host — there was no detail line to
    // reconstruct it from at all (see the final `else if` of this chain).
    let consumer = consumer_kind(
        policy.pyrowave_session,
        backend_is_vaapi,
        policy.backend_is_gpu,
    );
    let arm = resolved_capture_arm(&plan, importer.is_some(), want_dmabuf);
    tracing::info!(
        capture_arm = arm.as_str(),
        consumer = consumer.as_str(),
        modifier_count = modifiers.len(),
        // PW3(c): the latch state belongs on the same line as the arm. A `cpu` arm has two very
        // different explanations — "this host was never going to do dmabuf" and "something failed
        // earlier and we are still living with the verdict" — and only the second is a bug worth
        // chasing. Reading it here also means the retry/clear behaviour is observable rather than
        // inferred.
        raw_dmabuf_latch = pf_zerocopy::raw_dmabuf_latch_state(),
        "capture pipeline resolved: {} → {}",
        arm.as_str(),
        consumer.as_str()
    );
    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!(
            bgrx_count = modifiers.len(),
            bgra_count = modifiers_bgra.len(),
            // `sample` is TRUNCATED to 6, and LINEAR is pushed last — so reading the sample as the
            // whole list makes a perfectly good offer look tiled-only. That misreading cost a full
            // debugging session on 2026-08-14, hence stating the one bit that was actually wanted.
            linear_offered = modifiers.contains(&0),
            sample = ?&modifiers[..modifiers.len().min(6)],
            "zero-copy: advertising EGL-importable dmabuf modifiers (BGRx + BGRA pods)"
        );
    } else if consumer.cpu_is_downgrade() {
        // 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: the consumer then pays
        // full-frame CPU touches (mmap de-pad + whatever CSC/upload it needs) —
        // 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.
        //
        // The gate used to be `backend_is_vaapi && backend_is_gpu`, which is why a PyroWave
        // session's CPU downgrade was invisible on an NVIDIA/auto host: `backend_is_vaapi` reads
        // the host-global encoder pref, so a per-session PyroWave negotiation there is `false` and
        // fell out of the chain logging NOTHING. `consumer_kind` asks the per-session question
        // instead, and excludes only the software encoder (which wants CPU frames).
        tracing::warn!(
            consumer = consumer.as_str(),
            "{} encode with the CPU capture path (per-frame de-pad + CSC + upload) — \
             zero-copy is off for this capture ({}); set PUNKTFUNK_ZEROCOPY=1 to restore the \
             dmabuf default",
            consumer.as_str(),
            if std::env::var_os("PUNKTFUNK_ZEROCOPY").is_some() {
                "PUNKTFUNK_ZEROCOPY is set falsy"
            } else if want_hdr && !policy.hdr_cuda_ok {
                // Reachable and NOT the output format's doing: `build_importer` drops an HDR
                // capture whose encoder cannot take a packed 10-bit CUDA payload (libav's HDR
                // route swscales into a P010 hardware frame). Naming the output format here
                // would send the reader hunting the wrong knob.
                "this HDR session's encoder cannot ingest a 10-bit CUDA payload, so the capture \
                 stays on CPU frames"
            } 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)"
        );
    }

    // Deferred requeue (the rewrite-while-encoding fix): holds riding published frames release
    // their buffers through this channel from whatever thread drops them last; the receiver —
    // attached to the loop below, after the stream exists — is the single place a withheld
    // buffer rejoins the producer's pool.
    let (requeue_tx, requeue_rx) = pw::channel::channel::<(usize, u64)>();
    let defer = std::sync::Arc::new(DeferredRequeue {
        book: std::sync::Mutex::new(HoldBook::default()),
        tx: requeue_tx,
        logged_active: std::sync::atomic::AtomicBool::new(false),
        logged_shallow: std::sync::atomic::AtomicBool::new(false),
    });

    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,
        pts: crate::pts_provenance::PtsProvenance::new(),
        pts_reported: std::time::Instant::now(),
        rt_minus_mono_ns: realtime_minus_monotonic_ns(),
        hdr_pts_enabled: std::env::var("PUNKTFUNK_CAPTURE_HDR_PTS").as_deref() != Ok("0"),
        fence_wait: FenceWaitStats::default(),
        pool: PoolCensus::default(),
        passthrough_fallbacks: PassthroughFallbacks::default(),
        cursor: CursorState::new(cursor_id0_hides),
        expect_dims: if expect_exact_dims {
            preferred.map(|(w, h, _)| (w, h))
        } else {
            None
        },
        gate_skips: 0,
        gate_since: None,
        defer: defer.clone(),
    };

    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;
            }
            // Parse ONCE — `parse` takes `&mut self` — and report a failure instead of swallowing it.
            // On `Err`, `negotiated` stays false and `format`/`modifier`/`frame_size` keep their
            // previous values, so the capture dies on the generic "the compositor offered no format
            // this consumer accepts" timeout — sending the operator hunting a format mismatch when
            // the real fault was a malformed Format pod we DID accept.
            let parsed = ud.info.parse(param);
            if let Err(e) = &parsed {
                tracing::error!(
                    error = %e,
                    "pipewire: failed to parse the negotiated Format pod — capture will time out \
                     with no usable format"
                );
            }
            if parsed.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"
                    );
                }
            }
        })
        // PW5 stage 1 — the pool census. PipeWire fires these on the loop thread as it allocates
        // (and, on a renegotiation, frees then re-allocates) the stream's buffers. The census only
        // counts; `remove_buffer` additionally purges the buffer from the deferred-requeue book —
        // the buffer is being freed under any hold still riding a frame, so that hold's later
        // release must become a no-op (the generation check in `HoldBook::complete` also covers
        // the freed address being reused by a new pool's buffer).
        .add_buffer(|_stream, ud, _buf| ud.pool.add())
        .remove_buffer(|_stream, ud, buf| {
            ud.pool.remove();
            if let Ok(mut book) = ud.defer.book.lock() {
                book.purge(buf as usize);
            }
        })
        .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;
            }
            // PW5 stage 1: publish the depth the producer actually negotiated, once per distinct
            // value. MEASURED, not requested: `build_dmabuf_buffers` asks for a range and the
            // producer picks — this line is the only place the picked number is visible.
            //
            // Why it matters beyond curiosity: the depth is the budget the deferred requeue
            // (`HoldBook::try_hold`) spends withholding published buffers from the producer
            // while the encoder reads them. A pool of ≤ HOLD_POOL_RESERVE cannot defer at all —
            // those sessions run the old contract, where a requeued buffer may be rewritten
            // mid-encode and only pool depth keeps frames untorn.
            if let Some(depth) = ud.pool.note_frame() {
                tracing::info!(
                    pool_depth = depth,
                    high_water = ud.pool.high_water,
                    drained,
                    "pipewire buffer pool negotiated — the producer's ACTUAL count \
                     (add_buffer/remove_buffer): the deferred-requeue budget, and the rewrite \
                     window for any frame published without a hold"
                );
            }
            // 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 }
                };
                // The compositor's OWN stamp for this frame (WP-A3/B3), stamped upstream of the
                // delivery jitter our `SystemTime::now()` cannot see past. Located here because
                // the header already is; whether it is worth shipping is what the provenance
                // line measures.
                let hdr_pts = if hdr.is_null() {
                    None
                } else {
                    // SAFETY: as for `.flags` — non-null, from a lookup that demanded at least
                    // `size_of::<spa_meta_header>()` bytes (so `.pts` is in bounds), read while
                    // the buffer is still held.
                    Some(unsafe { (*hdr).pts })
                };
                // 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, newest, hdr_pts);
            }));
            // Hand `newest` back to the stream exactly once, on EVERY path — normal, corrupted-skip,
            // or a caught panic in the closure above — UNLESS a raw-passthrough publish withheld it
            // (`try_defer` put it in the hold book): then the requeue duty belongs to the frame's
            // `BufferHold`, and requeueing here too would hand the producer the same buffer twice.
            // The book is stable across this check: only this thread removes entries (the requeue
            // channel's callback / `remove_buffer`), and neither can run inside `.process` — a
            // consumer racing the frame to its drop merely queues the release message. A panic
            // AFTER the publish leaves the hold live on the published frame, so skipping the
            // immediate requeue remains correct on that path too.
            let withheld = ud
                .defer
                .book
                .lock()
                .map(|b| b.contains(newest as usize))
                .unwrap_or(false);
            if !withheld {
                // SAFETY: all reads of `spa_buf`/`newest` (update_cursor_meta, consume_frame)
                // completed inside the closure above; `newest` was dequeued from this stream,
                // not yet requeued, and — per the `withheld` check — carries no hold that would
                // requeue it a second time.
                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")?;

    // The deferred-requeue service. A `BufferHold` dropping on any thread only *sends*
    // `(buffer, generation)`; this callback — on the loop thread, like every other stream op —
    // is where a withheld buffer actually rejoins the producer's pool. `HoldBook::complete`
    // makes a release for a renegotiated-away buffer (or a freed address reused by a new
    // pool's buffer) a no-op, so a stale hold can never queue somebody else's buffer.
    let defer_cb = defer.clone();
    let stream_ptr = stream.as_raw_ptr() as usize;
    let _requeue_attach = requeue_rx.attach(mainloop.loop_(), move |(buf, generation)| {
        let requeue = defer_cb
            .book
            .lock()
            .map(|mut b| b.complete(buf, generation))
            .unwrap_or(false);
        if requeue {
            // SAFETY: `complete` returned true ⇒ this buffer was withheld by exactly this hold
            // and no `remove_buffer` has freed it since (that purges the book), so the pointer
            // is a live buffer of this stream that we own (dequeued, never requeued). The
            // stream outlives this attached receiver (declared after it, dropped before it),
            // and the loop stops dispatching once `run()` returns.
            let _ = unsafe {
                pw::sys::pw_stream_queue_buffer(
                    stream_ptr as *mut pw::sys::pw_stream,
                    buf as *mut pw::sys::pw_buffer,
                )
            };
        }
    });

    // 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 xBGR_210LE/xRGB_210LE LINEAR dmabufs with MANDATORY \
             BT.2020 + SMPTE-2084 (PQ) colorimetry (GNOME 50+ monitor stream)"
        );
        // ⚠ Order is the whole fix — see the NVIDIA note on `HDR_FORMAT_ORDER`. The first
        // compatible consumer pod wins, so this is what a gamescope session actually lands on.
        HDR_FORMAT_ORDER
            .iter()
            .map(|fmt| build_hdr_dmabuf_format(*fmt, preferred))
            .collect::<Result<Vec<_>>>()?
    } else if want_dmabuf {
        let mut pods = Vec::with_capacity(if prefer_native_nv12 { 3 } else { 2 });
        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)?);
        }
        if !modifiers.is_empty() {
            pods.push(build_dmabuf_format(
                VideoFormat::BGRx,
                &modifiers,
                preferred,
            )?);
        }
        // xdph (Hyprland/sway) offers ONLY **BGRA** on its dmabuf EnumFormat — it lists BGRA *and*
        // BGRx on the SHM pod, so a BGRx-only dmabuf offer intersects with nothing and PipeWire
        // fails the link outright:
        //     pw.link: negotiating -> error no more input formats (-22)
        // Measured 2026-08-14 on Hyprland 0.55.4 + xdph 1.3.12: the 12 tiled modifiers matched on
        // both sides perfectly — only the fourcc never did, which is why the failure reads like a
        // GPU/modifier problem and is not one.
        //
        // BGRA and BGRx are the same 32-bit layout; the alpha byte is ignored the whole way to the
        // encoder (`vk_util` maps both to `B8G8R8A8_UNORM`, VAAPI both to `Pixel::BGRA`), and the
        // dmabuf import is driven by the NEGOTIATED format's fourcc, so an AR24 frame imports as
        // AR24. Listed AFTER BGRx so a producer offering both still lands on the pre-existing path
        // — first compatible consumer pod wins, so this is purely additive.
        if !modifiers_bgra.is_empty() {
            pods.push(build_dmabuf_format(
                VideoFormat::BGRA,
                &modifiers_bgra,
                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,
            gpu_dmabuf_negotiation_failed: false,
            native_nv12_env_on: true,
            hdr_cuda_ok: 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 builds the importer on the NVENC path — its pods are LINEAR-only, so the frames
        //    take the Vulkan-bridge/CUDA arm (byte-exact for 4 Bpp), never the 8-bit de-tile
        //    blit. (The tiled half of the invariant is enforced per frame in `.process`, which
        //    sees the negotiated modifier this plan cannot.)
        for want_444 in [false, true] {
            let p = negotiation_plan(NegotiationInputs {
                want_hdr: true,
                want_444,
                ..nvenc()
            });
            assert!(p.build_importer, "HDR on NVENC keeps zero-copy");
        }
        // …but never under a raw passthrough (VAAPI/PyroWave import the dmabuf themselves).
        assert!(
            !negotiation_plan(NegotiationInputs {
                want_hdr: true,
                ..vaapi_native_nv12()
            })
            .build_importer
        );
        // …and never when the resolved encoder can't take a packed 10-bit CUDA payload (libav
        // NVENC: its HDR route swscales into a P010 hardware frame). SDR is unaffected — the term
        // is HDR-only, so a libav host keeps its 8-bit zero-copy.
        assert!(
            !negotiation_plan(NegotiationInputs {
                want_hdr: true,
                hdr_cuda_ok: false,
                ..nvenc()
            })
            .build_importer,
            "HDR must stay on the CPU path where the encoder can't ingest 10-bit CUDA"
        );
        assert!(
            negotiation_plan(NegotiationInputs {
                hdr_cuda_ok: false,
                ..nvenc()
            })
            .build_importer,
            "the HDR-only guard must not touch an SDR session"
        );

        // 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
            );
        }
    }

    /// The EGL→CUDA offer's own negotiation-timeout latch gates `build_importer` — the twin of
    /// the raw passthrough's latch, so a compositor that accepts none of the importer's modifiers
    /// stops being asked (previously it re-ran the same 10 s timeout on every session, forever).
    /// The raw passthrough is untouched by it.
    #[test]
    fn gpu_dmabuf_negotiation_latch_gates_only_the_importer() {
        let p = negotiation_plan(NegotiationInputs {
            gpu_dmabuf_negotiation_failed: true,
            ..nvenc()
        });
        assert!(!p.build_importer, "latched offer must not be re-made");
        assert!(p.gpu_import_latched, "the downgrade must be diagnosable");
        let p = negotiation_plan(NegotiationInputs {
            gpu_dmabuf_negotiation_failed: true,
            ..vaapi_native_nv12()
        });
        assert!(
            p.vaapi_passthrough,
            "the raw passthrough has its own latch — this one must not touch it"
        );
        assert!(!p.gpu_import_latched, "no importer was ever wanted here");
    }

    /// 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);
        // Reported for an HDR capture too — HDR takes the same importer (LINEAR/Vulkan-bridge
        // arm), so the latch really did cost it zero-copy.
        assert!(
            negotiation_plan(NegotiationInputs {
                gpu_import_disabled: true,
                want_hdr: true,
                ..nvenc()
            })
            .gpu_import_latched
        );
        // It is NOT reported for a capture that would never have built one anyway (a raw
        // passthrough) — that would misdirect the operator.
        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]));
        }
    }

    // ---- PW2: capture-arm observability (pure halves) -------------------------------------
    //
    // Env-var reads race under a shared test process, so these assert against the pure functions
    // the logging sites call — the same rule `negotiation_plan_invariants` follows.

    use super::{
        consumer_kind, resolved_capture_arm, CaptureArm, ConsumerKind, FenceWaitStats,
        PassthroughFallback, PassthroughFallbacks, PoolCensus, FENCE_WAIT_BUCKETS_US,
    };

    /// A PyroWave session is PyroWave even though it also flips `backend_is_vaapi` on (the
    /// `linux_zero_copy_is_vaapi` `Pyrowave` arm). Getting this precedence backwards is the exact
    /// shape of the bug PW2 fixes: the session gets reported as somebody else's backend.
    #[test]
    fn pyrowave_outranks_the_host_global_backend_pref() {
        assert_eq!(consumer_kind(true, true, true), ConsumerKind::PyroWave);
        // ...and on an NVIDIA/auto host, where `backend_is_vaapi` is false, it is still PyroWave —
        // the case that previously logged nothing at all.
        assert_eq!(consumer_kind(true, false, true), ConsumerKind::PyroWave);
    }

    /// The non-PyroWave consumers, and the one that must NOT warn on a CPU arm.
    #[test]
    fn consumer_kinds_and_which_ones_a_cpu_arm_degrades() {
        assert_eq!(consumer_kind(false, true, true), ConsumerKind::Vaapi);
        assert_eq!(consumer_kind(false, false, true), ConsumerKind::Nvenc);
        // No GPU backend ⇒ the software encoder, whose native input IS CPU frames.
        assert_eq!(consumer_kind(false, false, false), ConsumerKind::Software);
        assert!(ConsumerKind::PyroWave.cpu_is_downgrade());
        assert!(ConsumerKind::Vaapi.cpu_is_downgrade());
        assert!(ConsumerKind::Nvenc.cpu_is_downgrade());
        assert!(!ConsumerKind::Software.cpu_is_downgrade());
    }

    /// The arm is a function of the plan plus the two runtime facts. Pinned against every plan the
    /// resolver can produce, so the headline line can never claim an arm the session did not take.
    #[test]
    fn resolved_arm_matches_the_plan_that_produced_it() {
        // PyroWave/VAAPI raw passthrough.
        let p = negotiation_plan(NegotiationInputs {
            pyrowave_session: true,
            ..nvenc()
        });
        assert!(p.vaapi_passthrough);
        assert_eq!(
            resolved_capture_arm(&p, false, p.want_dmabuf(false, &[0])),
            CaptureArm::DmabufPassthrough
        );
        // NVENC via the EGL→CUDA importer.
        let p = negotiation_plan(nvenc());
        assert!(p.build_importer);
        assert_eq!(
            resolved_capture_arm(&p, true, p.want_dmabuf(true, &[0])),
            CaptureArm::CudaImport
        );
        // The importer was meant to be built but did not construct (no driver): CPU, not a
        // cuda-import the session never got.
        assert_eq!(
            resolved_capture_arm(&p, false, p.want_dmabuf(false, &[0])),
            CaptureArm::Cpu
        );
        // An empty modifier list is a CPU arm even under a live passthrough plan.
        let p = negotiation_plan(NegotiationInputs {
            pyrowave_session: true,
            ..nvenc()
        });
        assert_eq!(
            resolved_capture_arm(&p, false, p.want_dmabuf(false, &[])),
            CaptureArm::Cpu
        );
        // Forced SHM: CPU regardless of everything else.
        let p = negotiation_plan(NegotiationInputs {
            force_shm: true,
            ..nvenc()
        });
        assert_eq!(
            resolved_capture_arm(&p, true, p.want_dmabuf(true, &[0])),
            CaptureArm::Cpu
        );
    }

    /// The rate limiter: ONE line per distinct reason per session, counting every fall-through.
    /// `.process` runs per frame, so an off-by-one here is a log flood at the capture rate.
    #[test]
    fn fallback_log_budget_is_one_line_per_reason() {
        let mut f = PassthroughFallbacks::default();
        // First of a reason logs, and reports the running total (not a per-reason count).
        assert_eq!(f.note(PassthroughFallback::NotDmabuf), Some(1));
        // Repeats of the SAME reason never log again, but are still counted.
        for _ in 0..1_000 {
            assert_eq!(f.note(PassthroughFallback::NotDmabuf), None);
        }
        // A DIFFERENT reason is a different diagnosis and gets its own line, carrying the
        // now-large total — which is what distinguishes a persistent downgrade from a hiccup.
        assert_eq!(f.note(PassthroughFallback::DupFailed), Some(1002));
        assert_eq!(f.note(PassthroughFallback::DupFailed), None);
        // All four reasons fit the budget independently; the tally counts every frame.
        assert_eq!(f.note(PassthroughFallback::NoFormat), Some(1004));
        assert_eq!(f.note(PassthroughFallback::NoFourcc), Some(1005));
        // Budget spent: every reason has logged once, so nothing logs again however long the
        // session runs.
        for r in [
            PassthroughFallback::NoFormat,
            PassthroughFallback::NotDmabuf,
            PassthroughFallback::NoFourcc,
            PassthroughFallback::DupFailed,
        ] {
            assert_eq!(f.note(r), None);
        }
    }

    /// Every reason is distinguishable (a shared bit would silence one of them) and carries an
    /// actionable hint — a reason with no fix is a line the reader cannot use.
    #[test]
    fn every_fallback_reason_is_distinct_and_actionable() {
        let all = [
            PassthroughFallback::NoFormat,
            PassthroughFallback::NotDmabuf,
            PassthroughFallback::NoFourcc,
            PassthroughFallback::DupFailed,
        ];
        let mut f = PassthroughFallbacks::default();
        for r in all {
            assert!(
                f.note(r).is_some(),
                "{r:?} shares a bit with an earlier reason"
            );
            assert!(!r.as_str().is_empty());
            assert!(!r.hint().is_empty());
        }
        // Only `NoFormat` drops the frame; the other three downgrade it. The log line picks its
        // consequence clause off this, so an inverted answer would print the opposite of the truth.
        assert!(!PassthroughFallback::NoFormat.falls_back_to_cpu());
        assert!(PassthroughFallback::NotDmabuf.falls_back_to_cpu());
        assert!(PassthroughFallback::NoFourcc.falls_back_to_cpu());
        assert!(PassthroughFallback::DupFailed.falls_back_to_cpu());
    }

    // ---- PW4 step 1: the fence-wait histogram ------------------------------------------------
    //
    // This instrument decides whether PW4 ships, so its arithmetic has to be right: a p99 that
    // reads one bucket low would retire a package that was worth doing, and one that reads high
    // would justify moving load-bearing synchronisation off a thread for nothing.

    /// An empty histogram must say "no answer", not "zero" — those are different claims, and the
    /// second one would look like a decisive p99 ≈ 0 result.
    #[test]
    fn an_empty_histogram_has_no_quantile() {
        let s = FenceWaitStats::default();
        assert_eq!(s.quantile_bucket_us(0.99), None);
        assert_eq!(s.mean_us(), 0);
        assert!(!s.is_meaningful(), "it must not be trusted yet either");
    }

    /// The all-fast case — the one that RETIRES PW4. Every sample in the first bucket must put the
    /// p99 there too.
    #[test]
    fn an_all_fast_distribution_puts_p99_in_the_first_bucket() {
        let mut s = FenceWaitStats::default();
        for _ in 0..1000 {
            s.record(3);
        }
        assert_eq!(
            s.quantile_bucket_us(0.50),
            Some(Some(FENCE_WAIT_BUCKETS_US[0]))
        );
        assert_eq!(
            s.quantile_bucket_us(0.99),
            Some(Some(FENCE_WAIT_BUCKETS_US[0]))
        );
        assert_eq!(s.mean_us(), 3);
    }

    /// The case PW4 exists for: a fast median with a heavy tail. The p50 must stay low AND the p99
    /// must find the tail — a histogram that smeared them together could not tell the two worlds
    /// apart, which is the entire decision.
    #[test]
    fn a_heavy_tail_moves_p99_without_moving_p50() {
        let mut s = FenceWaitStats::default();
        for _ in 0..980 {
            s.record(10); // fast majority
        }
        for _ in 0..20 {
            s.record(6_000); // 2 % of frames stall milliseconds
        }
        assert_eq!(
            s.quantile_bucket_us(0.50),
            Some(Some(FENCE_WAIT_BUCKETS_US[0])),
            "the median is still free"
        );
        assert_eq!(
            s.quantile_bucket_us(0.99),
            Some(Some(10_000)),
            "...but the p99 must land in the 5-10ms bucket, not with the median"
        );
        assert_eq!(s.max_us, 6_000);
    }

    /// Anything past the last edge reports as overflow rather than being clamped into the last
    /// bucket — "worse than 10 ms" is a distinct finding and must not read as "10 ms".
    #[test]
    fn waits_past_the_last_edge_report_as_overflow() {
        let mut s = FenceWaitStats::default();
        s.record(99_000);
        assert_eq!(s.quantile_bucket_us(0.99), Some(None));
    }

    /// Bucket edges are inclusive upper bounds, so a sample exactly ON an edge belongs to that
    /// bucket and not the next one up.
    #[test]
    fn bucket_edges_are_inclusive() {
        for (i, &edge) in FENCE_WAIT_BUCKETS_US.iter().enumerate() {
            let mut s = FenceWaitStats::default();
            s.record(edge);
            assert_eq!(
                s.quantile_bucket_us(1.0),
                Some(Some(edge)),
                "a sample of exactly {edge}us belongs in bucket {i}"
            );
        }
    }

    /// PW5 stage 1: a stable pool logs ONE line, not one per frame. `.process` runs at the capture
    /// rate — an unconditional log here would be 240 lines a second of the same number.
    #[test]
    fn a_stable_pool_is_logged_once() {
        let mut p = PoolCensus::default();
        for _ in 0..8 {
            p.add();
        }
        assert_eq!(p.note_frame(), Some(8));
        for _ in 0..100 {
            assert_eq!(p.note_frame(), None, "the same depth must not re-log");
        }
    }

    /// A renegotiation frees the pool and re-allocates it. The LIVE count therefore dips (and the
    /// new depth is worth a second line), but `high_water` — the number a pipeline-depth decision
    /// keys on — must not follow the dip down.
    #[test]
    fn a_renegotiated_pool_relogs_but_the_high_water_holds() {
        let mut p = PoolCensus::default();
        for _ in 0..8 {
            p.add();
        }
        assert_eq!(p.note_frame(), Some(8));
        for _ in 0..8 {
            p.remove();
        }
        for _ in 0..4 {
            p.add();
        }
        assert_eq!(p.note_frame(), Some(4), "a changed depth is worth a line");
        assert_eq!(p.high_water, 8, "the deepest pool seen this session");
    }

    /// `remove_buffer` without a matching `add_buffer` must not wrap the count to `u32::MAX` —
    /// a depth gate reading that would happily pipeline against a pool of zero.
    #[test]
    fn unmatched_removes_saturate_at_zero() {
        let mut p = PoolCensus::default();
        p.remove();
        p.remove();
        assert_eq!(p.note_frame(), Some(0));
        assert_eq!(p.high_water, 0);
    }

    use super::{HoldBook, HOLD_POOL_RESERVE};

    /// The book must always leave [`HOLD_POOL_RESERVE`] buffers with the producer: an 8-pool
    /// spares 6, and the pools at or below the reserve spare NOTHING — those sessions must fall
    /// back to the immediate requeue rather than starve the compositor of render targets.
    #[test]
    fn hold_book_spends_at_most_pool_minus_reserve() {
        let mut b = HoldBook::default();
        for i in 0..6 {
            assert!(
                b.try_hold(0x1000 + i, 8).is_some(),
                "hold {i} within budget"
            );
        }
        assert!(
            b.try_hold(0x2000, 8).is_none(),
            "7th of 8 exceeds the budget"
        );
        assert!(
            HoldBook::default()
                .try_hold(0x1000, HOLD_POOL_RESERVE)
                .is_none(),
            "a pool of exactly the reserve cannot spare a buffer"
        );
        assert!(
            HoldBook::default()
                .try_hold(0x1000, HOLD_POOL_RESERVE + 1)
                .is_some(),
            "one past the reserve spares exactly one"
        );
    }

    /// One hold ⇒ one requeue: the first `complete` releases, a duplicate release (a bug shape,
    /// but also the benign stale-message case) must NOT requeue a second time — handing the
    /// producer the same buffer twice corrupts its pool.
    #[test]
    fn hold_book_releases_exactly_once() {
        let mut b = HoldBook::default();
        let g = b.try_hold(0x1000, 8).unwrap();
        assert!(b.complete(0x1000, g), "first release requeues");
        assert!(!b.complete(0x1000, g), "second release is a no-op");
        assert!(!b.contains(0x1000));
    }

    /// The renegotiation hazard the generation exists for: the pool is replaced (`remove_buffer`
    /// purges), a NEW buffer lands on the SAME address and is withheld, and only then does the
    /// OLD hold's release arrive. Matching by pointer alone would requeue the new tenant while
    /// its own hold is still out — the mid-encode rewrite race, reintroduced by the fix itself.
    #[test]
    fn hold_book_generation_outlives_an_address_reuse() {
        let mut b = HoldBook::default();
        let old = b.try_hold(0x1000, 8).unwrap();
        b.purge(0x1000); // remove_buffer: pool renegotiated away under the hold
        assert!(!b.complete(0x1000, old), "purged hold releases nothing");
        let new = b.try_hold(0x1000, 8).unwrap(); // new pool's buffer, same address
        assert!(
            !b.complete(0x1000, old),
            "the OLD hold cannot release the NEW tenant"
        );
        assert!(b.contains(0x1000), "new tenant still withheld");
        assert!(b.complete(0x1000, new), "its own hold releases it");
    }

    /// A buffer already out cannot be withheld again (one requeue duty per buffer): `.process`
    /// can only re-see an address after its requeue, so a duplicate try_hold means state
    /// confusion — refuse it and let the epilogue requeue immediately.
    #[test]
    fn hold_book_refuses_a_buffer_already_out() {
        let mut b = HoldBook::default();
        b.try_hold(0x1000, 8).unwrap();
        assert!(b.try_hold(0x1000, 8).is_none());
    }
}
