feat: echo cancellation via PipeWire module-echo-cancel

Add an opt-in Echo Cancellation toggle (Settings) that routes the call
through PipeWire's module-echo-cancel (WebRTC AEC + noise suppression + AGC)
instead of running an in-process canceller. PipeWire already sees both the
mic and the speaker monitor, so it handles the echo-reference alignment for
free and we avoid a C++ DSP dependency.

src/audio/echo_cancel.rs (new):
- enable(real_source, real_sink) loads the module via pactl (aec_method=webrtc),
  bound to the chosen devices with source_master/sink_master (defaults if unset),
  waits for the virtual nodes to appear, and returns an RAII guard that unloads
  the module on drop. Best-effort pre-clean of a stale instance from a crashed run.
- EC_SOURCE / EC_SINK are the virtual cleaned-mic source and reference sink.

core: when echo_cancellation is set on Join, load the module and point capture
at EC_SOURCE / playback at EC_SINK; stash the guard in ActiveSession so it
unloads on shutdown (after the audio streams release the nodes). Any failure
logs + warns the UI and falls back to the direct devices — never blocks the call.

config: new echo_cancellation_enabled (serde default false). app: Settings
checkbox under Mic Sensitivity, applied on next room join. An ignored live
smoke test covers the real load/unload path.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/app/mod.rs

@@ -46,6 +46,7 @@ pub enum AppMessage {
     NavigateToSettings,
     NavigateBack,
     ToggleNotifications(bool),
+    ToggleEchoCancellation(bool),
     CustomSoundPathChanged(Sound, String),
 }
 
@@ -196,6 +197,7 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
                     ticket: state.ticket_input.clone(),
                     input_device,
                     output_device,
+                    echo_cancellation: state.config.echo_cancellation_enabled,
                 });
             }
         }
@@ -208,6 +210,7 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
                 ticket: "create".to_string(),
                 input_device,
                 output_device,
+                echo_cancellation: state.config.echo_cancellation_enabled,
             });
         }
         AppMessage::LeavePressed => {
@@ -331,6 +334,11 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
             state.config.save();
             notify::set_enabled(enabled);
         }
+        AppMessage::ToggleEchoCancellation(enabled) => {
+            state.config.echo_cancellation_enabled = enabled;
+            state.config.save();
+            // Applied on the next join, since the audio graph is rebuilt then.
+        }
         AppMessage::CustomSoundPathChanged(sound, path) => {
             let path_opt = if path.trim().is_empty() { None } else { Some(path) };
             match sound {
@@ -515,6 +523,11 @@ fn view(state: &AppState) -> Element<'_, AppMessage> {
                         text(format!("Mic Sensitivity (Noise Gate): {:.1}%", state.config.noise_gate_threshold * 100.0)).size(14).color(color_subtext),
                         slider(0.0..=0.1, state.config.noise_gate_threshold, AppMessage::NoiseGateChanged).step(0.001),
                         text("Smoothly fades out audio below this level. 0% disables the gate.").size(11).color(color_subtext),
+                        vertical_space(4.0),
+                        checkbox(state.config.echo_cancellation_enabled)
+                            .label("Echo cancellation")
+                            .on_toggle(AppMessage::ToggleEchoCancellation),
+                        text("Cancels speaker echo + suppresses noise (PipeWire). Takes effect on your next room join.").size(11).color(color_subtext),
                     ].spacing(8).width(iced::Length::Fill),
                     column![
                         text("Network Privacy").size(14).color(color_subtext),

src/audio/echo_cancel.rs

@@ -0,0 +1,166 @@
+//! Acoustic echo cancellation via PipeWire's `module-echo-cancel`.
+//!
+//! Rather than running an echo canceller in-process (which would mean a C++ DSP
+//! dependency plus plumbing the playout signal back to the capture path as a
+//! delay-aligned reference), we delegate to the audio server. PipeWire already
+//! sees both the real microphone and the speaker monitor, so loading
+//! `module-echo-cancel` (WebRTC AEC + noise suppression + AGC) solves the
+//! reference-alignment problem for free.
+//!
+//! When enabled, the module creates two virtual nodes:
+//! - [`EC_SOURCE`] — the cleaned microphone (we capture from this).
+//! - [`EC_SINK`] — playout goes here; it doubles as the AEC reference (we play to
+//!   this, and the module forwards it to the real output while using it to cancel
+//!   the echo from the mic).
+//!
+//! Binding to a specific mic/speaker (the user's chosen devices) is done with the
+//! `source_master`/`sink_master` args; omitting them binds to the system defaults.
+//! The loaded module is owned by an [`EchoCancelGuard`] that unloads it on drop, so
+//! it never outlives a call (even if the join path bails out early).
+
+use std::process::Command;
+use std::time::{Duration, Instant};
+
+/// node.name of the virtual (cleaned) capture source the module creates.
+pub const EC_SOURCE: &str = "peerspeak_echocancel_source";
+/// node.name of the virtual playback sink (also the AEC reference).
+pub const EC_SINK: &str = "peerspeak_echocancel_sink";
+
+/// How long to wait for the virtual nodes to appear after loading the module
+/// before giving up — the nodes show up a beat after `load-module` returns.
+const NODE_READY_TIMEOUT: Duration = Duration::from_secs(3);
+
+/// Owns a loaded `module-echo-cancel` instance; unloads it on drop so the virtual
+/// nodes never leak past the call that created them.
+pub struct EchoCancelGuard {
+    module_index: String,
+}
+
+impl Drop for EchoCancelGuard {
+    fn drop(&mut self) {
+        let _ = Command::new("pactl")
+            .arg("unload-module")
+            .arg(&self.module_index)
+            .output();
+        crate::log_msg(&format!("Echo cancel: unloaded module {}", self.module_index));
+    }
+}
+
+/// Loads `module-echo-cancel` (WebRTC AEC) bound to the given real devices, waits
+/// for its virtual nodes to come up, and returns a guard that unloads it on drop.
+///
+/// `real_source`/`real_sink` are the node.name of the chosen mic/speaker; pass
+/// `None` (or an empty string) to bind to the system defaults. Returns `Err` with
+/// a human-readable reason if `pactl` is missing, the load fails, or the nodes
+/// don't appear — the caller should fall back to the direct devices.
+pub fn enable(real_source: Option<&str>, real_sink: Option<&str>) -> Result<EchoCancelGuard, String> {
+    // Best-effort: clear any stale instance left by a crashed prior run so we
+    // don't stack duplicate modules / fight over the virtual node names.
+    unload_stale();
+
+    let mut cmd = Command::new("pactl");
+    cmd.arg("load-module")
+        .arg("module-echo-cancel")
+        .arg("aec_method=webrtc")
+        .arg(format!("source_name={EC_SOURCE}"))
+        .arg(format!("sink_name={EC_SINK}"));
+    if let Some(src) = real_source.filter(|s| !s.is_empty()) {
+        cmd.arg(format!("source_master={src}"));
+    }
+    if let Some(sink) = real_sink.filter(|s| !s.is_empty()) {
+        cmd.arg(format!("sink_master={sink}"));
+    }
+
+    let out = cmd
+        .output()
+        .map_err(|e| format!("pactl not available: {e}"))?;
+    if !out.status.success() {
+        return Err(format!(
+            "pactl load-module failed: {}",
+            String::from_utf8_lossy(&out.stderr).trim()
+        ));
+    }
+
+    let module_index = String::from_utf8_lossy(&out.stdout).trim().to_string();
+    if module_index.parse::<u64>().is_err() {
+        return Err(format!("unexpected pactl output: {module_index:?}"));
+    }
+    let guard = EchoCancelGuard { module_index };
+
+    // The virtual nodes appear shortly after the module loads; wait for both so
+    // the subsequent capture/playback streams can actually target them. If they
+    // never show, drop the guard (unloads) and report failure.
+    if !wait_for_nodes() {
+        return Err("echo-cancel virtual nodes did not appear in time".to_string());
+    }
+
+    crate::log_msg(&format!(
+        "Echo cancel: loaded module {} (source_master={:?}, sink_master={:?})",
+        guard.module_index, real_source, real_sink
+    ));
+    Ok(guard)
+}
+
+/// Polls until both virtual nodes exist or the timeout elapses.
+fn wait_for_nodes() -> bool {
+    let deadline = Instant::now() + NODE_READY_TIMEOUT;
+    loop {
+        if node_present("sources", EC_SOURCE) && node_present("sinks", EC_SINK) {
+            return true;
+        }
+        if Instant::now() >= deadline {
+            return false;
+        }
+        std::thread::sleep(Duration::from_millis(100));
+    }
+}
+
+/// Whether `pactl list <kind> short` lists a node named `name`.
+/// `kind` is "sources" or "sinks".
+fn node_present(kind: &str, name: &str) -> bool {
+    let Ok(out) = Command::new("pactl").arg("list").arg(kind).arg("short").output() else {
+        return false;
+    };
+    String::from_utf8_lossy(&out.stdout)
+        .lines()
+        .any(|line| line.split('\t').nth(1) == Some(name))
+}
+
+/// Unloads any leftover `module-echo-cancel` instance we previously created
+/// (identified by our virtual node names in its argument string). Best-effort.
+fn unload_stale() {
+    let Ok(out) = Command::new("pactl").arg("list").arg("modules").arg("short").output() else {
+        return;
+    };
+    for line in String::from_utf8_lossy(&out.stdout).lines() {
+        let mut cols = line.split('\t');
+        let index = cols.next().unwrap_or("");
+        let name = cols.next().unwrap_or("");
+        let args = cols.next().unwrap_or("");
+        if name == "module-echo-cancel" && args.contains(EC_SOURCE) && index.parse::<u64>().is_ok() {
+            let _ = Command::new("pactl").arg("unload-module").arg(index).output();
+            crate::log_msg(&format!("Echo cancel: cleaned up stale module {index}"));
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Exercises the real load -> wait-for-nodes -> drop/unload path against the
+    /// live PipeWire daemon. Side-effecting (briefly creates virtual audio nodes),
+    /// so it's ignored by default; run with `cargo test -- --ignored echo_cancel`.
+    #[test]
+    #[ignore]
+    fn enable_creates_and_unloads_nodes() {
+        let guard = enable(None, None).expect("module-echo-cancel should load");
+        assert!(node_present("sources", EC_SOURCE), "cleaned source must exist");
+        assert!(node_present("sinks", EC_SINK), "reference sink must exist");
+        drop(guard);
+        // Give pactl a moment to tear the nodes down.
+        std::thread::sleep(Duration::from_millis(300));
+        assert!(!node_present("sources", EC_SOURCE), "source must be gone after unload");
+        assert!(!node_present("sinks", EC_SINK), "sink must be gone after unload");
+    }
+}

src/audio/mod.rs

@@ -51,6 +51,7 @@ pub trait AudioBackend: Send + Sync {
     fn stop(&self) -> Result<(), AudioError>;
 }
 
+pub mod echo_cancel;
 pub mod gate;
 pub mod pipewire_impl;
 pub mod pw_cli;

src/config.rs

@@ -47,6 +47,10 @@ pub struct AppConfig {
     pub noise_gate_threshold: f32,
     #[serde(default)]
     pub network_mode: NetworkMode,
+    /// Route audio through PipeWire's echo-cancel module (AEC + noise suppression).
+    /// Takes effect on the next room join. Off by default.
+    #[serde(default)]
+    pub echo_cancellation_enabled: bool,
     #[serde(default = "default_true")]
     pub notifications_enabled: bool,
     #[serde(default)]
@@ -74,6 +78,7 @@ impl Default for AppConfig {
             output_device: "".to_string(),
             noise_gate_threshold: 0.01,
             network_mode: NetworkMode::default(),
+            echo_cancellation_enabled: false,
             notifications_enabled: true,
             custom_sound_self_join: None,
             custom_sound_peer_join: None,

src/core/messages.rs

@@ -4,7 +4,7 @@ use iroh::EndpointId;
 
 #[derive(Debug, Clone)]
 pub enum CoreCommand {
-    Join { name: String, ticket: String, input_device: Option<String>, output_device: Option<String> },
+    Join { name: String, ticket: String, input_device: Option<String>, output_device: Option<String>, echo_cancellation: bool },
     Leave,
     ToggleMute,
     ToggleDeafen,

src/core/mod.rs

@@ -203,6 +203,8 @@ struct ActiveSession {
     conn_event_task: tokio::task::JoinHandle<()>,
     grace_timers: GraceTimers,
     transport: Arc<IrohTransport>,
+    /// Loaded PipeWire echo-cancel module (if enabled); unloads on drop.
+    echo_cancel: Option<crate::audio::echo_cancel::EchoCancelGuard>,
 }
 
 impl ActiveSession {
@@ -226,6 +228,10 @@ impl ActiveSession {
             crate::log_msg("Audio backend stopped");
         }).await;
 
+        // Unload the echo-cancel module now that the audio streams releasing its
+        // virtual nodes have stopped. (Dropping the guard runs `pactl unload`.)
+        drop(self.echo_cancel);
+
         crate::log_msg("Leaving room...");
         let _ = self.room_state.leave().await;
         // Close peer links with the graceful goodbye code so remotes evict us
@@ -266,7 +272,7 @@ async fn run_core_loop(
 
     while let Some(cmd) = cmd_rx.recv().await {
         match cmd {
-            CoreCommand::Join { name, ticket, input_device, output_device } => {
+            CoreCommand::Join { name, ticket, input_device, output_device, echo_cancellation } => {
                 current_name = name.clone();
 
                 // Clean up any existing session
@@ -363,7 +369,41 @@ async fn run_core_loop(
                 let (capture_tx, capture_rx) = std::sync::mpsc::channel();
                 let (playback_tx, playback_rx) = std::sync::mpsc::channel();
 
-                if let Err(e) = audio_backend.start_capture(capture_tx, input_device) {
+                // Echo cancellation: if enabled, load PipeWire's echo-cancel module
+                // bound to the chosen real devices and route capture/playback
+                // through its virtual nodes (the sink doubles as the AEC reference).
+                // The guard unloads the module on drop — including the early-return
+                // paths below, since it's a local until moved into the session. On
+                // any failure, warn and fall back to the direct devices.
+                let mut echo_cancel_guard = None;
+                let (capture_target, playback_target) = if echo_cancellation {
+                    match crate::audio::echo_cancel::enable(
+                        input_device.as_deref(),
+                        output_device.as_deref(),
+                    ) {
+                        Ok(guard) => {
+                            echo_cancel_guard = Some(guard);
+                            crate::log_msg("Echo cancellation enabled");
+                            (
+                                Some(crate::audio::echo_cancel::EC_SOURCE.to_string()),
+                                Some(crate::audio::echo_cancel::EC_SINK.to_string()),
+                            )
+                        }
+                        Err(e) => {
+                            crate::log_msg(&format!(
+                                "Echo cancellation unavailable, using direct devices: {e}"
+                            ));
+                            let _ = ui_tx
+                                .send(UiEvent::Error(format!("Echo cancellation unavailable: {e}")))
+                                .await;
+                            (input_device.clone(), output_device.clone())
+                        }
+                    }
+                } else {
+                    (input_device.clone(), output_device.clone())
+                };
+
+                if let Err(e) = audio_backend.start_capture(capture_tx, capture_target) {
                     let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await;
                     let _ = room_state.leave().await;
                     let _ = router.shutdown().await;
@@ -374,7 +414,7 @@ async fn run_core_loop(
                 // here (drain side + fill side); the mixer reads it to pace
                 // production to the hardware clock instead of a fixed timer.
                 let ring_fill = Arc::new(AtomicUsize::new(0));
-                if let Err(e) = audio_backend.start_playback(playback_rx, output_device, ring_fill.clone()) {
+                if let Err(e) = audio_backend.start_playback(playback_rx, playback_target, ring_fill.clone()) {
                     let _ = ui_tx.send(UiEvent::Error(format!("Failed to start playback: {}", e))).await;
                     let _ = audio_backend.stop();
                     let _ = room_state.leave().await;
@@ -675,6 +715,7 @@ async fn run_core_loop(
                     conn_event_task,
                     grace_timers,
                     transport: transport.clone(),
+                    echo_cancel: echo_cancel_guard,
                 };
 
                 let self_id = endpoint.id().to_string();