fix(host): close desktop audio failure gates

src/host/audio_plan.rs

@@ -21,6 +21,39 @@ use super::graph::BareCaptureSink;
 use super::health;
 use crate::cli::{CaptureMode, HostOpts};
 
+trait CapturePlanBackend {
+    async fn resolve_default_monitor(&mut self) -> Result<DefaultMonitor>;
+    async fn start_routing(&mut self, opts: &HostOpts, health: health::Reporter)
+    -> Result<Routing>;
+    async fn start_bare_capture_sink(
+        &mut self,
+        health: health::Reporter,
+    ) -> Result<BareCaptureSink>;
+}
+
+struct SystemCapturePlanBackend;
+
+impl CapturePlanBackend for SystemCapturePlanBackend {
+    async fn resolve_default_monitor(&mut self) -> Result<DefaultMonitor> {
+        DefaultMonitor::resolve().await
+    }
+
+    async fn start_routing(
+        &mut self,
+        opts: &HostOpts,
+        health: health::Reporter,
+    ) -> Result<Routing> {
+        Routing::start(opts, health).await
+    }
+
+    async fn start_bare_capture_sink(
+        &mut self,
+        health: health::Reporter,
+    ) -> Result<BareCaptureSink> {
+        BareCaptureSink::start(health).await
+    }
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub(super) enum CapturePlanKind {
     LegacyDesktop,
@@ -99,17 +132,35 @@ pub(super) enum CapturePlan {
 
 impl CapturePlan {
     pub(super) async fn start(opts: &HostOpts, health: health::Reporter) -> Result<Self> {
+        let mut backend = SystemCapturePlanBackend;
+        Self::start_with_backend(opts, health, &mut backend).await
+    }
+
+    /// One production construction path with an injectable system boundary.
+    ///
+    /// Keeping failure policy here makes rows 8d/8e falsifiable in release
+    /// builds: a capture-sink or readiness failure must escape this function.
+    /// It must never be translated into a second attempt through
+    /// `LegacyDesktop`, whose default-monitor source would reintroduce the
+    /// audio this mode exists to exclude.
+    async fn start_with_backend<B: CapturePlanBackend>(
+        opts: &HostOpts,
+        health: health::Reporter,
+        backend: &mut B,
+    ) -> Result<Self> {
         match CapturePlanKind::resolve(opts)? {
             CapturePlanKind::LegacyDesktop => Ok(Self::LegacyDesktop {
-                source: DefaultMonitor::resolve().await?,
+                source: backend.resolve_default_monitor().await?,
             }),
             CapturePlanKind::PerApp => Ok(Self::PerApp {
-                routing: Routing::start(opts, health)
+                routing: backend
+                    .start_routing(opts, health)
                     .await
                     .context("audio routing setup failed")?,
             }),
             CapturePlanKind::DesktopExcluding => Ok(Self::DesktopExcluding {
-                capture_sink: BareCaptureSink::start(health)
+                capture_sink: backend
+                    .start_bare_capture_sink(health)
                     .await
                     .context("desktop-excluding capture-sink setup failed")?,
             }),
@@ -148,6 +199,8 @@ impl CapturePlan {
 mod tests {
     use super::*;
     use crate::cli::Quality;
+    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::process::{Child, Command, Stdio};
@@ -224,6 +277,99 @@ mod tests {
         assert!(matches!(plan, CapturePlan::LegacyDesktop { .. }));
     }
 
+    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
+    enum BackendCall {
+        DefaultMonitor,
+        Routing,
+        BareCaptureSink,
+    }
+
+    struct FailingDesktopBackend {
+        failure: Option<anyhow::Error>,
+        calls: Vec<BackendCall>,
+    }
+
+    impl CapturePlanBackend for FailingDesktopBackend {
+        async fn resolve_default_monitor(&mut self) -> Result<DefaultMonitor> {
+            self.calls.push(BackendCall::DefaultMonitor);
+            panic!("a DesktopExcluding failure must not resolve the legacy default monitor")
+        }
+
+        async fn start_routing(
+            &mut self,
+            _opts: &HostOpts,
+            _health: health::Reporter,
+        ) -> Result<Routing> {
+            self.calls.push(BackendCall::Routing);
+            panic!("a DesktopExcluding failure must not construct legacy Routing")
+        }
+
+        async fn start_bare_capture_sink(
+            &mut self,
+            _health: health::Reporter,
+        ) -> Result<BareCaptureSink> {
+            self.calls.push(BackendCall::BareCaptureSink);
+            Err(self.failure.take().expect("one injected failure"))
+        }
+    }
+
+    async fn assert_desktop_failure_is_closed(failure: anyhow::Error, expected_cause: &str) {
+        let opts = opts(false, false, CaptureMode::DesktopExcluding, false);
+        assert_eq!(
+            CapturePlanKind::resolve(&opts).expect("fixture mode resolves"),
+            CapturePlanKind::DesktopExcluding
+        );
+        let mut backend = FailingDesktopBackend {
+            failure: Some(failure),
+            calls: Vec::new(),
+        };
+        let (health, _) = health::channel();
+        let error = match CapturePlan::start_with_backend(&opts, health, &mut backend).await {
+            Ok(_) => panic!("DesktopExcluding unexpectedly recovered through another plan"),
+            Err(error) => error,
+        };
+        assert_eq!(backend.calls, vec![BackendCall::BareCaptureSink]);
+        let chain = format!("{error:#}");
+        assert!(chain.contains("desktop-excluding capture-sink setup failed"));
+        assert!(
+            chain.contains(expected_cause),
+            "unexpected error chain: {chain}"
+        );
+    }
+
+    #[tokio::test]
+    async fn capture_sink_creation_failure_never_falls_back_to_legacy_desktop() {
+        assert_desktop_failure_is_closed(
+            anyhow::anyhow!("injected capture-sink creation failure"),
+            "injected capture-sink creation failure",
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    async fn readiness_epoch_timeout_never_falls_back_to_legacy_desktop() {
+        let identity = super::super::graph::SinkIdentity {
+            name: "pixelpass_capture_timeout_fixture".to_string(),
+            monitor_name: "pixelpass_capture_timeout_fixture.monitor".to_string(),
+            global_id: 77,
+            serial: 88,
+        };
+        let projection = Projection {
+            snapshot: Graph::new().build(),
+            pipewire_pulse_pid: None,
+            graph_ready: false,
+            readiness: Readiness::TimedOut,
+        };
+        let readiness_error = super::super::graph::fanout_readiness(Some(&projection), &identity)
+            .expect_err("the observer's sticky timeout must fail readiness");
+
+        assert_desktop_failure_is_closed(
+            readiness_error,
+            "registry observer reached its sticky readiness timeout",
+        )
+        .await;
+    }
+
     fn pulse_source_exists(name: &str) -> bool {
         std::process::Command::new("pactl")
             .args(["get-source-volume", name])

src/host/fanout.rs

@@ -51,6 +51,20 @@ impl DesiredLink {
     }
 }
 
+/// Revalidate every desired endpoint against the projection at the mutation
+/// edge. A plan made from an older graph is not authority to address a global
+/// id after that id has been recycled.
+pub(super) fn revalidated_links(
+    snapshot: &GraphSnapshot,
+    desired: &BTreeSet<DesiredLink>,
+) -> BTreeSet<DesiredLink> {
+    desired
+        .iter()
+        .copied()
+        .filter(|link| link.is_current(snapshot))
+        .collect()
+}
+
 fn endpoint_is_current(snapshot: &GraphSnapshot, endpoint: DesiredEndpoint) -> bool {
     snapshot
         .node(endpoint.node_serial)
@@ -587,11 +601,7 @@ mod tests {
             snapshot: &GraphSnapshot,
             desired: &BTreeSet<DesiredLink>,
         ) -> BTreeMap<DesiredLink, ManagedLinkState> {
-            let current: BTreeSet<DesiredLink> = desired
+            let current = revalidated_links(snapshot, desired);
-                .iter()
-                .copied()
-                .filter(|link| link.is_current(snapshot))
-                .collect();
             self.drops += self.held.difference(&current).count();
             self.creates += current.difference(&self.held).count();
             self.held = current.clone();
@@ -801,6 +811,45 @@ mod tests {
         );
     }
 
+    #[test]
+    fn identity_change_between_planning_and_create_makes_zero_create_calls() {
+        let (mut graph, app, sink, _) = stereo_graph();
+        let (planned_snapshot, decisions) = decisions(&graph);
+        let StreamPlan::Capture { links: planned } =
+            &plan(&planned_snapshot, &decisions, sink)[&app]
+        else {
+            panic!("fixture must produce a clean link plan");
+        };
+
+        // Recycle the output node's global id after evaluation but before the
+        // mutation edge sees the request. The old serial must authorize
+        // neither channel even though both numeric ids still exist.
+        let old_app_id = planned_snapshot.node(app).expect("old app").id;
+        let reborn = graph.node_with_id(
+            "reborn-between-plan-and-create",
+            MediaRole::StreamOutput,
+            old_app_id,
+            NodeProps::default(),
+        );
+        graph.port_on_channel(reborn, PortDirection::Out, false, Some("FL"));
+        graph.port_on_channel(reborn, PortDirection::Out, false, Some("FR"));
+        let mutation_snapshot = graph.build_without(&[crate::host::taint::fixture::NodeRef {
+            serial: app,
+            id: old_app_id,
+        }]);
+
+        let mut links = FakeLinks::default();
+        let states = links.reconcile(&mutation_snapshot, planned);
+        assert_eq!(links.creates, 0, "no stale intent may reach create_link");
+        assert!(links.held.is_empty());
+        assert!(
+            states
+                .values()
+                .all(|state| *state == ManagedLinkState::Failed),
+            "every stale channel must fail closed"
+        );
+    }
+
     #[test]
     fn duplicate_projection_is_idempotent_and_all_links_must_activate() {
         let (graph, app, sink, _) = stereo_graph();

src/host/graph.rs

@@ -11,8 +11,8 @@ use super::aec::AecConfig;
 use super::audio::parse_object_serial;
 use super::fanout::FanoutController;
 use super::health;
-use super::observer::Readiness;
 use super::observer::adapter::{FanoutControl, RegistryObserverHandle};
+use super::observer::{Projection, Readiness};
 use super::owned_thread::OwnedThread;
 use super::taint::snapshot::{GlobalId, Serial};
 use crate::common::output::AudioExclusionEvent;
@@ -293,17 +293,9 @@ async fn wait_for_fanout_ready(
 ) -> Result<()> {
     let deadline = Instant::now() + FANOUT_READY_BUDGET;
     loop {
-        if let Some(projection) = observer.latest() {
+        let projection = observer.latest();
-            if projection.readiness == Readiness::TimedOut {
+        if fanout_readiness(projection.as_ref(), identity)? {
-                bail!("registry observer reached its sticky readiness timeout");
+            return Ok(());
-            }
-            let sink_is_current = projection
-                .snapshot
-                .node(Serial(identity.serial))
-                .is_some_and(|node| node.id == GlobalId(identity.global_id));
-            if projection.graph_ready && sink_is_current {
-                return Ok(());
-            }
         }
         if Instant::now() >= deadline {
             bail!(
@@ -315,6 +307,23 @@ async fn wait_for_fanout_ready(
     }
 }
 
+pub(super) fn fanout_readiness(
+    projection: Option<&Projection>,
+    identity: &SinkIdentity,
+) -> Result<bool> {
+    let Some(projection) = projection else {
+        return Ok(false);
+    };
+    if projection.readiness == Readiness::TimedOut {
+        bail!("registry observer reached its sticky readiness timeout");
+    }
+    let sink_is_current = projection
+        .snapshot
+        .node(Serial(identity.serial))
+        .is_some_and(|node| node.id == GlobalId(identity.global_id));
+    Ok(projection.graph_ready && sink_is_current)
+}
+
 impl AudioGraphOwner {
     pub(super) async fn start(
         filter_name: Option<String>,

src/host/observer/adapter.rs

@@ -11,7 +11,9 @@ use super::{
     EventKind, LinkEndpoints, NodeObservation, Outcome, Projection, RegEvent, RegistryModel,
 };
 use crate::host::audio::parse_object_serial;
-use crate::host::fanout::{DesiredLink, LinkMutation, ManagedLinkState, MutationProjectionSink};
+use crate::host::fanout::{
+    DesiredLink, LinkMutation, ManagedLinkState, MutationProjectionSink, revalidated_links,
+};
 use crate::host::taint::snapshot::{
     ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial,
 };
@@ -303,12 +305,14 @@ impl LinkMutation for PipeWireLinkMutation {
         snapshot: &crate::host::taint::snapshot::GraphSnapshot,
         desired: &BTreeSet<DesiredLink>,
     ) -> BTreeMap<DesiredLink, ManagedLinkState> {
-        // Revoke before creating. An unsafe link is not allowed to coexist
+        // Revoke before creating. Revalidation applies to retained proxies as
-        // briefly with its replacement set.
+        // well as new requests: a stale serial-guarded intent is no longer
-        self.owned.retain(|link, _| desired.contains(link));
+        // authority merely because it already reached `owned`.
+        let current = revalidated_links(snapshot, desired);
+        self.owned.retain(|link, _| current.contains(link));
         self.states
             .borrow_mut()
-            .retain(|link, _| desired.contains(link));
+            .retain(|link, _| current.contains(link));
 
         // A Link that reached Error stays failed until the graph changes
         // enough to remove this exact serial-guarded intent. Retrying the same
@@ -323,18 +327,10 @@ impl LinkMutation for PipeWireLinkMutation {
             self.owned.remove(&link);
         }
 
-        for &link in desired {
+        for &link in &current {
             if self.owned.contains_key(&link) || self.states.borrow().contains_key(&link) {
                 continue;
             }
-            // Same-callback revalidation. Global ids are never sufficient:
-            // every node and port must still carry the serial the planner saw.
-            if !link.is_current(snapshot) {
-                self.states
-                    .borrow_mut()
-                    .insert(link, ManagedLinkState::Failed);
-                continue;
-            }
             match self.create_link(link) {
                 Ok(owned) => {
                     self.owned.insert(link, owned);