From 5a65f50c4b8881af4d06c15ef3f755afb39adc19 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Fri, 21 Aug 2026 16:31:11 -0400 Subject: [PATCH] feat(host): emit desktop audio exclusion status events --- src/common/output.rs | 127 +++++++++++++++++++++++ src/host/aec/mod.rs | 8 ++ src/host/fanout.rs | 242 +++++++++++++++++++++++++++++++++++++++++-- src/host/graph.rs | 37 ++++++- 4 files changed, 402 insertions(+), 12 deletions(-) diff --git a/src/common/output.rs b/src/common/output.rs index 2431d00..2dee599 100644 --- a/src/common/output.rs +++ b/src/common/output.rs @@ -17,6 +17,14 @@ use serde::Serialize; static JSON_ENABLED: AtomicBool = AtomicBool::new(false); +/// First wire version for the desktop-audio-exclusion status family. +/// +/// Existing event tags predate explicit versioning. These events are consumed +/// across the PixelPass/PeerSpeak process boundary and are landing before the +/// PeerSpeak parser, so their version is carried on every record rather than +/// inferred from either binary's package version. +pub(crate) const AUDIO_EXCLUSION_EVENT_VERSION: u8 = 1; + /// Turn JSON event output on. Called once at startup from `--output json`. pub fn set_json(enabled: bool) { JSON_ENABLED.store(enabled, Ordering::Relaxed); @@ -63,6 +71,21 @@ pub enum Event<'a> { /// stream went away. Under `--strict-audio`, `lost` means viewers currently /// hear silence; without it, viewers fall back to whole-desktop audio. AppAudio { state: AppAudioState }, + /// One otherwise-eligible playback stream could not be linked safely. + StreamUnsupported { + version: u8, + stream_serial: u64, + reason: &'a str, + }, + /// The configured AEC identity never appeared before its validation + /// deadline. Fan-out remains fail-closed. + AecFailed { version: u8, module_index: u64 }, + /// A previously validated AEC identity disappeared. Every owned fan-out + /// link is revoked before this transition is reported. + AecRevoked { version: u8, module_index: u64 }, + /// An echo-cancel group other than the configured PeerSpeak instance is + /// present and excluded from fan-out. + ForeignAecWarning { version: u8, link_group: &'a str }, } #[derive(Serialize)] @@ -79,6 +102,58 @@ pub enum AppAudioState { Lost, } +/// Owned form of the audio-exclusion status family. The PipeWire observer can +/// enqueue this through an unbounded sender without borrowing its snapshot; +/// a Tokio-side forwarder then converts it to the public JSON [`Event`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum AudioExclusionEvent { + StreamUnsupported { + stream_serial: u64, + reason: &'static str, + }, + AecFailed { + module_index: u64, + }, + AecRevoked { + module_index: u64, + }, + ForeignAecWarning { + link_group: String, + }, +} + +impl AudioExclusionEvent { + fn as_event(&self) -> Event<'_> { + match self { + Self::StreamUnsupported { + stream_serial, + reason, + } => Event::StreamUnsupported { + version: AUDIO_EXCLUSION_EVENT_VERSION, + stream_serial: *stream_serial, + reason, + }, + Self::AecFailed { module_index } => Event::AecFailed { + version: AUDIO_EXCLUSION_EVENT_VERSION, + module_index: *module_index, + }, + Self::AecRevoked { module_index } => Event::AecRevoked { + version: AUDIO_EXCLUSION_EVENT_VERSION, + module_index: *module_index, + }, + Self::ForeignAecWarning { link_group } => Event::ForeignAecWarning { + version: AUDIO_EXCLUSION_EVENT_VERSION, + link_group, + }, + } + } + + /// Emit this owned status record through the stable stdout protocol. + pub(crate) fn emit(&self) { + emit(self.as_event()); + } +} + /// Emit one event as a JSON line on stdout, flushed. No-op unless JSON /// output was enabled with [`set_json`], so call sites can sprinkle these /// unconditionally without branching. @@ -118,4 +193,56 @@ mod tests { .unwrap(); assert_eq!(lost, r#"{"event":"app_audio","state":"lost"}"#); } + + #[test] + fn audio_exclusion_event_wire_shapes_are_versioned_and_exact() { + let unsupported = serde_json::to_string( + &AudioExclusionEvent::StreamUnsupported { + stream_serial: 4_294_967_297, + reason: "link-creation-failed", + } + .as_event(), + ) + .unwrap(); + assert_eq!( + unsupported, + r#"{"event":"stream_unsupported","version":1,"stream_serial":4294967297,"reason":"link-creation-failed"}"# + ); + + let failed = serde_json::to_string( + &AudioExclusionEvent::AecFailed { + module_index: 536_870_919, + } + .as_event(), + ) + .unwrap(); + assert_eq!( + failed, + r#"{"event":"aec_failed","version":1,"module_index":536870919}"# + ); + + let revoked = serde_json::to_string( + &AudioExclusionEvent::AecRevoked { + module_index: 536_870_919, + } + .as_event(), + ) + .unwrap(); + assert_eq!( + revoked, + r#"{"event":"aec_revoked","version":1,"module_index":536870919}"# + ); + + let warning = serde_json::to_string( + &AudioExclusionEvent::ForeignAecWarning { + link_group: "echo-cancel-9999-13".to_string(), + } + .as_event(), + ) + .unwrap(); + assert_eq!( + warning, + r#"{"event":"foreign_aec_warning","version":1,"link_group":"echo-cancel-9999-13"}"# + ); + } } diff --git a/src/host/aec/mod.rs b/src/host/aec/mod.rs index a4b2218..b91980d 100644 --- a/src/host/aec/mod.rs +++ b/src/host/aec/mod.rs @@ -195,6 +195,14 @@ impl AecValidator { self.state } + /// The configured module index regardless of validation state. Status + /// reporting and foreign-AEC detection need to name the intended identity + /// without treating it as trusted for taint; only + /// [`Self::validated_module_id`] grants that trust. + pub fn configured_module_id(&self) -> Option { + self.target + } + /// The validated index to place in /// [`ExclusionCtx::aec_module_id`](crate::host::taint::ExclusionCtx) — /// `Some` **only** in [`AecState::Validated`]. `None` everywhere else, diff --git a/src/host/fanout.rs b/src/host/fanout.rs index d91211e..474f58f 100644 --- a/src/host/fanout.rs +++ b/src/host/fanout.rs @@ -18,7 +18,11 @@ use super::observer::{EventKind, Millis, Projection}; use super::taint::snapshot::{ GlobalId, GraphSnapshot, MediaRole, PortDirection, PortSnapshot, Serial, }; -use super::taint::{Decisions, ExclusionCtx, Reason, StickyState, evaluate}; +use super::taint::{ + Decisions, ECHO_CANCEL_GROUP_PREFIX, ExclusionCtx, Reason, StickyState, evaluate, +}; +use crate::common::output::AudioExclusionEvent; +use tokio::sync::mpsc; const AEC_VALIDATION_TIMEOUT_MILLIS: Millis = 5_000; @@ -148,15 +152,48 @@ pub(super) struct FanoutController { validator: AecValidator, sticky: StickyState, states: BTreeMap, + foreign_aec_groups: BTreeSet, + status_tx: Option>, } impl FanoutController { + #[cfg(test)] pub(super) fn new(capture_sink: Serial, aec: AecConfig) -> Self { + Self::build(capture_sink, aec, None) + } + + /// 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. + pub(super) fn with_status_sender( + capture_sink: Serial, + aec: AecConfig, + status_tx: mpsc::UnboundedSender, + ) -> Self { + Self::build(capture_sink, aec, Some(status_tx)) + } + + fn build( + capture_sink: Serial, + aec: AecConfig, + status_tx: Option>, + ) -> Self { Self { capture_sink, validator: AecValidator::new(aec, AEC_VALIDATION_TIMEOUT_MILLIS), sticky: StickyState::default(), states: BTreeMap::new(), + foreign_aec_groups: BTreeSet::new(), + status_tx, + } + } + + fn emit_status(&self, event: AudioExclusionEvent) { + if let Some(status_tx) = &self.status_tx { + // A closed receiver means the machine front-end no longer accepts + // notices. Capture ownership remains safe and should not be + // poisoned merely because its best-effort stdout consumer ended. + let _ = status_tx.send(event); } } @@ -174,9 +211,37 @@ impl MutationProjectionSink for FanoutController { now_us: u64, links: &mut dyn LinkMutation, ) { + let prior_aec_state = self.validator.state(); self.validator .observe(&projection.snapshot, projection.graph_ready, now_us / 1_000); let aec_state = self.validator.state(); + let aec_status = if prior_aec_state != aec_state { + self.validator + .configured_module_id() + .and_then(|module_index| match aec_state { + AecState::Failed => Some(AudioExclusionEvent::AecFailed { module_index }), + AecState::Revoked => Some(AudioExclusionEvent::AecRevoked { module_index }), + _ => None, + }) + } else { + None + }; + let configured_aec = self.validator.configured_module_id(); + let foreign_aec_groups: BTreeSet = projection + .snapshot + .nodes() + .filter_map(|node| { + let group = node.props.link_group.as_ref()?; + (group.starts_with(ECHO_CANCEL_GROUP_PREFIX) + && configured_aec + .is_none_or(|module| node.props.pulse_module_id != Some(module))) + .then(|| group.clone()) + }) + .collect(); + let new_foreign_groups: Vec = foreign_aec_groups + .difference(&self.foreign_aec_groups) + .cloned() + .collect(); let ctx = ExclusionCtx { aec_module_id: self.validator.validated_module_id(), pipewire_pulse_pid: projection.pipewire_pulse_pid, @@ -234,16 +299,23 @@ impl MutationProjectionSink for FanoutController { (serial, state) }) .collect(); + let mut stream_statuses = Vec::new(); for (&serial, state) in &next_states { if self.states.get(&serial) == Some(state) { continue; } match state { - StreamCaptureState::Unsupported { issue } => tracing::warn!( - serial = serial.0, - reason = issue.code(), - "audio fan-out: stream unsupported" - ), + StreamCaptureState::Unsupported { issue } => { + tracing::warn!( + serial = serial.0, + reason = issue.code(), + "audio fan-out: stream unsupported" + ); + stream_statuses.push(AudioExclusionEvent::StreamUnsupported { + stream_serial: serial.0, + reason: issue.code(), + }); + } StreamCaptureState::Captured => tracing::info!( serial = serial.0, "audio fan-out: every required link is active" @@ -252,6 +324,20 @@ impl MutationProjectionSink for FanoutController { } } self.states = next_states; + self.foreign_aec_groups = foreign_aec_groups; + + // `reconcile` above has already dropped every link forbidden by an AEC + // transition or a newly unsafe graph. Only report after that mutation + // boundary, so `aec_revoked` can never race ahead of link revocation. + if let Some(event) = aec_status { + self.emit_status(event); + } + for link_group in new_foreign_groups { + self.emit_status(AudioExclusionEvent::ForeignAecWarning { link_group }); + } + for event in stream_statuses { + self.emit_status(event); + } } } @@ -496,6 +582,20 @@ mod tests { controller.on_projection(&projection(graph), EventKind::Graph, 1_000, links); } + fn controller_with_status( + capture_sink: Serial, + aec: AecConfig, + ) -> ( + FanoutController, + mpsc::UnboundedReceiver, + ) { + let (status_tx, status_rx) = mpsc::unbounded_channel(); + ( + FanoutController::with_status_sender(capture_sink, aec, status_tx), + status_rx, + ) + } + #[test] fn stereo_ports_pair_by_channel_not_enumeration_order() { let (graph, app, sink, [app_fl, app_fr, sink_fl, sink_fr]) = stereo_graph(); @@ -745,6 +845,50 @@ mod tests { )); } + #[test] + fn aec_validation_deadline_emits_the_failed_cause_once() { + let (graph, app, sink, _) = stereo_graph(); + let mut links = FakeLinks::default(); + let (mut controller, mut status_rx) = + controller_with_status(sink, AecConfig::PulseModule(999)); + drive(&mut controller, &graph, &mut links); + assert!(matches!( + controller.states()[&app], + StreamCaptureState::Gated { + aec: AecState::Validating + } + )); + + controller.on_projection( + &projection(&graph), + EventKind::Tick, + (AEC_VALIDATION_TIMEOUT_MILLIS + 2) * 1_000, + &mut links, + ); + assert_eq!( + status_rx.try_recv(), + Ok(AudioExclusionEvent::AecFailed { module_index: 999 }) + ); + assert!(matches!( + controller.states()[&app], + StreamCaptureState::Gated { + aec: AecState::Failed + } + )); + + controller.on_projection( + &projection(&graph), + EventKind::Tick, + (AEC_VALIDATION_TIMEOUT_MILLIS + 3) * 1_000, + &mut links, + ); + assert_eq!( + status_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty), + "a sticky failed state must not repeat the event on every tick" + ); + } + #[test] fn link_creation_failure_is_sticky_for_the_same_intent() { let (graph, app, sink, _) = stereo_graph(); @@ -770,7 +914,7 @@ mod tests { *planned.iter().next().expect("stereo link"), ManagedLinkState::Failed, ); - let mut controller = FanoutController::new(sink, AecConfig::Off); + let (mut controller, mut status_rx) = controller_with_status(sink, AecConfig::Off); drive(&mut controller, &graph, &mut links); assert_eq!( controller.states()[&app], @@ -779,11 +923,23 @@ mod tests { } ); assert_eq!(links.creates, 2); + assert_eq!( + status_rx.try_recv(), + Ok(AudioExclusionEvent::StreamUnsupported { + stream_serial: app.0, + reason: "link-creation-failed", + }) + ); drive(&mut controller, &graph, &mut links); assert_eq!( links.creates, 2, "an identical failed intent must not hot-loop" ); + assert_eq!( + status_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty), + "an unchanged unsupported stream must not repeat its event" + ); } #[test] @@ -792,7 +948,8 @@ mod tests { let aec = graph.module_node("aec-playback", MediaRole::StreamOutput, 999); graph.port_on_channel(aec, PortDirection::Out, false, Some("FL")); let mut links = FakeLinks::default(); - let mut controller = FanoutController::new(sink, AecConfig::PulseModule(999)); + let (mut controller, mut status_rx) = + controller_with_status(sink, AecConfig::PulseModule(999)); drive(&mut controller, &graph, &mut links); assert_eq!(controller.states()[&app], StreamCaptureState::Captured); assert_eq!(links.held.len(), 2); @@ -811,5 +968,74 @@ mod tests { aec: AecState::Revoked } ); + assert_eq!( + status_rx.try_recv(), + Ok(AudioExclusionEvent::AecRevoked { module_index: 999 }) + ); + } + + #[test] + fn second_echo_cancel_group_emits_one_foreign_warning() { + use crate::host::taint::fixture::pulse_module; + + let (mut graph, _app, sink, _) = stereo_graph(); + let own_client = graph.client_of_module(999); + let mut own_props = pulse_module(own_client, 999, PULSE_PID); + own_props.link_group = Some("echo-cancel-owned-13".to_string()); + graph.node("owned-aec-playback", MediaRole::StreamOutput, own_props); + + let foreign_client = graph.client_of_module(1_000); + let mut foreign_props = pulse_module(foreign_client, 1_000, PULSE_PID); + foreign_props.link_group = Some("echo-cancel-foreign-14".to_string()); + graph.node( + "foreign-aec-playback", + MediaRole::StreamOutput, + foreign_props.clone(), + ); + graph.node("foreign-aec-capture", MediaRole::StreamInput, foreign_props); + + let mut links = FakeLinks::default(); + let (mut controller, mut status_rx) = + controller_with_status(sink, AecConfig::PulseModule(999)); + drive(&mut controller, &graph, &mut links); + assert_eq!( + status_rx.try_recv(), + Ok(AudioExclusionEvent::ForeignAecWarning { + link_group: "echo-cancel-foreign-14".to_string(), + }) + ); + assert_eq!( + status_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty), + "all legs in one foreign group collapse to one warning" + ); + + drive(&mut controller, &graph, &mut links); + assert_eq!( + status_rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty), + "a still-live foreign group must not warn on every graph event" + ); + } + + #[test] + fn unindexed_native_echo_cancel_group_is_still_foreign() { + let (mut graph, _app, sink, _) = stereo_graph(); + graph.group_node( + "native-aec-playback", + MediaRole::StreamOutput, + "echo-cancel-native-1", + 77, + ); + let mut links = FakeLinks::default(); + let (mut controller, mut status_rx) = controller_with_status(sink, AecConfig::Off); + + drive(&mut controller, &graph, &mut links); + assert_eq!( + status_rx.try_recv(), + Ok(AudioExclusionEvent::ForeignAecWarning { + link_group: "echo-cancel-native-1".to_string(), + }) + ); } } diff --git a/src/host/graph.rs b/src/host/graph.rs index 765ae25..fe3fcfe 100644 --- a/src/host/graph.rs +++ b/src/host/graph.rs @@ -15,6 +15,7 @@ use super::observer::Readiness; use super::observer::adapter::RegistryObserverHandle; use super::owned_thread::OwnedThread; use super::taint::snapshot::{GlobalId, Serial}; +use crate::common::output::AudioExclusionEvent; use crate::repair::plan as repair_plan; use anyhow::{Context, Result, bail}; use pipewire::proxy::ProxyT; @@ -143,8 +144,10 @@ pub(super) struct AudioGraphOwner { /// `Routing` remains a separate type in `host::audio`. pub(super) struct BareCaptureSink { // Drop order is load-bearing: close the fan-out observer connection (and - // therefore every non-lingering link) before the sink-owning connection. + // therefore every non-lingering link), then detach the status task, before + // the sink-owning connection. fanout: Option, + status_forwarder: Option>, graph_owner: Option, monitor_name: String, } @@ -156,21 +159,34 @@ impl BareCaptureSink { .await .context("failed to start the bare connection-owned capture sink")?; let identity = graph_owner.identity().clone(); - let fanout = RegistryObserverHandle::spawn_with_mutation_sink( - Box::new(FanoutController::new( + 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, )), health, ) - .context("failed to start the desktop-excluding fan-out observer")?; + .context("failed to start the desktop-excluding fan-out observer") + { + Ok(fanout) => fanout, + Err(error) => { + let _ = status_forwarder.await; + graph_owner.shutdown().await; + return Err(error); + } + }; if let Err(error) = wait_for_fanout_ready(&fanout, &identity).await { drop(fanout); + let _ = status_forwarder.await; graph_owner.shutdown().await; return Err(error).context("desktop-excluding fan-out did not become ready"); } Ok(Self { fanout: Some(fanout), + status_forwarder: Some(status_forwarder), graph_owner: Some(graph_owner), monitor_name: identity.monitor_name, }) @@ -193,6 +209,11 @@ impl BareCaptureSink { if let Some(fanout) = self.fanout.take() { drop(fanout); } + if let Some(status_forwarder) = self.status_forwarder.take() + && let Err(error) = status_forwarder.await + { + tracing::warn!(%error, "audio fan-out: status forwarder task failed"); + } if let Some(graph_owner) = self.graph_owner.take() && !graph_owner.shutdown().await { @@ -201,6 +222,14 @@ impl BareCaptureSink { } } +async fn forward_audio_exclusion_status( + mut status_rx: mpsc::UnboundedReceiver, +) { + while let Some(event) = status_rx.recv().await { + event.emit(); + } +} + async fn wait_for_fanout_ready( observer: &RegistryObserverHandle, identity: &SinkIdentity,