pixelpass/src/host/audio_plan.rs

//! Typed selection and ownership for the audio branch of a capture.
//!
//! The enum is deliberately shaped so the desktop-excluding mode cannot carry
//! either unsafe legacy input:
//!
//! - only [`CapturePlan::LegacyDesktop`] can contain [`DefaultMonitor`];
//! - only [`CapturePlan::PerApp`] can contain [`Routing`], whose legacy mode may
//!   load the default-monitor Pulse loopback;
//! - [`CapturePlan::DesktopExcluding`] contains only [`BareCaptureSink`], which
//!   has no module ledger or loopback constructor.
//!
//! Phase 6 feeds the last variant through retained, non-lingering native
//! PipeWire links. Phase 7 exposes the typed selector and versioned capability;
//! Phase 8 is the PeerSpeak-side capability-gated integration.

use anyhow::{Context, Result, bail};
use tokio::process::Command;

use super::aec::AecConfig;
use super::audio::Routing;
use super::graph::BareCaptureSink;
use super::health;
use crate::cli::{CaptureMode, HostOpts};

trait CapturePlanBackend {
    async fn resolve_default_monitor(&mut self) -> Result<DefaultMonitor>;
    async fn start_routing(&mut self, opts: &HostOpts, health: health::Reporter)
    -> Result<Routing>;
    async fn start_bare_capture_sink(
        &mut self,
        health: health::Reporter,
        aec: AecConfig,
    ) -> Result<BareCaptureSink>;
}

struct SystemCapturePlanBackend;

impl CapturePlanBackend for SystemCapturePlanBackend {
    async fn resolve_default_monitor(&mut self) -> Result<DefaultMonitor> {
        DefaultMonitor::resolve().await
    }

    async fn start_routing(
        &mut self,
        opts: &HostOpts,
        health: health::Reporter,
    ) -> Result<Routing> {
        Routing::start(opts, health).await
    }

    async fn start_bare_capture_sink(
        &mut self,
        health: health::Reporter,
        aec: AecConfig,
    ) -> Result<BareCaptureSink> {
        BareCaptureSink::start(health, aec).await
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum CapturePlanKind {
    LegacyDesktop,
    /// Also covers the legacy `PIXELPASS_AUDIO_VIA_NULL_SINK` dogfood path
    /// when no app is selected. The odd name is retained from the design of
    /// record so the three safety variants stay mechanically recognisable.
    PerApp,
    DesktopExcluding,
}

impl CapturePlanKind {
    pub(super) fn resolve(opts: &HostOpts) -> Result<Self> {
        if opts.capture_mode == CaptureMode::DesktopExcluding {
            if opts.app.is_some() {
                bail!(
                    "desktop-excluding capture cannot be combined with --app; legacy Routing is forbidden in this mode"
                );
            }
            if opts.legacy_null_sink {
                bail!(
                    "desktop-excluding capture cannot be combined with PIXELPASS_AUDIO_VIA_NULL_SINK; the default-monitor loopback is forbidden in this mode"
                );
            }
            return Ok(Self::DesktopExcluding);
        }

        if opts.app.is_some() || opts.legacy_null_sink {
            Ok(Self::PerApp)
        } else {
            Ok(Self::LegacyDesktop)
        }
    }
}

/// The real default sink's monitor. Construction is private to this module and
/// the value can only inhabit `LegacyDesktop`.
pub(super) struct DefaultMonitor(String);

impl DefaultMonitor {
    async fn resolve() -> Result<Self> {
        let output = Command::new("pactl")
            .arg("get-default-sink")
            .output()
            .await
            .context(
                "failed to run `pactl get-default-sink` (install pulseaudio-utils or pipewire-pulse)",
            )?;
        if !output.status.success() {
            bail!(
                "pactl get-default-sink failed: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
        }
        let sink = String::from_utf8(output.stdout)
            .context("default sink name was not UTF-8")?
            .trim()
            .to_string();
        if sink.is_empty() {
            bail!("pactl get-default-sink returned no name (is a sound server running?)");
        }
        Ok(Self(format!("{sink}.monitor")))
    }

    fn gst_device_arg(&self) -> String {
        format!("device={}", self.0)
    }
}

/// Owns both the typed source choice and every graph object needed to keep that
/// choice alive for the capture lifetime.
pub(super) enum CapturePlan {
    LegacyDesktop { source: DefaultMonitor },
    PerApp { routing: Routing },
    DesktopExcluding { capture_sink: BareCaptureSink },
}

impl CapturePlan {
    pub(super) async fn start(opts: &HostOpts, health: health::Reporter) -> Result<Self> {
        let mut backend = SystemCapturePlanBackend;
        Self::start_with_backend(opts, health, &mut backend).await
    }

    /// One production construction path with an injectable system boundary.
    ///
    /// Keeping failure policy here makes rows 8d/8e falsifiable in release
    /// builds: a capture-sink or readiness failure must escape this function.
    /// It must never be translated into a second attempt through
    /// `LegacyDesktop`, whose default-monitor source would reintroduce the
    /// audio this mode exists to exclude.
    async fn start_with_backend<B: CapturePlanBackend>(
        opts: &HostOpts,
        health: health::Reporter,
        backend: &mut B,
    ) -> Result<Self> {
        match CapturePlanKind::resolve(opts)? {
            CapturePlanKind::LegacyDesktop => Ok(Self::LegacyDesktop {
                source: backend.resolve_default_monitor().await?,
            }),
            CapturePlanKind::PerApp => Ok(Self::PerApp {
                routing: backend
                    .start_routing(opts, health)
                    .await
                    .context("audio routing setup failed")?,
            }),
            CapturePlanKind::DesktopExcluding => Ok(Self::DesktopExcluding {
                capture_sink: backend
                    .start_bare_capture_sink(health, opts.aec)
                    .await
                    .context("desktop-excluding capture-sink setup failed")?,
            }),
        }
    }

    /// The only conversion to GStreamer's stringly `pulsesrc device=...`
    /// boundary. Callers cannot supply a free-form source string.
    pub(super) fn gst_device_arg(&self) -> String {
        match self {
            Self::LegacyDesktop { source } => source.gst_device_arg(),
            Self::PerApp { routing } => format!("device={}.monitor", routing.sink_name()),
            Self::DesktopExcluding { capture_sink } => {
                format!("device={}", capture_sink.monitor_name())
            }
        }
    }

    #[cfg(test)]
    pub(super) fn legacy_fixture(monitor_name: &str) -> Self {
        Self::LegacyDesktop {
            source: DefaultMonitor(monitor_name.to_string()),
        }
    }

    pub(super) async fn shutdown(self) {
        match self {
            Self::LegacyDesktop { .. } => {}
            Self::PerApp { routing } => routing.shutdown().await,
            Self::DesktopExcluding { capture_sink } => capture_sink.shutdown().await,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cli::Quality;
    use crate::host::fanout::TestFanoutPolicy;
    use crate::host::observer::{Projection, Readiness};
    use crate::host::taint::fixture::Graph;
    use nix::sys::signal::Signal;
    use std::collections::{HashMap, HashSet};
    use std::f64::consts::PI;
    use std::io::{BufRead, BufReader, Read, Write};
    use std::path::PathBuf;
    use std::process::{Child, Command, Stdio};
    use std::sync::mpsc as std_mpsc;
    use std::time::{Duration, Instant};

    const PHASE6_SIGKILL_HELPER: &str = "PIXELPASS_PHASE6_SIGKILL_HELPER";
    const PHASE6_READY: &str = "PIXELPASS_PHASE6_READY=";

    fn opts(
        app: bool,
        strict_audio: bool,
        capture_mode: CaptureMode,
        legacy_null_sink: bool,
    ) -> HostOpts {
        HostOpts {
            window: false,
            app: app.then(|| "Firefox".to_string()),
            strict_audio,
            display_server: None,
            quality: Quality::Auto,
            bitrate: None,
            framerate: None,
            max_height: None,
            no_hwencode: false,
            max_viewers: None,
            interactive: false,
            capture_mode,
            aec: AecConfig::Off,
            legacy_null_sink,
            relay: None,
        }
    }

    #[test]
    fn complete_phase_0d_mode_matrix() {
        let mut rows = 0;
        for app in [false, true] {
            for strict_audio in [false, true] {
                for capture_mode in [CaptureMode::Legacy, CaptureMode::DesktopExcluding] {
                    for legacy_null_sink in [false, true] {
                        rows += 1;
                        let opts = opts(app, strict_audio, capture_mode, legacy_null_sink);
                        let actual = CapturePlanKind::resolve(&opts);
                        let conflicts = capture_mode == CaptureMode::DesktopExcluding
                            && (app || legacy_null_sink);
                        if conflicts {
                            assert!(
                                actual.is_err(),
                                "conflicting row unexpectedly resolved: {opts:?}"
                            );
                            continue;
                        }
                        let expected = if capture_mode == CaptureMode::DesktopExcluding {
                            CapturePlanKind::DesktopExcluding
                        } else if app || legacy_null_sink {
                            CapturePlanKind::PerApp
                        } else {
                            CapturePlanKind::LegacyDesktop
                        };
                        assert_eq!(actual.unwrap(), expected, "wrong plan for {opts:?}");
                    }
                }
            }
        }
        assert_eq!(rows, 16, "the full 2×2×2×2 mode matrix must run");
    }

    #[test]
    fn default_monitor_is_confined_to_the_legacy_variant() {
        let plan = CapturePlan::LegacyDesktop {
            source: DefaultMonitor("alsa_output.fixture.monitor".to_string()),
        };
        assert_eq!(plan.gst_device_arg(), "device=alsa_output.fixture.monitor");
        assert!(matches!(plan, CapturePlan::LegacyDesktop { .. }));
    }

    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
    enum BackendCall {
        DefaultMonitor,
        Routing,
        BareCaptureSink,
    }

    struct FailingDesktopBackend {
        failure: Option<anyhow::Error>,
        calls: Vec<BackendCall>,
        seen_aec: Option<AecConfig>,
    }

    impl CapturePlanBackend for FailingDesktopBackend {
        async fn resolve_default_monitor(&mut self) -> Result<DefaultMonitor> {
            self.calls.push(BackendCall::DefaultMonitor);
            panic!("a DesktopExcluding failure must not resolve the legacy default monitor")
        }

        async fn start_routing(
            &mut self,
            _opts: &HostOpts,
            _health: health::Reporter,
        ) -> Result<Routing> {
            self.calls.push(BackendCall::Routing);
            panic!("a DesktopExcluding failure must not construct legacy Routing")
        }

        async fn start_bare_capture_sink(
            &mut self,
            _health: health::Reporter,
            aec: AecConfig,
        ) -> Result<BareCaptureSink> {
            self.calls.push(BackendCall::BareCaptureSink);
            self.seen_aec = Some(aec);
            Err(self.failure.take().expect("one injected failure"))
        }
    }

    struct Phase6MeasurementBackend {
        aec: AecConfig,
        policy: TestFanoutPolicy,
    }

    impl CapturePlanBackend for Phase6MeasurementBackend {
        async fn resolve_default_monitor(&mut self) -> Result<DefaultMonitor> {
            panic!("the Phase-6 measurement must use the hidden DesktopExcluding selector")
        }

        async fn start_routing(
            &mut self,
            _opts: &HostOpts,
            _health: health::Reporter,
        ) -> Result<Routing> {
            panic!("the Phase-6 measurement must not construct legacy Routing")
        }

        async fn start_bare_capture_sink(
            &mut self,
            health: health::Reporter,
            aec: AecConfig,
        ) -> Result<BareCaptureSink> {
            assert_eq!(
                aec, self.aec,
                "selector must carry the requested AEC config"
            );
            BareCaptureSink::start_for_phase6_measurement(health, aec, self.policy).await
        }
    }

    async fn assert_desktop_failure_is_closed(failure: anyhow::Error, expected_cause: &str) {
        let opts = opts(false, false, CaptureMode::DesktopExcluding, false);
        assert_eq!(
            CapturePlanKind::resolve(&opts).expect("fixture mode resolves"),
            CapturePlanKind::DesktopExcluding
        );
        let mut backend = FailingDesktopBackend {
            failure: Some(failure),
            calls: Vec::new(),
            seen_aec: None,
        };
        let (health, _) = health::channel();
        let error = match CapturePlan::start_with_backend(&opts, health, &mut backend).await {
            Ok(_) => panic!("DesktopExcluding unexpectedly recovered through another plan"),
            Err(error) => error,
        };
        assert_eq!(backend.calls, vec![BackendCall::BareCaptureSink]);
        assert_eq!(backend.seen_aec, Some(AecConfig::Off));
        let chain = format!("{error:#}");
        assert!(chain.contains("desktop-excluding capture-sink setup failed"));
        assert!(
            chain.contains(expected_cause),
            "unexpected error chain: {chain}"
        );
    }

    #[tokio::test]
    async fn capture_sink_creation_failure_never_falls_back_to_legacy_desktop() {
        assert_desktop_failure_is_closed(
            anyhow::anyhow!("injected capture-sink creation failure"),
            "injected capture-sink creation failure",
        )
        .await;
    }

    #[tokio::test]
    async fn readiness_epoch_timeout_never_falls_back_to_legacy_desktop() {
        let identity = super::super::graph::SinkIdentity {
            name: "pixelpass_capture_timeout_fixture".to_string(),
            monitor_name: "pixelpass_capture_timeout_fixture.monitor".to_string(),
            global_id: 77,
            serial: 88,
        };
        let projection = Projection {
            snapshot: Graph::new().build(),
            pipewire_pulse_pid: None,
            graph_ready: false,
            readiness: Readiness::TimedOut,
        };
        let readiness_error = super::super::graph::fanout_readiness(Some(&projection), &identity)
            .expect_err("the observer's sticky timeout must fail readiness");

        assert_desktop_failure_is_closed(
            readiness_error,
            "registry observer reached its sticky readiness timeout",
        )
        .await;
    }

    #[tokio::test]
    async fn public_aec_identity_reaches_the_bare_sink_constructor() {
        let mut opts = opts(false, false, CaptureMode::DesktopExcluding, false);
        opts.aec = AecConfig::PulseModule(536_870_919);
        let mut backend = FailingDesktopBackend {
            failure: Some(anyhow::anyhow!(
                "stop after observing the constructor input"
            )),
            calls: Vec::new(),
            seen_aec: None,
        };
        let (health, _) = health::channel();
        let error = match CapturePlan::start_with_backend(&opts, health, &mut backend).await {
            Ok(_) => panic!("injected backend unexpectedly constructed a plan"),
            Err(error) => error,
        };
        assert!(error.to_string().contains("desktop-excluding"));
        assert_eq!(backend.calls, vec![BackendCall::BareCaptureSink]);
        assert_eq!(backend.seen_aec, Some(opts.aec));
    }

    fn pulse_source_exists(name: &str) -> bool {
        std::process::Command::new("pactl")
            .args(["get-source-volume", name])
            .output()
            .is_ok_and(|output| output.status.success())
    }

    fn value_u64(value: &serde_json::Value) -> Option<u64> {
        value
            .as_u64()
            .or_else(|| value.as_str().and_then(|value| value.parse::<u64>().ok()))
    }

    fn native_links(output_name: &str, sink_name: &str) -> Result<Vec<serde_json::Value>> {
        let output = std::process::Command::new("pw-dump")
            .output()
            .context("run pw-dump for the phase-0d graph assertion")?;
        if !output.status.success() {
            bail!(
                "pw-dump failed: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
        }
        let objects: serde_json::Value =
            serde_json::from_slice(&output.stdout).context("parse pw-dump JSON")?;
        let objects = objects
            .as_array()
            .context("pw-dump root was not an array")?;
        let node_id = |name: &str| {
            objects
                .iter()
                .find(|object| {
                    object.get("type").and_then(serde_json::Value::as_str)
                        == Some("PipeWire:Interface:Node")
                        && object
                            .pointer("/info/props/node.name")
                            .and_then(serde_json::Value::as_str)
                            == Some(name)
                })
                .and_then(|object| object.get("id"))
                .and_then(value_u64)
        };
        let output_id = node_id(output_name).context("fixture output was absent from pw-dump")?;
        let sink_id = node_id(sink_name).context("bare capture sink was absent from pw-dump")?;

        Ok(objects
            .iter()
            .filter(|object| {
                if object.get("type").and_then(serde_json::Value::as_str)
                    != Some("PipeWire:Interface:Link")
                {
                    return false;
                }
                let Some(input_node) = object
                    .pointer("/info/props/link.input.node")
                    .and_then(value_u64)
                else {
                    return false;
                };
                let Some(output_node) = object
                    .pointer("/info/props/link.output.node")
                    .and_then(value_u64)
                else {
                    return false;
                };
                input_node == sink_id && output_node == output_id
            })
            .cloned()
            .collect())
    }

    fn object_serial(object: &serde_json::Value) -> Option<u64> {
        object
            .pointer("/info/props/object.serial")
            .and_then(value_u64)
    }

    fn object_serial_is_live(serial: u64) -> Result<bool> {
        let output = Command::new("pw-dump")
            .output()
            .context("run pw-dump for the Phase-6 SIGKILL gate")?;
        if !output.status.success() {
            bail!(
                "pw-dump failed: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
        }
        let objects: serde_json::Value =
            serde_json::from_slice(&output.stdout).context("parse pw-dump JSON")?;
        Ok(objects
            .as_array()
            .context("pw-dump root was not an array")?
            .iter()
            .any(|object| object_serial(object) == Some(serial)))
    }

    fn sink_serial(name: &str) -> Result<u64> {
        sink_global_identity(name).map(|(_, serial)| serial)
    }

    fn sink_global_identity(name: &str) -> Result<(u32, u64)> {
        let output = Command::new("pw-dump")
            .output()
            .context("run pw-dump for the Phase-6 sink identity")?;
        let objects: serde_json::Value =
            serde_json::from_slice(&output.stdout).context("parse pw-dump JSON")?;
        let object = objects
            .as_array()
            .context("pw-dump root was not an array")?
            .iter()
            .find(|object| {
                object
                    .pointer("/info/props/node.name")
                    .and_then(serde_json::Value::as_str)
                    == Some(name)
            })
            .context("capture sink was absent from pw-dump")?;
        let id = object
            .get("id")
            .and_then(value_u64)
            .and_then(|id| u32::try_from(id).ok())
            .context("capture sink had no usable global id")?;
        let serial = object_serial(object).context("capture sink had no object.serial")?;
        Ok((id, serial))
    }

    /// Subprocess half of the SIGKILL gate. The outer test kills this process,
    /// deliberately skipping every Rust destructor. Its GStreamer child uses
    /// the contained spawn path so parent death removes the fixture stream too.
    #[tokio::test]
    async fn phase6_sigkill_helper() {
        if std::env::var_os(PHASE6_SIGKILL_HELPER).is_none() {
            return;
        }

        let opts = opts(false, false, CaptureMode::DesktopExcluding, false);
        let sink_name = crate::repair::plan::sink_name_for(std::process::id());
        let (health, _) = health::channel();
        let _plan = CapturePlan::start(&opts, health)
            .await
            .expect("start Phase-6 SIGKILL helper plan");
        let fixture_name = format!("pixelpass_phase6_sigkill_{}", std::process::id());
        let mut command = Command::new("gst-launch-1.0");
        command
            .args([
                "-q",
                "audiotestsrc",
                "is-live=true",
                "volume=0",
                "!",
                "audioconvert",
                "!",
                "audio/x-raw,channels=2",
                "!",
                "pulsesink",
            ])
            .env("PULSE_PROP", format!("node.name={fixture_name}"))
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::inherit());
        // This helper is deliberately SIGKILLed by the outer test, so neither it nor this
        // retained child can reach a normal wait. `contained::spawn` gives the fixture a
        // parent-death signal; the outer test proves that path removes the fixture and links.
        #[allow(
            clippy::zombie_processes,
            reason = "the SIGKILL cleanup gate intentionally bypasses normal child waiting"
        )]
        let _fixture = crate::common::contained::spawn(&mut command)
            .expect("start contained Phase-6 SIGKILL fixture");

        let deadline = Instant::now() + Duration::from_secs(5);
        let links = loop {
            match native_links(&fixture_name, &sink_name) {
                Ok(links)
                    if links.len() == 2
                        && links.iter().all(|link| {
                            link.pointer("/info/state")
                                .and_then(serde_json::Value::as_str)
                                == Some("active")
                        }) =>
                {
                    break links;
                }
                _ if Instant::now() < deadline => {
                    tokio::time::sleep(Duration::from_millis(20)).await;
                }
                _ => panic!("SIGKILL fixture never reached two ACTIVE links"),
            }
        };
        let sink_serial = sink_serial(&sink_name).expect("read capture sink serial");
        let link_serials: Vec<u64> = links
            .iter()
            .map(|link| object_serial(link).expect("owned link object.serial"))
            .collect();
        println!(
            "{PHASE6_READY}{sink_name} {sink_serial} {},{}",
            link_serials[0], link_serials[1]
        );
        std::io::stdout().flush().expect("flush Phase-6 readiness");
        std::future::pending::<()>().await;
    }

    struct Phase6Host {
        child: Option<Child>,
        pid: u32,
    }

    impl Drop for Phase6Host {
        fn drop(&mut self) {
            if let Some(mut child) = self.child.take() {
                let _ = crate::common::contained::signal_group(self.pid, Signal::SIGKILL);
                let _ = child.wait();
            }
        }
    }

    #[test]
    #[ignore = "live: SIGKILLs a graph-mutating helper; run alone with --test-threads=1"]
    fn live_sigkill_removes_capture_sink_and_fanout_links() {
        let executable = std::env::current_exe().expect("locate test executable");
        let mut command = Command::new(executable);
        command
            .args([
                "--exact",
                "host::audio_plan::tests::phase6_sigkill_helper",
                "--nocapture",
                "--test-threads=1",
            ])
            .env(PHASE6_SIGKILL_HELPER, "1")
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::inherit());
        let child =
            crate::common::contained::spawn(&mut command).expect("spawn contained Phase-6 helper");
        let pid = child.id();
        let mut host = Phase6Host {
            child: Some(child),
            pid,
        };
        let stdout = host
            .child
            .as_mut()
            .expect("helper stayed owned")
            .stdout
            .take()
            .expect("helper stdout was piped");
        let (line_tx, line_rx) = std_mpsc::sync_channel(1);
        std::thread::spawn(move || {
            for line in BufReader::new(stdout).lines().map_while(Result::ok) {
                if line.contains(PHASE6_READY) {
                    let _ = line_tx.send(line);
                    return;
                }
            }
        });
        let line = line_rx
            .recv_timeout(Duration::from_secs(10))
            .expect("Phase-6 helper did not report readiness");
        let marker = line
            .split_once(PHASE6_READY)
            .map(|(_, marker)| marker)
            .expect("malformed Phase-6 readiness marker");
        let mut fields = marker.split_whitespace();
        let sink_name = fields
            .next()
            .expect("readiness omitted sink name")
            .to_string();
        let sink_serial: u64 = fields
            .next()
            .expect("readiness omitted sink serial")
            .parse()
            .expect("sink serial was not numeric");
        let link_serials: Vec<u64> = fields
            .next()
            .expect("readiness omitted link serials")
            .split(',')
            .map(|serial| serial.parse().expect("link serial was not numeric"))
            .collect();
        assert_eq!(link_serials.len(), 2);

        crate::common::contained::signal_group(pid, Signal::SIGKILL)
            .expect("SIGKILL the Phase-6 helper");
        host.child
            .as_mut()
            .expect("helper stayed owned")
            .wait()
            .expect("reap the Phase-6 helper");
        host.child.take();

        let deadline = Instant::now() + Duration::from_secs(3);
        while Instant::now() < deadline
            && (pulse_source_exists(&format!("{sink_name}.monitor"))
                || object_serial_is_live(sink_serial).unwrap_or(true)
                || link_serials
                    .iter()
                    .any(|serial| object_serial_is_live(*serial).unwrap_or(true)))
        {
            std::thread::sleep(Duration::from_millis(20));
        }
        assert!(!pulse_source_exists(&format!("{sink_name}.monitor")));
        assert!(!object_serial_is_live(sink_serial).expect("inspect sink residue"));
        for serial in link_serials {
            assert!(
                !object_serial_is_live(serial).expect("inspect link residue"),
                "fan-out link serial {serial} survived its owning process"
            );
        }
    }

    /// First Phase-6 live mutation gate: the hidden production path links a
    /// late eligible stream into the native sink without constructing the
    /// forbidden Pulse default-monitor loopback. Removing the stream revokes
    /// the link; shutting down removes the sink and every retained proxy.
    #[tokio::test]
    #[ignore = "live: mutates the shared PipeWire graph; run alone with --test-threads=1"]
    async fn live_desktop_excluding_fans_out_and_cleans_up_native_links() {
        let opts = opts(false, false, CaptureMode::DesktopExcluding, false);
        let sink_name = crate::repair::plan::sink_name_for(std::process::id());
        assert!(!pulse_source_exists(&format!("{sink_name}.monitor")));

        let (health, _) = health::channel();
        let plan = CapturePlan::start(&opts, health.clone())
            .await
            .expect("start the hidden desktop-excluding plan");
        let capture_sink = match &plan {
            CapturePlan::DesktopExcluding { capture_sink } => capture_sink,
            _ => panic!("hidden mode constructed the wrong capture-plan variant"),
        };
        assert_eq!(capture_sink.sink_name(), sink_name);
        assert!(pulse_source_exists(capture_sink.monitor_name()));

        let mut pulse = crate::repair::introspect::PulseSession::connect()
            .expect("connect to the local Pulse server");
        let modules = pulse.list_modules().expect("read the Pulse module table");
        let feeding_modules: Vec<_> = modules
            .into_iter()
            .filter(|module| {
                module.name == "module-loopback"
                    && module.args.contains(&format!("sink={sink_name}"))
            })
            .collect();
        assert!(
            feeding_modules.is_empty(),
            "DesktopExcluding must not construct a Pulse loopback into its sink: {feeding_modules:?}"
        );
        let fixture_name = format!("pixelpass_phase6_fixture_{}", std::process::id());
        let mut fixture = tokio::process::Command::new("gst-launch-1.0");
        fixture
            .args([
                "-q",
                "audiotestsrc",
                "is-live=true",
                "volume=0",
                "!",
                "audioconvert",
                "!",
                "audio/x-raw,channels=2",
                "!",
                "pulsesink",
            ])
            .env("PULSE_PROP", format!("node.name={fixture_name}"))
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::piped())
            .kill_on_drop(true);
        let mut fixture = fixture
            .spawn()
            .expect("start the late eligible audio fixture");

        let deadline = Instant::now() + Duration::from_secs(5);
        let owned_links = loop {
            match native_links(&fixture_name, &sink_name) {
                Ok(links)
                    if links.len() == 2
                        && links.iter().all(|link| {
                            link.pointer("/info/state")
                                .and_then(serde_json::Value::as_str)
                                == Some("active")
                        }) =>
                {
                    break Some(links);
                }
                Ok(_) | Err(_) if Instant::now() < deadline => {
                    tokio::time::sleep(Duration::from_millis(20)).await;
                }
                _ => break None,
            }
        };

        fixture
            .kill()
            .await
            .expect("stop the late eligible fixture");
        let deadline = Instant::now() + Duration::from_secs(2);
        let link_was_revoked = loop {
            if native_links(&fixture_name, &sink_name)
                .map(|links| links.is_empty())
                .unwrap_or(true)
            {
                break true;
            }
            if Instant::now() >= deadline {
                break false;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        };

        plan.shutdown().await;
        let deadline = Instant::now() + Duration::from_secs(2);
        while pulse_source_exists(&format!("{sink_name}.monitor")) && Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        assert!(!pulse_source_exists(&format!("{sink_name}.monitor")));
        assert!(health.fault().is_none());
        assert!(
            owned_links.is_some(),
            "the late eligible stereo stream never reached two ACTIVE native fan-out links"
        );
        assert!(
            link_was_revoked,
            "the owned fan-out link survived its output stream"
        );
    }

    /// Phase-6 row 7: remove the live connection-owned sink from the server,
    /// then prove the owner recreates it with a new serial and the fan-out
    /// controller drops both stale proxies before returning both replacement
    /// channel links to ACTIVE.
    #[tokio::test]
    #[ignore = "live: destroys and recreates the shared PipeWire capture sink; run alone with --test-threads=1"]
    async fn live_capture_sink_replacement_relinks_every_channel() {
        let opts = opts(false, false, CaptureMode::DesktopExcluding, false);
        let sink_name = crate::repair::plan::sink_name_for(std::process::id());
        let fixture_name = format!("pixelpass_phase6_replacement_{}", std::process::id());
        assert!(!pulse_source_exists(&format!("{sink_name}.monitor")));

        let (health, _) = health::channel();
        let plan = CapturePlan::start(&opts, health.clone())
            .await
            .expect("start desktop-excluding replacement fixture");
        let mut fixture = tokio::process::Command::new("gst-launch-1.0");
        fixture
            .args([
                "-q",
                "audiotestsrc",
                "is-live=true",
                "volume=0",
                "!",
                "audioconvert",
                "!",
                "audio/x-raw,channels=2",
                "!",
                "pulsesink",
            ])
            .env("PULSE_PROP", format!("node.name={fixture_name}"))
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::piped())
            .kill_on_drop(true);
        let mut fixture = fixture.spawn().expect("start replacement audio fixture");

        let initial_deadline = Instant::now() + Duration::from_secs(5);
        let old_links = loop {
            match native_links(&fixture_name, &sink_name) {
                Ok(links)
                    if links.len() == 2
                        && links.iter().all(|link| {
                            link.pointer("/info/state")
                                .and_then(serde_json::Value::as_str)
                                == Some("active")
                        }) =>
                {
                    break links;
                }
                _ if Instant::now() < initial_deadline => {
                    tokio::time::sleep(Duration::from_millis(20)).await;
                }
                _ => panic!("initial sink never reached two ACTIVE links"),
            }
        };
        let old_link_serials: Vec<u64> = old_links
            .iter()
            .map(|link| object_serial(link).expect("old link object.serial"))
            .collect();
        let (old_global_id, old_sink_serial) =
            sink_global_identity(&sink_name).expect("read initial sink identity");

        let destroy = Command::new("pw-cli")
            .args(["destroy", &old_global_id.to_string()])
            .status()
            .expect("run pw-cli destroy for the owned sink");
        assert!(
            destroy.success(),
            "pw-cli refused to destroy the owned sink"
        );

        let replacement_deadline = Instant::now() + Duration::from_secs(5);
        let (new_sink_serial, new_links) = loop {
            let identity = sink_global_identity(&sink_name);
            let links = native_links(&fixture_name, &sink_name);
            if let (Ok((_, serial)), Ok(links)) = (identity, links)
                && serial != old_sink_serial
                && links.len() == 2
                && links.iter().all(|link| {
                    link.pointer("/info/state")
                        .and_then(serde_json::Value::as_str)
                        == Some("active")
                })
            {
                break (serial, links);
            }
            if Instant::now() >= replacement_deadline {
                panic!(
                    "replacement sink never returned every channel link to ACTIVE; identity={:?}, links={:?}, health={:?}",
                    sink_global_identity(&sink_name),
                    native_links(&fixture_name, &sink_name).map(|links| {
                        links
                            .iter()
                            .map(|link| {
                                (
                                    object_serial(link),
                                    link.pointer("/info/state")
                                        .and_then(serde_json::Value::as_str)
                                        .map(str::to_string),
                                )
                            })
                            .collect::<Vec<_>>()
                    }),
                    health.fault(),
                );
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        };
        assert_ne!(new_sink_serial, old_sink_serial);
        let new_link_serials: Vec<u64> = new_links
            .iter()
            .map(|link| object_serial(link).expect("replacement link object.serial"))
            .collect();
        assert!(
            old_link_serials
                .iter()
                .all(|serial| !object_serial_is_live(*serial).unwrap_or(true)),
            "a stale old-sink link proxy remained live"
        );
        assert!(
            new_link_serials
                .iter()
                .all(|serial| !old_link_serials.contains(serial)),
            "replacement must bind fresh non-lingering link objects"
        );

        fixture
            .kill()
            .await
            .expect("stop replacement audio fixture");
        plan.shutdown().await;
        let residue_deadline = Instant::now() + Duration::from_secs(2);
        while pulse_source_exists(&format!("{sink_name}.monitor"))
            && Instant::now() < residue_deadline
        {
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        assert!(!pulse_source_exists(&format!("{sink_name}.monitor")));
        assert!(health.fault().is_none());
    }

    #[derive(Clone, Copy, Debug)]
    struct LeakArm {
        label: &'static str,
        play_remote_tone: bool,
        policy: TestFanoutPolicy,
    }

    #[derive(Debug)]
    struct LeakMeasurement {
        label: &'static str,
        desktop_dbfs: f64,
        remote_dbfs: f64,
        remote_floor_dbfs: f64,
    }

    const PN_SAMPLE_RATE: usize = 48_000;
    const PN_CHIP_FRAMES: usize = 24;
    const PN_SEQUENCE_CHIPS: usize = 4_096;
    const PN_WINDOW_CHIPS: usize = 1_024;
    const PN_FILE_SECONDS: usize = 8;
    const PN_SKIP_FRAMES: usize = PN_SAMPLE_RATE / 2;
    const DESKTOP_PN_SEED: u64 = 0x1357_9bdf;
    const REMOTE_PN_SEED: u64 = 0x2468_ace1;

    #[derive(Debug)]
    struct CorrelationMeasurement {
        label: &'static str,
        desktop_by_channel: [Vec<f64>; 2],
        remote_by_channel: [Vec<f64>; 2],
        xrun_delta: u64,
    }

    impl CorrelationMeasurement {
        fn desktop_min(&self) -> f64 {
            self.desktop_by_channel
                .iter()
                .flatten()
                .copied()
                .fold(f64::INFINITY, f64::min)
        }

        fn remote_min(&self) -> f64 {
            self.remote_by_channel
                .iter()
                .flatten()
                .copied()
                .fold(f64::INFINITY, f64::min)
        }

        fn remote_max(&self) -> f64 {
            self.remote_by_channel
                .iter()
                .flatten()
                .copied()
                .fold(f64::NEG_INFINITY, f64::max)
        }
    }

    struct PulseModuleGuard {
        id: Option<u64>,
        sink_name: String,
        source_name: String,
    }

    impl PulseModuleGuard {
        fn load(label: &str) -> Result<Self> {
            let suffix = format!("{}_{}", std::process::id(), label);
            let sink_name = format!("pixelpass_phase6_aec_{suffix}_sink");
            let source_name = format!("pixelpass_phase6_aec_{suffix}_source");
            let output = Command::new("pactl")
                .args([
                    "load-module",
                    "module-echo-cancel",
                    "aec_method=webrtc",
                    &format!("sink_name={sink_name}"),
                    &format!("source_name={source_name}"),
                ])
                .output()
                .context("load the controlled Phase-6 echo-cancel module")?;
            if !output.status.success() {
                bail!(
                    "pactl load-module module-echo-cancel failed: {}",
                    String::from_utf8_lossy(&output.stderr).trim()
                );
            }
            let id = String::from_utf8(output.stdout)
                .context("echo-cancel module id was not UTF-8")?
                .trim()
                .parse::<u64>()
                .context("echo-cancel module id was not a bare u64")?;
            Ok(Self {
                id: Some(id),
                sink_name,
                source_name,
            })
        }

        fn id(&self) -> u64 {
            self.id.expect("live Pulse module guard")
        }

        fn unload(&mut self) -> Result<()> {
            let Some(id) = self.id else {
                return Ok(());
            };
            let output = Command::new("pactl")
                .args(["unload-module", &id.to_string()])
                .output()
                .context("unload the exact Phase-6 echo-cancel module")?;
            if !output.status.success() {
                bail!(
                    "pactl unload-module {id} failed: {}",
                    String::from_utf8_lossy(&output.stderr).trim()
                );
            }
            self.id = None;
            Ok(())
        }
    }

    impl Drop for PulseModuleGuard {
        fn drop(&mut self) {
            if let Err(error) = self.unload() {
                eprintln!("Phase-6 module cleanup failed: {error:#}");
            }
        }
    }

    struct ContainedProcess {
        child: Option<Child>,
        pid: u32,
        label: &'static str,
    }

    impl ContainedProcess {
        fn spawn(command: &mut Command, label: &'static str) -> Result<Self> {
            let child = crate::common::contained::spawn(command)
                .with_context(|| format!("spawn contained {label}"))?;
            let pid = child.id();
            Ok(Self {
                child: Some(child),
                pid,
                label,
            })
        }

        fn ensure_running(&mut self) -> Result<()> {
            let status = self
                .child
                .as_mut()
                .context("process was already stopped")?
                .try_wait()
                .with_context(|| format!("inspect {}", self.label))?;
            if let Some(status) = status {
                let stderr = read_child_stderr(
                    self.child
                        .as_mut()
                        .expect("exited process stayed available for stderr"),
                );
                self.child.take();
                bail!("{} exited early with {status}: {stderr}", self.label);
            }
            Ok(())
        }

        fn stop(&mut self) -> Result<()> {
            let Some(mut child) = self.child.take() else {
                return Ok(());
            };
            let _ = crate::common::contained::signal_group(self.pid, Signal::SIGTERM);
            let deadline = Instant::now() + Duration::from_secs(2);
            let status = loop {
                if let Some(status) = child
                    .try_wait()
                    .with_context(|| format!("wait for {}", self.label))?
                {
                    break status;
                }
                if Instant::now() >= deadline {
                    let _ = crate::common::contained::signal_group(self.pid, Signal::SIGKILL);
                    break child
                        .wait()
                        .with_context(|| format!("reap {} after SIGKILL", self.label))?;
                }
                std::thread::sleep(Duration::from_millis(20));
            };
            let stderr = read_child_stderr(&mut child);
            if !stderr.trim().is_empty() {
                bail!("{} stopped with {status} and stderr: {stderr}", self.label);
            }
            Ok(())
        }
    }

    impl Drop for ContainedProcess {
        fn drop(&mut self) {
            if let Some(mut child) = self.child.take() {
                let _ = crate::common::contained::signal_group(self.pid, Signal::SIGKILL);
                let _ = child.wait();
                let stderr = read_child_stderr(&mut child);
                if !stderr.trim().is_empty() {
                    eprintln!("{} cleanup stderr: {stderr}", self.label);
                }
            }
        }
    }

    fn read_child_stderr(child: &mut Child) -> String {
        let mut stderr = String::new();
        if let Some(mut pipe) = child.stderr.take()
            && let Err(error) = pipe.read_to_string(&mut stderr)
        {
            return format!("<failed to read stderr: {error}>");
        }
        stderr.trim().to_string()
    }

    struct RawCaptureFile(PathBuf);

    impl RawCaptureFile {
        fn new(label: &str) -> Self {
            Self(std::env::temp_dir().join(format!(
                "pixelpass-phase6-{}-{label}.s16le",
                std::process::id()
            )))
        }
    }

    impl Drop for RawCaptureFile {
        fn drop(&mut self) {
            let _ = std::fs::remove_file(&self.0);
        }
    }

    fn pw_dump_objects(context: &str) -> Result<Vec<serde_json::Value>> {
        let output = Command::new("pw-dump")
            .output()
            .with_context(|| format!("run pw-dump for {context}"))?;
        if !output.status.success() {
            bail!(
                "pw-dump failed for {context}: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
        }
        serde_json::from_slice::<Vec<serde_json::Value>>(&output.stdout)
            .with_context(|| format!("parse pw-dump JSON for {context}"))
    }

    fn aec_playback_identity(module_id: u64) -> Result<(u64, String)> {
        let objects = pw_dump_objects("the Phase-6 AEC playback identity")?;
        // The role is deliberately filtered before the module identity. A
        // module-echo-cancel index labels four nodes; only its playback stream
        // is the hazardous fan-out candidate.
        let matches: Vec<(u64, String)> = objects
            .iter()
            .filter(|object| {
                object.get("type").and_then(serde_json::Value::as_str)
                    == Some("PipeWire:Interface:Node")
            })
            .filter(|object| {
                object
                    .pointer("/info/props/media.class")
                    .and_then(serde_json::Value::as_str)
                    == Some("Stream/Output/Audio")
            })
            .filter(|object| {
                object
                    .pointer("/info/props/pulse.module.id")
                    .and_then(value_u64)
                    == Some(module_id)
            })
            .filter_map(|object| {
                Some((
                    object.get("id").and_then(value_u64)?,
                    object
                        .pointer("/info/props/node.name")
                        .and_then(serde_json::Value::as_str)?
                        .to_string(),
                ))
            })
            .collect();
        match matches.as_slice() {
            [identity] => Ok(identity.clone()),
            [] => bail!("module {module_id} has no Stream/Output/Audio playback node"),
            _ => bail!("module {module_id} has multiple Stream/Output/Audio nodes: {matches:?}"),
        }
    }

    fn native_links_from_node_id(
        output_id: u64,
        sink_name: &str,
    ) -> Result<Vec<serde_json::Value>> {
        let objects = pw_dump_objects("the exact AEC playback links")?;
        let sink_id = objects
            .iter()
            .find(|object| {
                object.get("type").and_then(serde_json::Value::as_str)
                    == Some("PipeWire:Interface:Node")
                    && object
                        .pointer("/info/props/node.name")
                        .and_then(serde_json::Value::as_str)
                        == Some(sink_name)
            })
            .and_then(|object| object.get("id"))
            .and_then(value_u64)
            .context("link destination sink was absent from pw-dump")?;
        Ok(objects
            .into_iter()
            .filter(|object| {
                object.get("type").and_then(serde_json::Value::as_str)
                    == Some("PipeWire:Interface:Link")
                    && object
                        .pointer("/info/props/link.output.node")
                        .and_then(value_u64)
                        == Some(output_id)
                    && object
                        .pointer("/info/props/link.input.node")
                        .and_then(value_u64)
                        == Some(sink_id)
            })
            .collect())
    }

    fn every_link_active(links: &[serde_json::Value]) -> bool {
        links.iter().all(|link| {
            link.pointer("/info/state")
                .and_then(serde_json::Value::as_str)
                == Some("active")
        })
    }

    async fn wait_for_named_links(output: &str, sink: &str, expected: usize) -> Result<()> {
        let deadline = Instant::now() + Duration::from_secs(5);
        loop {
            if let Ok(links) = native_links(output, sink)
                && links.len() == expected
                && every_link_active(&links)
            {
                return Ok(());
            }
            if Instant::now() >= deadline {
                bail!(
                    "timed out waiting for {expected} ACTIVE links from {output} to {sink}: {:?}",
                    native_links(output, sink)
                );
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
    }

    async fn wait_for_id_links(output_id: u64, sink: &str, expected: usize) -> Result<()> {
        let deadline = Instant::now() + Duration::from_secs(5);
        loop {
            if let Ok(links) = native_links_from_node_id(output_id, sink)
                && links.len() == expected
                && every_link_active(&links)
            {
                return Ok(());
            }
            if Instant::now() >= deadline {
                bail!(
                    "timed out waiting for {expected} ACTIVE links from node {output_id} to {sink}: {:?}",
                    native_links_from_node_id(output_id, sink)
                );
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
    }

    async fn wait_for_id_links_present(output_id: u64, sink: &str, expected: usize) -> Result<()> {
        let deadline = Instant::now() + Duration::from_secs(5);
        loop {
            if let Ok(links) = native_links_from_node_id(output_id, sink)
                && links.len() == expected
            {
                return Ok(());
            }
            if Instant::now() >= deadline {
                bail!(
                    "timed out waiting for {expected} links from node {output_id} to {sink}: {:?}",
                    native_links_from_node_id(output_id, sink)
                );
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
    }

    async fn wait_for_aec_playback(module_id: u64) -> Result<(u64, String)> {
        let deadline = Instant::now() + Duration::from_secs(5);
        loop {
            match aec_playback_identity(module_id) {
                Ok(identity) => return Ok(identity),
                Err(_) if Instant::now() < deadline => {
                    tokio::time::sleep(Duration::from_millis(20)).await;
                }
                Err(error) => return Err(error),
            }
        }
    }

    fn default_sink_name() -> Result<String> {
        let output = Command::new("pactl")
            .arg("get-default-sink")
            .output()
            .context("read the default sink for the Phase-6 graph proof")?;
        if !output.status.success() {
            bail!(
                "pactl get-default-sink failed: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
        }
        let name = String::from_utf8(output.stdout)
            .context("default sink name was not UTF-8")?
            .trim()
            .to_string();
        if name.is_empty() {
            bail!("pactl get-default-sink returned an empty name");
        }
        Ok(name)
    }

    fn start_tone(
        frequency: u32,
        node_name: &str,
        sink: Option<&str>,
        owned: bool,
    ) -> Result<ContainedProcess> {
        let mut command = Command::new("gst-launch-1.0");
        command.args([
            "-q",
            "audiotestsrc",
            "is-live=true",
            "wave=sine",
            &format!("freq={frequency}"),
            "volume=0.02",
            "!",
            "audioconvert",
            "!",
            "audioresample",
            "!",
            "audio/x-raw,rate=48000,channels=2",
            "!",
            "pulsesink",
            "sync=true",
        ]);
        if let Some(sink) = sink {
            command.arg(format!("device={sink}"));
        }
        let pulse_props = if owned {
            format!("node.name={node_name} peerspeak.owned=1")
        } else {
            format!("node.name={node_name}")
        };
        command
            .env("PULSE_PROP", pulse_props)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::piped());
        ContainedProcess::spawn(&mut command, "Phase-6 tone generator")
    }

    fn pn_sequence(mut state: u64) -> Vec<f64> {
        assert_ne!(state, 0, "PN seed must be non-zero");
        (0..PN_SEQUENCE_CHIPS)
            .map(|_| {
                // Deterministic xorshift64 sequence. The two Phase-9 seeds are
                // separately pinned and their maximum cyclic correlation is a
                // pure gate below; no statistical quality is assumed.
                state ^= state << 13;
                state ^= state >> 7;
                state ^= state << 17;
                if state & 1 == 0 { -1.0 } else { 1.0 }
            })
            .collect()
    }

    fn write_pn_probe(label: &str, chips: &[f64]) -> Result<RawCaptureFile> {
        let raw = RawCaptureFile::new(&format!("phase9-{label}-probe"));
        let file = std::fs::File::create(&raw.0)
            .with_context(|| format!("create PN probe {}", raw.0.display()))?;
        let mut writer = std::io::BufWriter::new(file);
        let frames = PN_SAMPLE_RATE * PN_FILE_SECONDS;
        for frame in 0..frames {
            let chip = chips[(frame / PN_CHIP_FRAMES) % chips.len()];
            let sample = (chip * 0.02 * f64::from(i16::MAX)).round() as i16;
            let bytes = sample.to_le_bytes();
            writer.write_all(&bytes)?;
            writer.write_all(&bytes)?;
        }
        writer.flush()?;
        Ok(raw)
    }

    fn start_pn_probe(
        raw: &RawCaptureFile,
        node_name: &str,
        sink: Option<&str>,
        owned: bool,
    ) -> Result<ContainedProcess> {
        let mut command = Command::new("gst-launch-1.0");
        command.args([
            "-q",
            "filesrc",
            &format!("location={}", raw.0.display()),
            "!",
            "rawaudioparse",
            "format=pcm",
            "pcm-format=s16le",
            "sample-rate=48000",
            "num-channels=2",
            "!",
            "audioconvert",
            "!",
            "audioresample",
            "!",
            "audio/x-raw,rate=48000,channels=2",
            "!",
            "pulsesink",
            "sync=true",
        ]);
        if let Some(sink) = sink {
            command.arg(format!("device={sink}"));
        }
        let pulse_props = if owned {
            format!("node.name={node_name} peerspeak.owned=1")
        } else {
            format!("node.name={node_name}")
        };
        command
            .env("PULSE_PROP", pulse_props)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::piped());
        ContainedProcess::spawn(&mut command, "Phase-9 PN generator")
    }

    fn normalized_correlation(window: &[f64], probe: &[f64], lag: usize) -> f64 {
        let window_mean = window.iter().sum::<f64>() / window.len() as f64;
        let probe_mean = (0..window.len())
            .map(|index| probe[(lag + index) % probe.len()])
            .sum::<f64>()
            / window.len() as f64;
        let mut dot = 0.0;
        let mut window_energy = 0.0;
        let mut probe_energy = 0.0;
        for (index, sample) in window.iter().enumerate() {
            let x = *sample - window_mean;
            let y = probe[(lag + index) % probe.len()] - probe_mean;
            dot += x * y;
            window_energy += x * x;
            probe_energy += y * y;
        }
        if window_energy <= f64::EPSILON || probe_energy <= f64::EPSILON {
            return 0.0;
        }
        (dot / (window_energy * probe_energy).sqrt()).abs()
    }

    fn max_cyclic_correlation(window: &[f64], probe: &[f64]) -> f64 {
        (0..probe.len())
            .map(|lag| normalized_correlation(window, probe, lag))
            .fold(0.0, f64::max)
    }

    fn capture_channel_chips(raw: &[u8], channel: usize) -> Result<Vec<f64>> {
        const CHANNELS: usize = 2;
        let frame_bytes = std::mem::size_of::<i16>() * CHANNELS;
        let frames = raw.len() / frame_bytes;
        if frames <= PN_SKIP_FRAMES + PN_WINDOW_CHIPS * PN_CHIP_FRAMES {
            bail!("capture was too short for PN analysis: {frames} stereo frames");
        }
        let chip_count = (frames - PN_SKIP_FRAMES) / PN_CHIP_FRAMES;
        let mut chips = Vec::with_capacity(chip_count);
        for chip in 0..chip_count {
            let first_frame = PN_SKIP_FRAMES + chip * PN_CHIP_FRAMES;
            let mut sum = 0.0;
            for offset in 0..PN_CHIP_FRAMES {
                let byte = (first_frame + offset) * frame_bytes + channel * 2;
                sum += f64::from(i16::from_le_bytes([raw[byte], raw[byte + 1]]));
            }
            chips.push(sum / PN_CHIP_FRAMES as f64 / f64::from(i16::MAX));
        }
        Ok(chips)
    }

    fn windowed_channel_correlations(raw: &[u8], probe: &[f64]) -> Result<[Vec<f64>; 2]> {
        let analyze = |channel| -> Result<Vec<f64>> {
            let chips = capture_channel_chips(raw, channel)?;
            let correlations = chips
                .chunks_exact(PN_WINDOW_CHIPS)
                .map(|window| max_cyclic_correlation(window, probe))
                .collect::<Vec<_>>();
            if correlations.len() < 3 {
                bail!(
                    "capture yielded only {} complete PN windows on channel {channel}",
                    correlations.len()
                );
            }
            Ok(correlations)
        };
        Ok([analyze(0)?, analyze(1)?])
    }

    fn pw_top_errors() -> Result<HashMap<u64, u64>> {
        let output = Command::new("pw-top")
            .args(["--batch-mode", "--iterations=1"])
            .output()
            .context("snapshot PipeWire xrun counters with pw-top")?;
        if !output.status.success() {
            bail!(
                "pw-top failed: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
        }
        let mut errors = HashMap::new();
        for line in String::from_utf8_lossy(&output.stdout).lines().skip(1) {
            let fields = line.split_whitespace().collect::<Vec<_>>();
            if fields.len() >= 9
                && let (Ok(id), Ok(count)) = (fields[1].parse::<u64>(), fields[8].parse::<u64>())
            {
                errors.insert(id, count);
            }
        }
        if errors.is_empty() {
            bail!("pw-top returned no parseable PipeWire xrun counters");
        }
        Ok(errors)
    }

    fn node_ids_named(names: &[&str]) -> Result<HashSet<u64>> {
        let objects = pw_dump_objects("the Phase-9 xrun node set")?;
        let wanted = names.iter().copied().collect::<HashSet<_>>();
        let found = objects
            .iter()
            .filter_map(|object| {
                let name = object
                    .pointer("/info/props/node.name")
                    .and_then(serde_json::Value::as_str)?;
                wanted
                    .contains(name)
                    .then(|| object.get("id").and_then(value_u64))?
            })
            .collect::<HashSet<_>>();
        if found.len() != wanted.len() {
            bail!(
                "Phase-9 xrun snapshot resolved {} of {} required nodes ({names:?})",
                found.len(),
                wanted.len()
            );
        }
        Ok(found)
    }

    fn xrun_delta(
        before: &HashMap<u64, u64>,
        after: &HashMap<u64, u64>,
        relevant: &HashSet<u64>,
    ) -> u64 {
        relevant
            .iter()
            .map(|id| {
                after
                    .get(id)
                    .copied()
                    .unwrap_or_default()
                    .saturating_sub(before.get(id).copied().unwrap_or_default())
            })
            .sum()
    }

    async fn record_capture(monitor_name: &str, label: &str) -> Result<Vec<u8>> {
        let raw = RawCaptureFile::new(label);
        let output_file =
            std::fs::File::create(&raw.0).with_context(|| format!("create {}", raw.0.display()))?;
        let mut command = Command::new("parec");
        command
            .args([
                "--record",
                "--raw",
                &format!("--device={monitor_name}"),
                "--rate=48000",
                "--format=s16le",
                "--channels=2",
                "--client-name=pixelpass-phase6-measurement",
            ])
            .stdin(Stdio::null())
            .stdout(Stdio::from(output_file))
            .stderr(Stdio::piped());
        let mut recorder = ContainedProcess::spawn(&mut command, "Phase-6 parec recorder")?;
        tokio::time::sleep(Duration::from_secs(3)).await;
        recorder.ensure_running()?;
        recorder.stop()?;
        std::fs::read(&raw.0).with_context(|| format!("read {}", raw.0.display()))
    }

    fn tone_dbfs(raw: &[u8], frequency: f64) -> Result<f64> {
        const SAMPLE_RATE: f64 = 48_000.0;
        const CHANNELS: usize = 2;
        let frames = raw.len() / (std::mem::size_of::<i16>() * CHANNELS);
        if frames < SAMPLE_RATE as usize {
            bail!("capture was too short: {frames} stereo frames");
        }
        // Drop startup transients, then use a Hann-windowed exact-frequency
        // projection. This is a gross-leak qualification, not the Phase-9
        // PN/MLS intelligibility rig.
        let skip = SAMPLE_RATE as usize / 2;
        let count = frames - skip;
        let mut real = 0.0;
        let mut imag = 0.0;
        let mut window_sum = 0.0;
        for index in 0..count {
            let frame = skip + index;
            let offset = frame * 4;
            let left = i16::from_le_bytes([raw[offset], raw[offset + 1]]) as f64;
            let right = i16::from_le_bytes([raw[offset + 2], raw[offset + 3]]) as f64;
            let sample = (left + right) / 2.0;
            let window = 0.5 - 0.5 * (2.0 * PI * index as f64 / (count - 1) as f64).cos();
            let phase = 2.0 * PI * frequency * index as f64 / SAMPLE_RATE;
            real += sample * window * phase.cos();
            imag -= sample * window * phase.sin();
            window_sum += window;
        }
        let amplitude = 2.0 * real.hypot(imag) / window_sum / i16::MAX as f64;
        Ok(20.0 * amplitude.max(1.0e-12).log10())
    }

    fn local_floor_dbfs(raw: &[u8], frequency: f64) -> Result<f64> {
        // A single coherent projection of stochastic sub-LSB noise can land
        // in a deep null and is not a stable "control floor". Estimate this
        // run's actual analysis resolution from nearby frequencies outside the
        // Hann main lobe, then use the 90th percentile so one quiet bin cannot
        // manufacture sensitivity the capture did not have.
        let mut neighbors = Vec::new();
        for offset in (5..=25).step_by(2) {
            neighbors.push(tone_dbfs(raw, frequency - f64::from(offset))?);
            neighbors.push(tone_dbfs(raw, frequency + f64::from(offset))?);
        }
        neighbors.sort_by(f64::total_cmp);
        let rank = (neighbors.len() * 9).div_ceil(10).saturating_sub(1);
        Ok(neighbors[rank])
    }

    fn any_echo_cancel_module_loaded() -> Result<bool> {
        let output = Command::new("pactl")
            .args(["list", "short", "modules"])
            .output()
            .context("inspect pre-existing Pulse modules")?;
        if !output.status.success() {
            bail!(
                "pactl list short modules failed: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
        }
        Ok(String::from_utf8_lossy(&output.stdout).contains("module-echo-cancel"))
    }

    async fn run_leak_arm(arm: LeakArm) -> Result<LeakMeasurement> {
        let mut aec = PulseModuleGuard::load(arm.label)?;
        let module_id = aec.id();
        let (aec_playback_id, aec_playback_name) = wait_for_aec_playback(module_id).await?;
        let capture_sink_name = crate::repair::plan::sink_name_for(std::process::id());
        let mut opts = opts(false, false, CaptureMode::DesktopExcluding, false);
        opts.aec = AecConfig::PulseModule(module_id);
        let (health, _) = health::channel();
        let mut backend = Phase6MeasurementBackend {
            aec: AecConfig::PulseModule(module_id),
            policy: arm.policy,
        };
        let plan = CapturePlan::start_with_backend(&opts, health.clone(), &mut backend)
            .await
            .with_context(|| format!("start the {} production-path capture plan", arm.label))?;
        let capture_sink = match &plan {
            CapturePlan::DesktopExcluding { capture_sink } => capture_sink,
            _ => bail!("{} arm did not select DesktopExcluding", arm.label),
        };
        let monitor_name = capture_sink.monitor_name().to_string();
        let default_sink = default_sink_name()?;
        let desktop_name = format!("pixelpass_phase6_{}_440", arm.label);
        let remote_name = format!("peerspeak_owned_phase6_{}_1500", arm.label);
        let mut desktop = start_tone(440, &desktop_name, None, false)?;
        let mut remote = if arm.play_remote_tone {
            Some(start_tone(1_500, &remote_name, Some(&aec.sink_name), true)?)
        } else {
            None
        };

        wait_for_named_links(&desktop_name, &capture_sink_name, 2).await?;
        wait_for_named_links(&desktop_name, &default_sink, 2).await?;
        if let Some(remote) = remote.as_mut() {
            wait_for_named_links(&remote_name, &aec.sink_name, 2).await?;
            wait_for_named_links(&remote_name, &capture_sink_name, 0).await?;
            remote.ensure_running()?;
        }
        // This is the load-bearing policy observation: the same exact module
        // playback node has zero capture links in control/guarded and two only
        // in the deliberately-naive test arm. Its normal speaker link remains
        // live in all three arms.
        if arm.play_remote_tone {
            wait_for_id_links(aec_playback_id, &default_sink, 2).await?;
        } else {
            // With no 1500 Hz input the module is correctly idle: the two
            // passive speaker links exist but PipeWire reports them paused.
            wait_for_id_links_present(aec_playback_id, &default_sink, 2).await?;
        }
        let expected_aec_links = if arm.policy == TestFanoutPolicy::IncludeConfiguredAec {
            2
        } else {
            0
        };
        wait_for_id_links(aec_playback_id, &capture_sink_name, expected_aec_links).await?;
        desktop.ensure_running()?;

        let raw = record_capture(&monitor_name, arm.label).await?;
        let measurement = LeakMeasurement {
            label: arm.label,
            desktop_dbfs: tone_dbfs(&raw, 440.0)?,
            remote_dbfs: tone_dbfs(&raw, 1_500.0)?,
            remote_floor_dbfs: local_floor_dbfs(&raw, 1_500.0)?,
        };

        if let Some(remote) = remote.as_mut() {
            remote.stop()?;
        }
        desktop.stop()?;
        plan.shutdown().await;
        aec.unload()?;
        let residue_deadline = Instant::now() + Duration::from_secs(3);
        while (pulse_source_exists(&monitor_name)
            || pulse_source_exists(&aec.source_name)
            || Command::new("pactl")
                .args(["get-sink-volume", &aec.sink_name])
                .output()
                .is_ok_and(|output| output.status.success()))
            && Instant::now() < residue_deadline
        {
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        if pulse_source_exists(&monitor_name)
            || pulse_source_exists(&aec.source_name)
            || Command::new("pactl")
                .args(["get-sink-volume", &aec.sink_name])
                .output()
                .is_ok_and(|output| output.status.success())
        {
            bail!("{} arm left a Pulse source or sink behind", arm.label);
        }
        if let Some(fault) = health.fault() {
            bail!("{} arm poisoned audio health: {fault}", arm.label);
        }
        eprintln!(
            "Phase-6 {}: 440 Hz {:.2} dBFS, 1500 Hz {:.2} dBFS (local floor {:.2} dBFS); AEC playback node {} ({})",
            measurement.label,
            measurement.desktop_dbfs,
            measurement.remote_dbfs,
            measurement.remote_floor_dbfs,
            aec_playback_name,
            aec_playback_id,
        );
        Ok(measurement)
    }

    async fn run_correlation_arm(
        arm: LeakArm,
        desktop_probe: &[f64],
        remote_probe: &[f64],
    ) -> Result<CorrelationMeasurement> {
        let mut aec = PulseModuleGuard::load(arm.label)?;
        let module_id = aec.id();
        let (aec_playback_id, aec_playback_name) = wait_for_aec_playback(module_id).await?;
        let capture_sink_name = crate::repair::plan::sink_name_for(std::process::id());
        let mut opts = opts(false, false, CaptureMode::DesktopExcluding, false);
        opts.aec = AecConfig::PulseModule(module_id);
        let (health, _) = health::channel();
        let mut backend = Phase6MeasurementBackend {
            aec: AecConfig::PulseModule(module_id),
            policy: arm.policy,
        };
        let plan = CapturePlan::start_with_backend(&opts, health.clone(), &mut backend)
            .await
            .with_context(|| format!("start the {} Phase-9 capture plan", arm.label))?;
        let capture_sink = match &plan {
            CapturePlan::DesktopExcluding { capture_sink } => capture_sink,
            _ => bail!("{} arm did not select DesktopExcluding", arm.label),
        };
        let monitor_name = capture_sink.monitor_name().to_string();
        let default_sink = default_sink_name()?;
        let desktop_name = format!("pixelpass_phase9_{}_desktop_pn", arm.label);
        let remote_name = format!("peerspeak_owned_phase9_{}_remote_pn", arm.label);
        let desktop_file = write_pn_probe(&format!("{}-desktop", arm.label), desktop_probe)?;
        let remote_file = write_pn_probe(&format!("{}-remote", arm.label), remote_probe)?;
        let mut desktop = start_pn_probe(&desktop_file, &desktop_name, None, false)?;
        let mut remote = if arm.play_remote_tone {
            Some(start_pn_probe(
                &remote_file,
                &remote_name,
                Some(&aec.sink_name),
                true,
            )?)
        } else {
            None
        };

        wait_for_named_links(&desktop_name, &capture_sink_name, 2).await?;
        wait_for_named_links(&desktop_name, &default_sink, 2).await?;
        if let Some(remote) = remote.as_mut() {
            wait_for_named_links(&remote_name, &aec.sink_name, 2).await?;
            wait_for_named_links(&remote_name, &capture_sink_name, 0).await?;
            remote.ensure_running()?;
        }
        if arm.play_remote_tone {
            wait_for_id_links(aec_playback_id, &default_sink, 2).await?;
        } else {
            wait_for_id_links_present(aec_playback_id, &default_sink, 2).await?;
        }
        let expected_aec_links = if arm.policy == TestFanoutPolicy::IncludeConfiguredAec {
            2
        } else {
            0
        };
        wait_for_id_links(aec_playback_id, &capture_sink_name, expected_aec_links).await?;
        desktop.ensure_running()?;

        let mut xrun_names = vec![
            capture_sink_name.as_str(),
            default_sink.as_str(),
            desktop_name.as_str(),
            aec_playback_name.as_str(),
        ];
        if arm.play_remote_tone {
            xrun_names.push(remote_name.as_str());
            xrun_names.push(aec.sink_name.as_str());
        }
        let relevant_xrun_nodes = node_ids_named(&xrun_names)?;
        let xruns_before = pw_top_errors()?;
        let raw = record_capture(&monitor_name, &format!("phase9-{}", arm.label)).await?;
        let xruns_after = pw_top_errors()?;

        // Original routes must still be intact after the measurement, not only
        // during setup. The capture links remain the exact safe/naive partition.
        wait_for_named_links(&desktop_name, &default_sink, 2).await?;
        wait_for_named_links(&desktop_name, &capture_sink_name, 2).await?;
        if arm.play_remote_tone {
            wait_for_named_links(&remote_name, &aec.sink_name, 2).await?;
            wait_for_id_links(aec_playback_id, &default_sink, 2).await?;
            wait_for_id_links(aec_playback_id, &capture_sink_name, expected_aec_links).await?;
        }

        let measurement = CorrelationMeasurement {
            label: arm.label,
            desktop_by_channel: windowed_channel_correlations(&raw, desktop_probe)?,
            remote_by_channel: windowed_channel_correlations(&raw, remote_probe)?,
            xrun_delta: xrun_delta(&xruns_before, &xruns_after, &relevant_xrun_nodes),
        };

        if let Some(remote) = remote.as_mut() {
            remote.stop()?;
        }
        desktop.stop()?;
        plan.shutdown().await;
        aec.unload()?;
        let residue_deadline = Instant::now() + Duration::from_secs(3);
        while (pulse_source_exists(&monitor_name)
            || pulse_source_exists(&aec.source_name)
            || Command::new("pactl")
                .args(["get-sink-volume", &aec.sink_name])
                .output()
                .is_ok_and(|output| output.status.success()))
            && Instant::now() < residue_deadline
        {
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        if pulse_source_exists(&monitor_name)
            || pulse_source_exists(&aec.source_name)
            || Command::new("pactl")
                .args(["get-sink-volume", &aec.sink_name])
                .output()
                .is_ok_and(|output| output.status.success())
        {
            bail!("{} arm left a Pulse source or sink behind", arm.label);
        }
        if let Some(fault) = health.fault() {
            bail!("{} arm poisoned audio health: {fault}", arm.label);
        }
        eprintln!(
            "Phase-9 {}: desktop correlations L={:?} R={:?}; remote correlations L={:?} R={:?}; xrun delta={}",
            measurement.label,
            measurement.desktop_by_channel[0],
            measurement.desktop_by_channel[1],
            measurement.remote_by_channel[0],
            measurement.remote_by_channel[1],
            measurement.xrun_delta,
        );
        Ok(measurement)
    }

    #[test]
    fn phase9_pn_probes_are_orthogonal_and_the_correlator_finds_phase() {
        let desktop = pn_sequence(DESKTOP_PN_SEED);
        let remote = pn_sequence(REMOTE_PN_SEED);
        let cross = max_cyclic_correlation(&desktop, &remote);
        assert!(
            cross <= 0.08,
            "the pinned PN pair is not sufficiently orthogonal: {cross:.4}"
        );

        let shift = 733;
        let shifted = (0..PN_WINDOW_CHIPS)
            .map(|index| desktop[(shift + index) % desktop.len()])
            .collect::<Vec<_>>();
        assert!(max_cyclic_correlation(&shifted, &desktop) > 0.999);
        assert!(max_cyclic_correlation(&vec![0.0; PN_WINDOW_CHIPS], &desktop) == 0.0);
    }

    /// Phase-9 rig upgrade: the Phase-6 production-path topology is driven by
    /// two pinned, independently generated PN probes. Every non-overlapping
    /// half-second window is normalized and correlated separately on left and
    /// right while searching all cyclic lags. The absent/control and guarded
    /// arms bound false correlation; the naive arm proves the excluded probe is
    /// detectable. `pw-top` snapshots pin zero new xruns on every involved node.
    #[tokio::test]
    #[ignore = "live: plays low-level PN probes and mutates the shared audio graph; run alone with --test-threads=1"]
    async fn live_phase9_three_arm_pn_correlation_measurement() -> Result<()> {
        if any_echo_cancel_module_loaded()? {
            bail!("refusing the controlled measurement while another module-echo-cancel is live");
        }
        let desktop_probe = pn_sequence(DESKTOP_PN_SEED);
        let remote_probe = pn_sequence(REMOTE_PN_SEED);

        // Declared before the run: every eligible-probe window/channel must be
        // strongly present; every absent/excluded-probe window/channel must stay
        // below the fixed criterion; the naive arm must recover the remote probe;
        // and no involved PipeWire node may increment its ERR/xrun counter.
        const MIN_PRESENT_CORRELATION: f64 = 0.25;
        const MAX_EXCLUDED_CORRELATION: f64 = 0.20;
        const MAX_XRUN_DELTA: u64 = 0;

        let control = run_correlation_arm(
            LeakArm {
                label: "pn-control",
                play_remote_tone: false,
                policy: TestFanoutPolicy::Safe,
            },
            &desktop_probe,
            &remote_probe,
        )
        .await?;
        let guarded = run_correlation_arm(
            LeakArm {
                label: "pn-guarded",
                play_remote_tone: true,
                policy: TestFanoutPolicy::Safe,
            },
            &desktop_probe,
            &remote_probe,
        )
        .await?;
        let naive = run_correlation_arm(
            LeakArm {
                label: "pn-naive",
                play_remote_tone: true,
                policy: TestFanoutPolicy::IncludeConfiguredAec,
            },
            &desktop_probe,
            &remote_probe,
        )
        .await?;

        for measurement in [&control, &guarded, &naive] {
            assert!(
                measurement.desktop_min() >= MIN_PRESENT_CORRELATION,
                "eligible desktop probe missing from a window/channel: {measurement:?}"
            );
            assert_eq!(
                measurement.xrun_delta, MAX_XRUN_DELTA,
                "measurement introduced an xrun: {measurement:?}"
            );
        }
        assert!(
            control.remote_max() <= MAX_EXCLUDED_CORRELATION,
            "control falsely detected the absent remote probe: {control:?}"
        );
        assert!(
            guarded.remote_max() <= MAX_EXCLUDED_CORRELATION,
            "guarded capture detected the excluded remote probe: {guarded:?}"
        );
        assert!(
            naive.remote_min() >= MIN_PRESENT_CORRELATION,
            "naive positive control missed the remote probe in a window/channel: {naive:?}"
        );
        Ok(())
    }

    /// Final Phase-6 qualification: record the real hidden selector → native
    /// sink → fan-out observer/link-manager → Pulse monitor source path. The
    /// naive arm changes only the test-compiled AEC-identity decision, making
    /// it a positive control for a gross remote-call leak.
    #[tokio::test]
    #[ignore = "live: plays low-level tones and mutates the shared audio graph; run alone with --test-threads=1"]
    async fn live_phase6_three_arm_aec_leak_measurement() -> Result<()> {
        if any_echo_cancel_module_loaded()? {
            bail!("refusing the controlled measurement while another module-echo-cancel is live");
        }
        let control = run_leak_arm(LeakArm {
            label: "control",
            play_remote_tone: false,
            policy: TestFanoutPolicy::Safe,
        })
        .await?;
        let guarded = run_leak_arm(LeakArm {
            label: "guarded",
            play_remote_tone: true,
            policy: TestFanoutPolicy::Safe,
        })
        .await?;
        let naive = run_leak_arm(LeakArm {
            label: "naive",
            play_remote_tone: true,
            policy: TestFanoutPolicy::IncludeConfiguredAec,
        })
        .await?;

        // Thresholds are declared before interpreting this run. They qualify
        // gross steady-state leakage only; Phase 9 replaces tones with PN/MLS
        // material for intelligibility and field variance.
        const DESKTOP_PRESENT_DBFS: f64 = -65.0;
        const GUARDED_DESKTOP_DRIFT_DB: f64 = 6.0;
        const GUARDED_FLOOR_TOLERANCE_DB: f64 = 3.0;
        const POSITIVE_CONTROL_MARGIN_DB: f64 = 18.0;
        const NAIVE_TONE_BALANCE_DB: f64 = 10.0;
        assert!(control.desktop_dbfs > DESKTOP_PRESENT_DBFS, "{control:?}");
        assert!(guarded.desktop_dbfs > DESKTOP_PRESENT_DBFS, "{guarded:?}");
        assert!(naive.desktop_dbfs > DESKTOP_PRESENT_DBFS, "{naive:?}");
        assert!(
            (guarded.desktop_dbfs - control.desktop_dbfs).abs() <= GUARDED_DESKTOP_DRIFT_DB,
            "guarded desktop level drifted from control: control={control:?}, guarded={guarded:?}"
        );
        let resolved_control_floor = control.remote_dbfs.max(control.remote_floor_dbfs);
        assert!(
            guarded.remote_dbfs <= resolved_control_floor + GUARDED_FLOOR_TOLERANCE_DB,
            "guarded 1500 Hz energy rose above the locally resolved control floor ({resolved_control_floor:.2} dBFS): control={control:?}, guarded={guarded:?}"
        );
        assert!(
            naive.remote_dbfs >= control.remote_dbfs + POSITIVE_CONTROL_MARGIN_DB
                && naive.remote_dbfs >= guarded.remote_dbfs + POSITIVE_CONTROL_MARGIN_DB,
            "the naive positive control did not expose a gross 1500 Hz leak: control={control:?}, guarded={guarded:?}, naive={naive:?}"
        );
        assert!(
            (naive.remote_dbfs - naive.desktop_dbfs).abs() <= NAIVE_TONE_BALANCE_DB,
            "the naive arm did not capture both tones at comparable levels: {naive:?}"
        );
        Ok(())
    }
}