//! PipeWire I/O adapter for the pure registry observer.
//!
//! This module owns a read-only PipeWire main-loop thread, translates registry
//! callbacks into [`RegEvent`]s, and publishes the latest [`Projection`] for
//! consumers running outside the PipeWire thread.

use super::classify::{DeviceClaim, DeviceProps};
use super::{
    EventKind, LinkEndpoints, NodeObservation, Outcome, Projection, RegEvent, RegistryModel,
};
use crate::host::audio::parse_object_serial;
use crate::host::taint::snapshot::{
    ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial,
};
use anyhow::{Context, Result};
use pipewire::{self as pw, types::ObjectType};
use std::cell::{Cell, RefCell};
use std::collections::{BTreeMap, VecDeque};
use std::rc::Rc;
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

const READINESS_TIMEOUT_MILLIS: u64 = 2_000;
const TICK_INTERVAL: Duration = Duration::from_millis(250);

/// A consumer that sees **every** projection, one per applied registry event,
/// on the observer thread.
///
/// This exists because polling [`RegistryObserverHandle::latest`] coalesces, and
/// some consumers cannot tolerate that. Phase 4's AEC validator is the concrete
/// case: it detects a module unload by observing the *empty gap* before the next
/// module appears, and PipeWire reuses module indices verbatim across an
/// unload/reload (v3.4 §5.2 correction 3), so a consumer that misses the gap
/// silently aliases a fresh module onto a dead module's validated identity.
///
/// **Implementations run inline on the PipeWire loop thread.** Whatever they do
/// delays the next registry callback, so they must be bounded and must not
/// block. The phase-5 audit is the only implementor and measures its own cost
/// for exactly this reason.
pub trait ProjectionSink: Send {
    /// `now_us` is monotonic microseconds since the observer started — the same
    /// clock that drives [`RegEvent::Tick`], so a sink's notion of time cannot
    /// drift from the readiness epoch's.
    fn on_projection(&mut self, projection: &Projection, kind: EventKind, now_us: u64);
}

/// Tokio-side access to the observer's most recent coherent projection.
pub struct RegistryObserverHandle {
    latest: Arc<Mutex<Option<Projection>>>,
    shutdown_tx: pw::channel::Sender<()>,
    thread: Option<JoinHandle<()>>,
}

impl RegistryObserverHandle {
    /// Spawn the read-only PipeWire registry observer.
    pub fn spawn() -> Result<Self> {
        Self::spawn_with_sink(None)
    }

    /// Spawn the observer with a per-event [`ProjectionSink`] attached.
    ///
    /// The sink is moved onto the observer thread and dropped when that thread
    /// exits, which is what lets a sink emit a final summary on shutdown without
    /// the caller arranging one.
    pub fn spawn_with_sink(sink: Option<Box<dyn ProjectionSink>>) -> Result<Self> {
        let latest = Arc::new(Mutex::new(None));
        let latest_for_thread = Arc::clone(&latest);
        let (shutdown_tx, shutdown_rx) = pw::channel::channel::<()>();
        let thread = std::thread::Builder::new()
            .name("pixelpass-pw-observer".to_string())
            .spawn(move || {
                if let Err(e) = run_observer(latest_for_thread, shutdown_rx, sink) {
                    tracing::warn!(
                        "registry observer: libpipewire thread exited with error: {e:#}"
                    );
                }
            })
            .context("failed to spawn libpipewire registry observer thread")?;

        Ok(Self {
            latest,
            shutdown_tx,
            thread: Some(thread),
        })
    }

    /// Return a clone of the latest projection, or `None` before the first
    /// registry event has been applied.
    pub fn latest(&self) -> Option<Projection> {
        self.latest
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone()
    }
}

impl Drop for RegistryObserverHandle {
    fn drop(&mut self) {
        let _ = self.shutdown_tx.send(());
        if let Some(thread) = self.thread.take()
            && let Err(e) = thread.join()
        {
            tracing::warn!("registry observer: pw thread join failed: {e:?}");
        }
    }
}

enum BoundProxy {
    Node {
        _listener: pw::node::NodeListener,
        _proxy: pw::node::Node,
    },
    Device {
        _listener: pw::device::DeviceListener,
        _proxy: pw::device::Device,
    },
    Link {
        _listener: pw::link::LinkListener,
        _proxy: pw::link::Link,
    },
}

struct LiveGlobal {
    serial: Serial,
    bound_proxy: Option<BoundProxy>,
}

struct ObserverState {
    model: RegistryModel,
    latest: Arc<Mutex<Option<Projection>>>,
    last_candidate: Option<u32>,
    live_globals: BTreeMap<GlobalId, VecDeque<LiveGlobal>>,
    sink: Option<Box<dyn ProjectionSink>>,
    started_at: Instant,
}

impl ObserverState {
    /// `started_at` is the observer's single time origin, shared with the
    /// readiness tick timer — so a sink's `now_us` and a `RegEvent::Tick`'s
    /// `now` are the same clock, not two that drift.
    fn new(
        latest: Arc<Mutex<Option<Projection>>>,
        sink: Option<Box<dyn ProjectionSink>>,
        started_at: Instant,
    ) -> Self {
        Self {
            model: RegistryModel::new(0, READINESS_TIMEOUT_MILLIS),
            latest,
            last_candidate: None,
            live_globals: BTreeMap::new(),
            sink,
            started_at,
        }
    }

    fn apply(&mut self, event: RegEvent) -> Outcome {
        // Taken before the model consumes the event: the sink is told what kind
        // of observation produced the projection, and deriving that from the
        // event itself is what stops the two from ever disagreeing.
        let kind = event.kind();
        let event_outcome = self.model.apply(event);
        let mut outcome = event_outcome;

        let candidate = self.model.pulse_pid_candidate();
        if candidate != self.last_candidate {
            self.last_candidate = candidate;
            if let Some(pid) = candidate {
                let comm = std::fs::read_to_string(format!("/proc/{pid}/comm"))
                    .ok()
                    .map(|comm| comm.trim_end_matches(['\r', '\n']).to_string());
                // Folded into the model directly rather than through `apply`, so
                // one registry event still yields exactly one sink call — the
                // no-coalescing contract cuts both ways, and a *duplicated*
                // observation would make the O5 event rate a fiction.
                if self.model.apply(RegEvent::ProcCommProbed { pid, comm }) == Outcome::Applied {
                    outcome = Outcome::Applied;
                }
            }
        }

        // v3.5 §6.7 decision 2: a projection the model proved identical is not
        // published. Only the model can make that claim soundly, which is why
        // it is [`Outcome`] and not a diff of two snapshots here.
        if outcome == Outcome::Applied {
            self.publish(kind);
        }
        event_outcome
    }

    fn publish(&mut self, kind: EventKind) {
        let projection = self.model.project();
        if let Some(sink) = self.sink.as_mut() {
            let now_us = u64::try_from(self.started_at.elapsed().as_micros()).unwrap_or(u64::MAX);
            sink.on_projection(&projection, kind, now_us);
        }
        // Published after the sink has seen it, so the projection is moved
        // rather than cloned — the snapshot is the largest thing the observer
        // owns and this runs on every event.
        *self
            .latest
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(projection);
    }

    /// Record the global's id and apply its add event as one step, so the
    /// bound-proxy FIFO stays provably lockstep with the model's own `live_ids`
    /// index. Recording only on *applied* adds (never on unknown object types
    /// or globals dropped for a missing serial) is what keeps the two id
    /// queues the same length per id — otherwise a phantom slot could pop
    /// another generation's proxy after an id is recycled.
    fn add(&mut self, serial: Serial, id: GlobalId, event: RegEvent) {
        if self.apply(event) == Outcome::Applied {
            self.live_globals
                .entry(id)
                .or_default()
                .push_back(LiveGlobal {
                    serial,
                    bound_proxy: None,
                });
        }
    }

    /// Return a proxy that could not be attached so its listener is dropped
    /// after the caller releases the `RefCell` borrow.
    fn attach_bound_proxy(
        &mut self,
        id: GlobalId,
        serial: Serial,
        bound_proxy: BoundProxy,
    ) -> Option<BoundProxy> {
        let Some(global) = self
            .live_globals
            .get_mut(&id)
            .and_then(|globals| globals.iter_mut().find(|global| global.serial == serial))
        else {
            tracing::warn!(
                global_id = id.0,
                serial = serial.0,
                "registry observer: bind completed without a live global slot"
            );
            return Some(bound_proxy);
        };
        if global.bound_proxy.is_some() {
            tracing::warn!(
                global_id = id.0,
                serial = serial.0,
                "registry observer: live global slot already has a bound proxy"
            );
            return Some(bound_proxy);
        }
        global.bound_proxy = Some(bound_proxy);
        None
    }

    fn remove_global(&mut self, id: GlobalId) -> Option<BoundProxy> {
        let (bound_proxy, empty) = {
            let globals = self.live_globals.get_mut(&id)?;
            let bound_proxy = globals.pop_front().and_then(|global| global.bound_proxy);
            (bound_proxy, globals.is_empty())
        };
        if empty {
            self.live_globals.remove(&id);
        }
        bound_proxy
    }
}

fn run_observer(
    latest: Arc<Mutex<Option<Projection>>>,
    shutdown_rx: pw::channel::Receiver<()>,
    sink: Option<Box<dyn ProjectionSink>>,
) -> Result<()> {
    let started_at = Instant::now();
    let main_loop =
        pw::main_loop::MainLoopRc::new(None).context("pw main loop construction failed")?;
    let context =
        pw::context::ContextRc::new(&main_loop, None).context("pw context construction failed")?;
    let core = context
        .connect_rc(None)
        .context("pw core connect failed (is the daemon running?)")?;
    let registry = core.get_registry_rc().context("pw get_registry failed")?;
    let state = Rc::new(RefCell::new(ObserverState::new(latest, sink, started_at)));

    let main_loop_for_shutdown = main_loop.clone();
    let _shutdown_receiver = shutdown_rx.attach(main_loop.loop_(), move |()| {
        main_loop_for_shutdown.quit();
    });

    let pending_sync = Rc::new(Cell::new(None));
    let pending_sync_for_done = Rc::clone(&pending_sync);
    let state_for_done = Rc::clone(&state);
    let _core_listener = core
        .add_listener_local()
        .done(move |id, seq| {
            if id == pw::core::PW_ID_CORE && pending_sync_for_done.get() == Some(seq) {
                pending_sync_for_done.set(None);
                state_for_done.borrow_mut().apply(RegEvent::ServerSynced);
            }
        })
        .error(|id, seq, res, message| {
            tracing::warn!(
                id,
                seq,
                result = res,
                %message,
                "registry observer: PipeWire core error"
            );
        })
        .register();

    let registry_weak = registry.downgrade();
    let state_for_global = Rc::clone(&state);
    let state_for_remove = Rc::clone(&state);
    let _registry_listener = registry
        .add_listener_local()
        .global(move |obj| {
            let id = GlobalId(obj.id);

            match obj.type_ {
                ObjectType::Node => {
                    // ⚠️ v3.5 §6.7: the global is an INDEX. Only `object.serial`
                    // is read here; every property the engine reasons about
                    // comes from the bind's `info` (phase 3r).
                    let Some(props) = obj.props.as_ref() else {
                        tracing::warn!(
                            node_id = obj.id,
                            "registry observer: Node has no properties; dropping"
                        );
                        return;
                    };
                    let Some(serial) = parse_serial(obj.id, "Node", props.get("object.serial"))
                    else {
                        return;
                    };
                    state_for_global.borrow_mut().add(
                        serial,
                        id,
                        RegEvent::NodeAdded { serial, id },
                    );

                    let Some(registry) = registry_weak.upgrade() else {
                        return;
                    };
                    let node: pw::node::Node = match registry.bind(obj) {
                        Ok(node) => node,
                        Err(e) => {
                            tracing::warn!(
                                node_id = obj.id,
                                "registry observer: failed to bind Node for properties: {e}"
                            );
                            return;
                        }
                    };
                    // This bit only recognizes the initial callback for the
                    // change-mask fast path. Admission vs update remains
                    // entirely the model's decision.
                    let first_info = Cell::new(true);
                    let state_for_info = Rc::downgrade(&state_for_global);
                    let listener = node
                        .add_listener_local()
                        .info(move |info| {
                            let Some(props) = info.props() else {
                                return;
                            };
                            let first = first_info.replace(false);
                            if !first
                                && !info.change_mask().contains(pw::node::NodeChangeMask::PROPS)
                            {
                                return;
                            }
                            if let Some(state) = state_for_info.upgrade() {
                                state.borrow_mut().apply(RegEvent::NodeInfo {
                                    serial,
                                    observation: node_observation_from_props(props),
                                });
                            }
                        })
                        .register();
                    let unattached = state_for_global.borrow_mut().attach_bound_proxy(
                        id,
                        serial,
                        BoundProxy::Node {
                            _listener: listener,
                            _proxy: node,
                        },
                    );
                    drop(unattached);
                }
                ObjectType::Port => {
                    let Some(props) = obj.props.as_ref() else {
                        tracing::warn!(
                            port_id = obj.id,
                            "registry observer: Port has no properties; dropping"
                        );
                        return;
                    };
                    let Some(serial) = parse_serial(obj.id, "Port", props.get("object.serial"))
                    else {
                        return;
                    };
                    let Some(node) = props
                        .get("node.id")
                        .and_then(|value| value.parse::<u32>().ok())
                        .map(GlobalId)
                    else {
                        tracing::warn!(
                            port_id = obj.id,
                            node_id = props.get("node.id").unwrap_or("<absent>"),
                            "registry observer: Port has no usable node.id; dropping"
                        );
                        return;
                    };
                    let direction = match props.get("port.direction") {
                        Some("in") => PortDirection::In,
                        Some("out") => PortDirection::Out,
                        direction => {
                            tracing::warn!(
                                port_id = obj.id,
                                direction = direction.unwrap_or("<absent>"),
                                "registry observer: Port has no usable direction; dropping"
                            );
                            return;
                        }
                    };
                    state_for_global.borrow_mut().add(
                        serial,
                        id,
                        RegEvent::PortAdded(PortSnapshot {
                            serial,
                            id,
                            node,
                            direction,
                            exclusive: truthy(props.get("port.exclusive")),
                            monitor: truthy(props.get("port.monitor")),
                        }),
                    );
                }
                ObjectType::Client => {
                    let Some(props) = obj.props.as_ref() else {
                        tracing::warn!(
                            client_id = obj.id,
                            "registry observer: Client has no properties; dropping"
                        );
                        return;
                    };
                    let Some(serial) = parse_serial(obj.id, "Client", props.get("object.serial"))
                    else {
                        return;
                    };
                    state_for_global.borrow_mut().add(
                        serial,
                        id,
                        RegEvent::ClientAdded(ClientSnapshot {
                            serial,
                            id,
                            sec_pid: props
                                .get("pipewire.sec.pid")
                                .and_then(|value| value.parse::<u32>().ok()),
                        }),
                    );
                }
                ObjectType::Device => {
                    // Index only, exactly as for a Node: `device.api` and
                    // `alsa.driver_name` live on the bind's `info` (v3.5 §6.7
                    // decision 4), not here.
                    let Some(props) = obj.props.as_ref() else {
                        tracing::warn!(
                            device_id = obj.id,
                            "registry observer: Device has no properties; dropping"
                        );
                        return;
                    };
                    let Some(serial) = parse_serial(obj.id, "Device", props.get("object.serial"))
                    else {
                        return;
                    };
                    state_for_global.borrow_mut().add(
                        serial,
                        id,
                        RegEvent::DeviceAdded { serial, id },
                    );

                    let Some(registry) = registry_weak.upgrade() else {
                        return;
                    };
                    let device: pw::device::Device = match registry.bind(obj) {
                        Ok(device) => device,
                        Err(e) => {
                            tracing::warn!(
                                device_id = obj.id,
                                "registry observer: failed to bind Device for properties: {e}"
                            );
                            return;
                        }
                    };
                    let first_info = Cell::new(true);
                    let state_for_info = Rc::downgrade(&state_for_global);
                    let listener = device
                        .add_listener_local()
                        .info(move |info| {
                            let Some(props) = info.props() else {
                                return;
                            };
                            let first = first_info.replace(false);
                            if !first
                                && !info
                                    .change_mask()
                                    .contains(pw::device::DeviceChangeMask::PROPS)
                            {
                                return;
                            }
                            if let Some(state) = state_for_info.upgrade() {
                                state.borrow_mut().apply(RegEvent::DeviceInfo {
                                    serial,
                                    props: device_props_from_props(props),
                                });
                            }
                        })
                        .register();
                    let unattached = state_for_global.borrow_mut().attach_bound_proxy(
                        id,
                        serial,
                        BoundProxy::Device {
                            _listener: listener,
                            _proxy: device,
                        },
                    );
                    drop(unattached);
                }
                ObjectType::Link => {
                    let Some(props) = obj.props.as_ref() else {
                        tracing::warn!(
                            link_id = obj.id,
                            "registry observer: Link has no properties; dropping"
                        );
                        return;
                    };
                    let Some(serial) = parse_serial(obj.id, "Link", props.get("object.serial"))
                    else {
                        return;
                    };
                    let endpoints = link_endpoints_from_props(props);
                    state_for_global.borrow_mut().add(
                        serial,
                        id,
                        RegEvent::LinkAdded {
                            serial,
                            id,
                            endpoints,
                        },
                    );
                    if endpoints.is_some() {
                        return;
                    }

                    let Some(registry) = registry_weak.upgrade() else {
                        return;
                    };
                    let link: pw::link::Link = match registry.bind(obj) {
                        Ok(link) => link,
                        Err(e) => {
                            tracing::warn!(
                                link_id = obj.id,
                                "registry observer: failed to bind Link for endpoints: {e}"
                            );
                            return;
                        }
                    };
                    let resolved = Rc::new(Cell::new(false));
                    let resolved_for_info = Rc::clone(&resolved);
                    let state_for_info = Rc::downgrade(&state_for_global);
                    let listener = link
                        .add_listener_local()
                        .info(move |info| {
                            if resolved_for_info.replace(true) {
                                return;
                            }
                            let endpoints = LinkEndpoints {
                                output_node: GlobalId(info.output_node_id()),
                                input_node: GlobalId(info.input_node_id()),
                                output_port: optional_global_id(info.output_port_id()),
                                input_port: optional_global_id(info.input_port_id()),
                            };
                            if let Some(state) = state_for_info.upgrade() {
                                state
                                    .borrow_mut()
                                    .apply(RegEvent::LinkEndpointsResolved { serial, endpoints });
                            }
                        })
                        .register();
                    let unattached = state_for_global.borrow_mut().attach_bound_proxy(
                        id,
                        serial,
                        BoundProxy::Link {
                            _listener: listener,
                            _proxy: link,
                        },
                    );
                    drop(unattached);
                }
                _ => {}
            }
        })
        .global_remove(move |id| {
            let id = GlobalId(id);
            let bound_proxy = state_for_remove.borrow_mut().remove_global(id);
            state_for_remove
                .borrow_mut()
                .apply(RegEvent::Removed { id });
            drop(bound_proxy);
        })
        .register();

    pending_sync.set(Some(
        core.sync(0)
            .context("registry observer: initial core.sync failed")?,
    ));

    let state_for_tick = Rc::clone(&state);
    let timer = main_loop.loop_().add_timer(move |_| {
        let now = u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
        state_for_tick.borrow_mut().apply(RegEvent::Tick { now });
    });
    timer
        .update_timer(Some(TICK_INTERVAL), Some(TICK_INTERVAL))
        .into_result()
        .context("registry observer: failed to arm readiness timer")?;

    tracing::info!("registry observer: pw thread running");
    main_loop.run();
    tracing::info!("registry observer: pw thread exiting");
    Ok(())
}

fn parse_serial(id: u32, kind: &str, raw: Option<&str>) -> Option<Serial> {
    match raw.and_then(parse_object_serial) {
        Some(serial) => Some(Serial(serial)),
        None => {
            tracing::warn!(
                global_id = id,
                object_type = kind,
                serial = raw.unwrap_or("<absent>"),
                "registry observer: global has no usable object.serial; dropping"
            );
            None
        }
    }
}

fn truthy(value: Option<&str>) -> bool {
    value.is_some_and(|value| value != "false" && value != "0")
}

fn node_observation_from_props(props: &pw::spa::utils::dict::DictRef) -> NodeObservation {
    NodeObservation {
        name: props.get("node.name").map(str::to_string),
        role: MediaRole::parse(props.get("media.class")),
        props: NodeProps {
            peerspeak_owned: truthy(props.get("peerspeak.owned")),
            pulse_module_id: props
                .get("pulse.module.id")
                .and_then(|value| value.parse::<u64>().ok()),
            link_group: props.get("node.link-group").map(str::to_string),
            client_id: props
                .get("client.id")
                .and_then(|value| value.parse::<u32>().ok())
                .map(GlobalId),
            process_id: props
                .get("application.process.id")
                .and_then(|value| value.parse::<u32>().ok()),
            passthrough: truthy(props.get("node.passthrough")),
            session_device: false,
        },
        device_claim: DeviceClaim {
            device_id: props
                .get("device.id")
                .and_then(|value| value.parse::<u32>().ok())
                .map(GlobalId),
            device_api: props.get("device.api").map(str::to_string),
            factory_name: props.get("factory.name").map(str::to_string),
            alsa_driver_name: props.get("alsa.driver_name").map(str::to_string),
        },
    }
}

fn device_props_from_props(props: &pw::spa::utils::dict::DictRef) -> DeviceProps {
    DeviceProps {
        device_api: props.get("device.api").map(str::to_string),
        alsa_driver_name: props.get("alsa.driver_name").map(str::to_string),
    }
}

fn link_endpoints_from_props(props: &pw::spa::utils::dict::DictRef) -> Option<LinkEndpoints> {
    let output_node = props.get("link.output.node")?.parse::<u32>().ok()?;
    let input_node = props.get("link.input.node")?.parse::<u32>().ok()?;
    Some(LinkEndpoints {
        output_node: GlobalId(output_node),
        input_node: GlobalId(input_node),
        output_port: props
            .get("link.output.port")
            .and_then(|value| value.parse::<u32>().ok())
            .map(GlobalId),
        input_port: props
            .get("link.input.port")
            .and_then(|value| value.parse::<u32>().ok())
            .map(GlobalId),
    })
}

fn optional_global_id(id: u32) -> Option<GlobalId> {
    (id != pw::constants::ID_ANY).then_some(GlobalId(id))
}

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

    struct PactlModule {
        id: Option<u32>,
    }

    impl PactlModule {
        fn load(name: &str, args: &[String]) -> Self {
            let output = Command::new("pactl")
                .arg("load-module")
                .arg(name)
                .args(args)
                .output()
                .expect("pactl must be installed for the live observer test");
            assert!(
                output.status.success(),
                "pactl load-module {name} failed: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
            let id = String::from_utf8(output.stdout)
                .expect("pactl module id must be UTF-8")
                .trim()
                .parse::<u32>()
                .expect("pactl module id must be a u32");
            Self { id: Some(id) }
        }

        fn unload(mut self) {
            let id = self.id.take().expect("module must still be loaded");
            let output = Command::new("pactl")
                .arg("unload-module")
                .arg(id.to_string())
                .output()
                .expect("pactl must be installed for the live observer test");
            assert!(
                output.status.success(),
                "pactl unload-module {id} failed: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
        }
    }

    impl Drop for PactlModule {
        fn drop(&mut self) {
            if let Some(id) = self.id.take() {
                let _ = Command::new("pactl")
                    .arg("unload-module")
                    .arg(id.to_string())
                    .output();
            }
        }
    }

    fn wait_for(
        observer: &RegistryObserverHandle,
        predicate: impl Fn(&Projection) -> bool,
    ) -> Projection {
        let deadline = Instant::now() + Duration::from_secs(10);
        while Instant::now() < deadline {
            if let Some(projection) = observer.latest()
                && predicate(&projection)
            {
                return projection;
            }
            std::thread::sleep(Duration::from_millis(50));
        }
        panic!("timed out waiting for the registry projection");
    }

    fn has_node(projection: &Projection, name: &str) -> bool {
        projection
            .snapshot
            .nodes()
            .any(|node| node.name.as_deref() == Some(name))
    }

    /// Phase 3r exit-gate row 1, the Device half — and the reason it needs its
    /// own test.
    ///
    /// `live_bound_properties_recover_node_and_device_inputs` asserts
    /// `session_device`, which the classifier grants on a **union**:
    /// `device.api` and `alsa.driver_name` may come from the bound Device *or*
    /// from the node's own copies. On this host (WirePlumber 0.5.15 ≥ 0.5.13)
    /// the session manager *does* copy both onto ALSA nodes, so that assertion
    /// passes through the node fallback and would keep passing if the Device
    /// bind delivered nothing at all — leaving v3.5 §6.7 decision 4, the whole
    /// authoritative path, ungated on the machine we develop on.
    ///
    /// So assert the Device side directly: bind every Device global and require
    /// that at least one ALSA card announces **both** keys on its `info` props.
    /// A failure here means the fix for the phase-3 review's owed finding (a
    /// real card over-excluded on installs that do not copy `alsa.*` onto the
    /// node) rests on nothing.
    #[test]
    #[ignore = "needs live pipewire"]
    fn live_device_bind_carries_api_and_driver_name() {
        pw::init();
        let main_loop = pw::main_loop::MainLoopRc::new(None).expect("pw main loop");
        let context = pw::context::ContextRc::new(&main_loop, None).expect("pw context");
        let core = context.connect_rc(None).expect("pw core connect");
        let registry = core.get_registry_rc().expect("pw registry");

        // Devices bound off the registry, each holding its proxy + listener so
        // the callback lives long enough to fire, exactly as the adapter does.
        let bound: Rc<RefCell<Vec<(pw::device::Device, pw::device::DeviceListener)>>> =
            Rc::new(RefCell::new(Vec::new()));
        let observed: Rc<RefCell<Vec<DeviceProps>>> = Rc::new(RefCell::new(Vec::new()));

        let bound_for_global = Rc::clone(&bound);
        let observed_for_global = Rc::clone(&observed);
        let registry_weak = registry.downgrade();
        let _listener = registry
            .add_listener_local()
            .global(move |obj| {
                if obj.type_ != ObjectType::Device {
                    return;
                }
                let Some(registry) = registry_weak.upgrade() else {
                    return;
                };
                let Ok(device) = registry.bind::<pw::device::Device, _>(obj) else {
                    return;
                };
                let observed_for_info = Rc::clone(&observed_for_global);
                let listener = device
                    .add_listener_local()
                    .info(move |info| {
                        if let Some(props) = info.props() {
                            observed_for_info
                                .borrow_mut()
                                .push(device_props_from_props(props));
                        }
                    })
                    .register();
                bound_for_global.borrow_mut().push((device, listener));
            })
            .register();

        // Two seconds is the same budget the observer gives its own binds.
        let main_loop_for_timer = main_loop.clone();
        let timer = main_loop
            .loop_()
            .add_timer(move |_| main_loop_for_timer.quit());
        timer
            .update_timer(Some(Duration::from_secs(2)), None)
            .into_result()
            .expect("arm the test deadline");
        main_loop.run();

        let observed = observed.borrow();
        assert!(
            !observed.is_empty(),
            "no Device delivered info props at all — the Device bind path is dead"
        );
        assert!(
            observed.iter().any(|props| {
                props.device_api.as_deref() == Some("alsa") && props.alsa_driver_name.is_some()
            }),
            "no bound Device carried both device.api=alsa and alsa.driver_name; \
             observed: {observed:?}"
        );
    }

    // Phase 3r exit-gate row 1: failure means the observation boundary regressed.
    #[test]
    #[ignore = "needs live pipewire"]
    fn live_bound_properties_recover_node_and_device_inputs() {
        pw::init();
        let observer = RegistryObserverHandle::spawn().expect("observer thread must spawn");
        wait_for(&observer, |projection| projection.graph_ready);

        let unique = format!("pixelpass_observer_props_test_{}", std::process::id());
        let capture_name = format!("{unique}_capture");
        let playback_name = format!("{unique}_playback");
        let null_sink = PactlModule::load(
            "module-null-sink",
            &[
                format!("sink_name={unique}"),
                "sink_properties=peerspeak.owned=true node.passthrough=true".to_string(),
            ],
        );
        let null_sink_id = null_sink.id.expect("null-sink module must have an id");
        let loopback = PactlModule::load(
            "module-loopback",
            &[
                format!("source={unique}.monitor"),
                format!("sink={unique}"),
                format!("source_output_properties=node.name={capture_name}"),
                format!("sink_input_properties=node.name={playback_name}"),
            ],
        );

        let projection = wait_for(&observer, |projection| {
            projection.graph_ready
                && has_node(projection, &unique)
                && has_node(projection, &capture_name)
                && has_node(projection, &playback_name)
        });
        let tagged_sink = projection
            .snapshot
            .nodes()
            .find(|node| node.name.as_deref() == Some(&unique))
            .expect("tagged null sink must be projected");
        assert!(tagged_sink.props.peerspeak_owned);
        assert!(tagged_sink.props.passthrough);
        assert_eq!(
            tagged_sink.props.pulse_module_id,
            Some(u64::from(null_sink_id))
        );

        let capture = projection
            .snapshot
            .nodes()
            .find(|node| node.name.as_deref() == Some(&capture_name))
            .expect("loopback capture leg must be projected");
        let playback = projection
            .snapshot
            .nodes()
            .find(|node| node.name.as_deref() == Some(&playback_name))
            .expect("loopback playback leg must be projected");
        let capture_group = capture
            .props
            .link_group
            .as_ref()
            .expect("loopback capture leg must carry node.link-group");
        let playback_group = playback
            .props
            .link_group
            .as_ref()
            .expect("loopback playback leg must carry node.link-group");
        assert_eq!(capture_group, playback_group);
        assert!(
            projection
                .snapshot
                .nodes()
                .any(|node| node.props.process_id.is_some()),
            "at least one projected node must carry application.process.id"
        );
        let session_device = projection.snapshot.nodes().find(|node| {
            node.props.session_device
                && (node
                    .name
                    .as_deref()
                    .is_some_and(|name| name.contains("alsa"))
                    || matches!(node.role, MediaRole::Sink | MediaRole::Source))
        });
        assert!(
            session_device.is_some(),
            "a named ALSA or Audio/Sink/Audio/Source node must classify as a session device"
        );
        assert!(projection.graph_ready);

        loopback.unload();
        null_sink.unload();
        wait_for(&observer, |projection| {
            !has_node(projection, &unique)
                && !has_node(projection, &capture_name)
                && !has_node(projection, &playback_name)
        });
    }

    #[test]
    #[ignore = "needs live pipewire"]
    fn live_topology_diff_tracks_null_sink_and_loopback() {
        pw::init();
        let observer = RegistryObserverHandle::spawn().expect("observer thread must spawn");
        let baseline = wait_for(&observer, |projection| projection.graph_ready);
        let baseline_links = baseline.snapshot.links().count();

        let unique = format!("pixelpass_observer_test_{}", std::process::id());
        let capture_name = format!("{unique}_capture");
        let playback_name = format!("{unique}_playback");
        let null_sink = PactlModule::load("module-null-sink", &[format!("sink_name={unique}")]);
        let with_sink = wait_for(&observer, |projection| has_node(projection, &unique));
        let sink_links = with_sink.snapshot.links().count();

        let loopback = PactlModule::load(
            "module-loopback",
            &[
                format!("source={unique}.monitor"),
                format!("sink={unique}"),
                format!("source_output_properties=node.name={capture_name}"),
                format!("sink_input_properties=node.name={playback_name}"),
            ],
        );
        wait_for(&observer, |projection| {
            has_node(projection, &capture_name)
                && has_node(projection, &playback_name)
                && projection.snapshot.links().count() > sink_links
        });

        loopback.unload();
        null_sink.unload();
        wait_for(&observer, |projection| {
            !has_node(projection, &unique)
                && !has_node(projection, &capture_name)
                && !has_node(projection, &playback_name)
                && projection.snapshot.links().count() <= baseline_links
        });
    }
}