test(host): qualify Phase 6 AEC leak path

src/host/audio_plan.rs

@@ -199,10 +199,14 @@ impl CapturePlan {
 mod tests {
     use super::*;
     use crate::cli::Quality;
+    use crate::host::aec::AecConfig;
+    use crate::host::fanout::TestFanoutPolicy;
     use crate::host::observer::{Projection, Readiness};
     use crate::host::taint::fixture::Graph;
     use nix::sys::signal::Signal;
-    use std::io::{BufRead, BufReader, Write};
+    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};
@@ -313,6 +317,32 @@ mod tests {
         }
     }
 
+    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,
+        ) -> Result<BareCaptureSink> {
+            BareCaptureSink::start_for_phase6_measurement(health, self.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!(
@@ -935,4 +965,653 @@ mod tests {
         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,
+    }
+
+    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")
+    }
+
+    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 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 opts = opts(false, false, CaptureMode::DesktopExcluding, false);
+        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)?,
+        };
+
+        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; AEC playback node {} ({})",
+            measurement.label,
+            measurement.desktop_dbfs,
+            measurement.remote_dbfs,
+            aec_playback_name,
+            aec_playback_id,
+        );
+        Ok(measurement)
+    }
+
+    /// 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:?}"
+        );
+        assert!(
+            guarded.remote_dbfs <= control.remote_dbfs + GUARDED_FLOOR_TOLERANCE_DB,
+            "guarded 1500 Hz energy rose above the control floor: 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(())
+    }
 }

src/host/fanout.rs

@@ -15,6 +15,8 @@ use std::collections::{BTreeMap, BTreeSet};
 
 use super::aec::{AecConfig, AecState, AecValidator};
 use super::observer::{EventKind, Millis, Projection};
+#[cfg(test)]
+use super::taint::Eligibility;
 use super::taint::snapshot::{
     GlobalId, GraphSnapshot, MediaRole, PortDirection, PortSnapshot, Serial,
 };
@@ -181,6 +183,33 @@ pub(super) struct FanoutController {
     states: BTreeMap<Serial, StreamCaptureState>,
     foreign_aec_groups: BTreeSet<String>,
     status_tx: Option<mpsc::UnboundedSender<AudioExclusionEvent>>,
+    #[cfg(test)]
+    test_policy: TestFanoutPolicy,
+}
+
+/// Test-only control for the Phase-6 leak measurement. The shipping binary
+/// has no representation of the deliberately unsafe policy: this field,
+/// constructor, and rewrite are all removed before production type checking.
+#[cfg(test)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub(super) enum TestFanoutPolicy {
+    #[default]
+    Safe,
+    IncludeConfiguredAec,
+}
+
+#[cfg(test)]
+impl TestFanoutPolicy {
+    fn apply(self, mut decisions: Decisions) -> Decisions {
+        if self == Self::IncludeConfiguredAec {
+            for decision in decisions.candidates.values_mut() {
+                if decision.reason() == Some(Reason::AecIdentity) {
+                    decision.eligibility = Eligibility::Eligible;
+                }
+            }
+        }
+        decisions
+    }
 }
 
 impl FanoutController {
@@ -189,6 +218,18 @@ impl FanoutController {
         Self::build(capture_sink, aec, None)
     }
 
+    #[cfg(test)]
+    pub(super) fn with_test_policy(
+        capture_sink: Serial,
+        aec: AecConfig,
+        status_tx: mpsc::UnboundedSender<AudioExclusionEvent>,
+        test_policy: TestFanoutPolicy,
+    ) -> Self {
+        let mut controller = Self::build(capture_sink, aec, Some(status_tx));
+        controller.test_policy = test_policy;
+        controller
+    }
+
     /// Construct the production controller with a non-blocking status path.
     /// The PipeWire callback only enqueues owned records; JSON serialization
     /// and stdout I/O happen on a Tokio task outside the graph loop.
@@ -212,6 +253,8 @@ impl FanoutController {
             states: BTreeMap::new(),
             foreign_aec_groups: BTreeSet::new(),
             status_tx,
+            #[cfg(test)]
+            test_policy: TestFanoutPolicy::Safe,
         }
     }
 
@@ -276,6 +319,8 @@ impl MutationProjectionSink for FanoutController {
             graph_ready: projection.graph_ready,
         };
         let (decisions, sticky) = evaluate(&projection.snapshot, &ctx, &self.sticky);
+        #[cfg(test)]
+        let decisions = self.test_policy.apply(decisions);
         self.sticky = sticky;
         let plans = plan(&projection.snapshot, &decisions, self.capture_sink);
 
@@ -1289,6 +1334,45 @@ mod tests {
         );
     }
 
+    #[test]
+    fn phase6_naive_control_rewrites_only_the_configured_aec_candidate() {
+        let (mut graph, app, sink, _) = stereo_graph();
+        let aec = graph.module_node("aec-playback", MediaRole::StreamOutput, 999);
+        graph.port_on_channel(aec, PortDirection::Out, false, Some("FL"));
+        graph.port_on_channel(aec, PortDirection::Out, false, Some("FR"));
+
+        let mut safe_links = FakeLinks::default();
+        let (status_tx, _status_rx) = mpsc::unbounded_channel();
+        let mut safe = FanoutController::with_test_policy(
+            sink,
+            AecConfig::PulseModule(999),
+            status_tx,
+            TestFanoutPolicy::Safe,
+        );
+        drive(&mut safe, &graph, &mut safe_links);
+        assert_eq!(safe_links.held.len(), 2);
+        assert_eq!(safe.states()[&app], StreamCaptureState::Captured);
+        assert_eq!(
+            safe.states()[&aec.serial],
+            StreamCaptureState::Excluded {
+                reason: Reason::AecIdentity,
+            }
+        );
+
+        let mut naive_links = FakeLinks::default();
+        let (status_tx, _status_rx) = mpsc::unbounded_channel();
+        let mut naive = FanoutController::with_test_policy(
+            sink,
+            AecConfig::PulseModule(999),
+            status_tx,
+            TestFanoutPolicy::IncludeConfiguredAec,
+        );
+        drive(&mut naive, &graph, &mut naive_links);
+        assert_eq!(naive_links.held.len(), 4);
+        assert_eq!(naive.states()[&app], StreamCaptureState::Captured);
+        assert_eq!(naive.states()[&aec.serial], StreamCaptureState::Captured);
+    }
+
     #[test]
     fn second_echo_cancel_group_emits_one_foreign_warning() {
         use crate::host::taint::fixture::pulse_module;

src/host/graph.rs

@@ -9,7 +9,9 @@
 
 use super::aec::AecConfig;
 use super::audio::parse_object_serial;
-use super::fanout::FanoutController;
+#[cfg(test)]
+use super::fanout::TestFanoutPolicy;
+use super::fanout::{FanoutController, MutationProjectionSink};
 use super::health;
 use super::observer::adapter::{FanoutControl, RegistryObserverHandle};
 use super::observer::{Projection, Readiness};
@@ -159,6 +161,40 @@ pub(super) struct BareCaptureSink {
 
 impl BareCaptureSink {
     pub(super) async fn start(health: health::Reporter) -> Result<Self> {
+        Self::start_with_controller(health, |capture_sink, status_tx| {
+            Box::new(FanoutController::with_status_sender(
+                capture_sink,
+                AecConfig::Off,
+                status_tx,
+            ))
+        })
+        .await
+    }
+
+    #[cfg(test)]
+    pub(super) async fn start_for_phase6_measurement(
+        health: health::Reporter,
+        aec: AecConfig,
+        test_policy: TestFanoutPolicy,
+    ) -> Result<Self> {
+        Self::start_with_controller(health, move |capture_sink, status_tx| {
+            Box::new(FanoutController::with_test_policy(
+                capture_sink,
+                aec,
+                status_tx,
+                test_policy,
+            ))
+        })
+        .await
+    }
+
+    async fn start_with_controller<F>(health: health::Reporter, controller: F) -> Result<Self>
+    where
+        F: FnOnce(
+            Serial,
+            mpsc::UnboundedSender<AudioExclusionEvent>,
+        ) -> Box<dyn MutationProjectionSink>,
+    {
         let spec = CaptureSinkSpec::for_pid(std::process::id());
         let (graph_owner, _event_rx, identity_rx) =
             AudioGraphOwner::start(None, spec, health.clone())
@@ -168,11 +204,7 @@ impl BareCaptureSink {
         let (status_tx, status_rx) = mpsc::unbounded_channel();
         let status_forwarder = tokio::spawn(forward_audio_exclusion_status(status_rx));
         let fanout = match RegistryObserverHandle::spawn_with_mutation_sink(
-            Box::new(FanoutController::with_status_sender(
+            controller(Serial(identity.serial), status_tx),
-                Serial(identity.serial),
-                AecConfig::Off,
-                status_tx,
-            )),
             health.clone(),
         )
         .context("failed to start the desktop-excluding fan-out observer")