diff --git a/src/host/audio_plan.rs b/src/host/audio_plan.rs index 60f9fd5..e45d4a1 100644 --- a/src/host/audio_plan.rs +++ b/src/host/audio_plan.rs @@ -206,6 +206,7 @@ mod tests { 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; @@ -1015,6 +1016,49 @@ mod tests { 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; 2], + remote_by_channel: [Vec; 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, sink_name: String, @@ -1418,6 +1462,215 @@ mod tests { ContainedProcess::spawn(&mut command, "Phase-6 tone generator") } + fn pn_sequence(mut state: u64) -> Vec { + 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 { + 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 { + 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::() / window.len() as f64; + let probe_mean = (0..window.len()) + .map(|index| probe[(lag + index) % probe.len()]) + .sum::() + / 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> { + const CHANNELS: usize = 2; + let frame_bytes = std::mem::size_of::() * 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; 2]> { + let analyze = |channel| -> Result> { + let chips = capture_channel_chips(raw, channel)?; + let correlations = chips + .chunks_exact(PN_WINDOW_CHIPS) + .map(|window| max_cyclic_correlation(window, probe)) + .collect::>(); + 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> { + 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::>(); + if fields.len() >= 9 + && let (Ok(id), Ok(count)) = (fields[1].parse::(), fields[8].parse::()) + { + 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> { + let objects = pw_dump_objects("the Phase-9 xrun node set")?; + let wanted = names.iter().copied().collect::>(); + 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::>(); + 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, + after: &HashMap, + relevant: &HashSet, + ) -> 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> { let raw = RawCaptureFile::new(label); let output_file = @@ -1609,6 +1862,237 @@ mod tests { Ok(measurement) } + async fn run_correlation_arm( + arm: LeakArm, + desktop_probe: &[f64], + remote_probe: &[f64], + ) -> 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 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::>(); + 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