From e027bc65e525e479252f6a155000e86847fb484d Mon Sep 17 00:00:00 2001 From: Mollusk Date: Fri, 21 Aug 2026 20:59:08 -0400 Subject: [PATCH] test(host): qualify Phase 6 AEC leak path --- src/host/audio_plan.rs | 681 ++++++++++++++++++++++++++++++++++++++++- src/host/fanout.rs | 84 +++++ src/host/graph.rs | 44 ++- 3 files changed, 802 insertions(+), 7 deletions(-) diff --git a/src/host/audio_plan.rs b/src/host/audio_plan.rs index 88ca789..5218afc 100644 --- a/src/host/audio_plan.rs +++ b/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 { + panic!("the Phase-6 measurement must use the hidden DesktopExcluding selector") + } + + async fn start_routing( + &mut self, + _opts: &HostOpts, + _health: health::Reporter, + ) -> Result { + panic!("the Phase-6 measurement must not construct legacy Routing") + } + + async fn start_bare_capture_sink( + &mut self, + health: health::Reporter, + ) -> Result { + 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, + sink_name: String, + source_name: String, + } + + impl PulseModuleGuard { + fn load(label: &str) -> Result { + 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::() + .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, + pid: u32, + label: &'static str, + } + + impl ContainedProcess { + fn spawn(command: &mut Command, label: &'static str) -> Result { + 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!(""); + } + 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> { + 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::>(&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> { + 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 { + 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 { + 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> { + 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 { + const SAMPLE_RATE: f64 = 48_000.0; + const CHANNELS: usize = 2; + let frames = raw.len() / (std::mem::size_of::() * 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 { + 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 { + 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(()) + } } diff --git a/src/host/fanout.rs b/src/host/fanout.rs index 27537a2..8cc2100 100644 --- a/src/host/fanout.rs +++ b/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, foreign_aec_groups: BTreeSet, status_tx: Option>, + #[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, + 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; diff --git a/src/host/graph.rs b/src/host/graph.rs index e1d2639..a1e7b44 100644 --- a/src/host/graph.rs +++ b/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::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::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(health: health::Reporter, controller: F) -> Result + where + F: FnOnce( + Serial, + mpsc::UnboundedSender, + ) -> Box, + { 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( - Serial(identity.serial), - AecConfig::Off, - status_tx, - )), + controller(Serial(identity.serial), status_tx), health.clone(), ) .context("failed to start the desktop-excluding fan-out observer")