fix(host): contain capture owners and fail closed

Bound the libpipewire router shutdown without detaching its OS handle, contain GStreamer and pactl children in parent-bound process groups, and make ownership failures terminal through the capture supervisor.\n\nAdd focused lifecycle tests plus a serialized live router teardown gate.

src/host/audio.rs

@@ -37,11 +37,14 @@ use std::io::{self, Read};
 use std::process::{Child, Command, ExitStatus, Stdio};
 use std::rc::Rc;
 use std::sync::Arc;
-use std::thread::JoinHandle;
+use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use std::time::{Duration, Instant};
 
 use crate::cli::HostOpts;
+use crate::common::contained;
+use crate::host::health;
 use crate::host::ledger::{self, LedgerError, ModuleLedger, UnloadOutcome};
+use crate::host::owned_thread::OwnedThread;
 use crate::repair::plan::{self as repair_plan, Fingerprint, Shape};
 
 /// How long a `pactl load-module` worker may run before it is killed and reaped.
@@ -54,6 +57,8 @@ use crate::repair::plan::{self as repair_plan, Fingerprint, Shape};
 /// connect/list/unload requests instead of a second pactl connection.
 const PACTL_BUDGET: Duration = Duration::from_secs(5);
 const PACTL_REAP_BUDGET: Duration = Duration::from_secs(1);
+const ROUTER_RUNNING_STOP_BUDGET: Duration = Duration::from_secs(2);
+const ROUTER_STARTING_STOP_BUDGET: Duration = Duration::from_secs(5);
 
 /// Owns the pactl-loaded modules plus, when filtering is active, the
 /// libpipewire stream-router thread. Drop unloads modules as a backstop;
@@ -71,12 +76,13 @@ pub struct Routing {
     sink_name: String,
     stream_router: Option<StreamRouter>,
     event_task: Option<tokio::task::JoinHandle<()>>,
+    health: health::Reporter,
 }
 
 impl Routing {
     /// Create the per-PID null-sink + loopback. If `opts.app` is set,
     /// also spawn the libpipewire thread that reroutes matching streams.
-    pub async fn start(opts: &HostOpts) -> Result<Self> {
+    pub(super) async fn start(opts: &HostOpts, health: health::Reporter) -> Result<Self> {
         let pid = std::process::id();
         let sink_name = repair_plan::sink_name_for(pid);
         let ledger = ModuleLedger::new();
@@ -90,6 +96,7 @@ impl Routing {
             sink_name: sink_name.clone(),
             stream_router: None,
             event_task: None,
+            health: health.clone(),
         };
 
         // Every module this host loads carries an ownership token, minted per
@@ -125,7 +132,8 @@ impl Routing {
         );
 
         if let Some(app) = &opts.app {
-            let (router, mut event_rx) = StreamRouter::spawn(app.clone(), sink_name.clone())?;
+            let (router, mut event_rx) =
+                StreamRouter::spawn(app.clone(), sink_name.clone(), health.clone())?;
             let ledger_for_task = Arc::clone(&ledger);
             let strict = opts.strict_audio;
             let event_task = tokio::spawn(async move {
@@ -235,27 +243,44 @@ impl Routing {
         // task cannot register another mutation even if it receives one last
         // router event while shutdown is in progress.
         self.ledger.close();
-        if let Some(router) = self.stream_router.take() {
-            // ⚠️ Still an unbounded join: a wedged PipeWire thread parks this
-            // task indefinitely. That is the pre-existing defect S3b exists for.
-            // Nothing here makes it worse, and the ledger is what will make
-            // bounding it safe when it lands.
-            router.shutdown();
-        }
+        let router_stopped = if let Some(router) = self.stream_router.take() {
+            router.shutdown().await
+        } else {
+            true
+        };
         if let Some(mut task) = self.event_task.take() {
-            // The router's exit drops the event senders, so the task normally
-            // ends by itself. Abort is the fallback, and it is awaited through
-            // `&mut JoinHandle` so the future is genuinely dropped — and with it
-            // any in-flight permit — before reconciliation reads the ledger.
-            // Dropping the handle instead would *detach* the task, which is how a
-            // load could still land after teardown believed it was finished.
-            if tokio::time::timeout(PACTL_BUDGET, &mut task).await.is_err() {
-                tracing::warn!(
-                    "audio routing: the event task did not finish within {PACTL_BUDGET:?}; \
-                     cancelling it"
-                );
+            if !router_stopped {
+                // A quarantined router still owns its event sender, so this task
+                // cannot finish naturally. Cancel and await it before ledger
+                // reconciliation; the router timeout already poisoned the host.
                 task.abort();
                 let _ = task.await;
+            } else {
+                // The router's exit drops the event senders, so the task normally
+                // ends by itself. Abort is the fallback, and it is awaited through
+                // `&mut JoinHandle` so the future is genuinely dropped — and with
+                // it any in-flight permit — before reconciliation reads the
+                // ledger. Dropping the handle instead would *detach* the task,
+                // which is how a load could still land after teardown believed it
+                // was finished.
+                match tokio::time::timeout(PACTL_BUDGET, &mut task).await {
+                    Ok(Ok(())) => {}
+                    Ok(Err(e)) => {
+                        self.health
+                            .poison(format!("audio routing event task failed: {e}"));
+                    }
+                    Err(_) => {
+                        tracing::warn!(
+                            "audio routing: the event task did not finish within {PACTL_BUDGET:?}; \
+                             cancelling it"
+                        );
+                        self.health.poison(format!(
+                            "audio routing event task did not stop within {PACTL_BUDGET:?}"
+                        ));
+                        task.abort();
+                        let _ = task.await;
+                    }
+                }
             }
         }
 
@@ -269,6 +294,8 @@ impl Routing {
         .await
         {
             tracing::warn!("audio routing: module-operation wait task failed: {e}");
+            self.health
+                .poison(format!("audio module-operation wait task failed: {e}"));
         }
         cleanup_modules(&self.ledger).await;
 
@@ -278,6 +305,9 @@ impl Routing {
                 "audio routing: some audio modules could not be removed safely; \
                  `pixelpass --repair` will clean up anything left behind"
             );
+            self.health.poison(
+                "audio routing teardown left module ownership unresolved; repair is required",
+            );
         }
     }
 }
@@ -292,7 +322,7 @@ impl Drop for Routing {
     fn drop(&mut self) {
         self.ledger.close();
         if let Some(router) = self.stream_router.take() {
-            router.shutdown();
+            drop(router);
         }
         if let Some(task) = self.event_task.take() {
             task.abort();
@@ -305,6 +335,8 @@ impl Drop for Routing {
                 "audio routing: torn down with modules that could not be removed safely; \
                  run `pixelpass --repair` to clean up anything left behind"
             );
+            self.health
+                .poison("audio routing Drop left module ownership unresolved; repair is required");
         }
     }
 }
@@ -635,7 +667,7 @@ impl Drop for ReapedChild {
         if self.reaped {
             return;
         }
-        let _ = self.child.kill();
+        self.kill_group();
         match self.reap_within(PACTL_REAP_BUDGET) {
             Ok(Some(_)) => {}
             Ok(None) => tracing::warn!(
@@ -647,6 +679,12 @@ impl Drop for ReapedChild {
 }
 
 impl ReapedChild {
+    fn kill_group(&mut self) {
+        let _ = contained::signal_group(self.child.id(), nix::sys::signal::Signal::SIGKILL);
+        // Backstop in case the group disappeared between lookup and signal.
+        let _ = self.child.kill();
+    }
+
     fn reap_within(&mut self, budget: Duration) -> io::Result<Option<ExitStatus>> {
         let deadline = Instant::now() + budget;
         loop {
@@ -665,7 +703,7 @@ impl ReapedChild {
 fn bounded_output(command: &mut Command, budget: Duration) -> io::Result<BoundedOutput> {
     command.stdout(Stdio::piped()).stderr(Stdio::piped());
     let mut child = ReapedChild {
-        child: command.spawn()?,
+        child: contained::spawn(command)?,
         reaped: false,
     };
     let stdout = child
@@ -700,7 +738,7 @@ fn bounded_output(command: &mut Command, budget: Duration) -> io::Result<Bounded
             break (status, false);
         }
         if Instant::now() >= deadline {
-            let _ = child.child.kill();
+            child.kill_group();
             let Some(status) = child.reap_within(PACTL_REAP_BUDGET)? else {
                 return Err(io::Error::new(
                     io::ErrorKind::TimedOut,
@@ -807,24 +845,53 @@ enum Event {
 /// Handle to the libpipewire stream-router thread.
 pub struct StreamRouter {
     cmd_tx: pipewire::channel::Sender<Cmd>,
-    thread: Option<JoinHandle<()>>,
+    thread: OwnedThread,
+    phase: Arc<AtomicU8>,
 }
 
+const ROUTER_STARTING: u8 = 0;
+const ROUTER_RUNNING: u8 = 1;
+const ROUTER_EXITED: u8 = 2;
+
 impl StreamRouter {
     /// Spawn the libpipewire thread. Returns the router handle and the
     /// event receiver tokio side polls.
     fn spawn(
         filter_name: String,
         sink_name: String,
+        health: health::Reporter,
     ) -> Result<(Self, tokio::sync::mpsc::UnboundedReceiver<Event>)> {
         let (cmd_tx, cmd_rx) = pipewire::channel::channel::<Cmd>();
         let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel::<Event>();
+        let phase = Arc::new(AtomicU8::new(ROUTER_STARTING));
+        let phase_for_thread = Arc::clone(&phase);
+        let shutdown_observed = Arc::new(AtomicBool::new(false));
+        let shutdown_for_thread = Arc::clone(&shutdown_observed);
+        let health_for_thread = health.clone();
 
         let thread = std::thread::Builder::new()
             .name("pixelpass-pw-router".to_string())
             .spawn(move || {
-                if let Err(e) = run_router(filter_name, sink_name, cmd_rx, event_tx) {
-                    tracing::warn!("audio routing: libpipewire thread exited with error: {e:#}");
+                let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                    run_router(
+                        filter_name,
+                        sink_name,
+                        cmd_rx,
+                        event_tx,
+                        Arc::clone(&phase_for_thread),
+                        Arc::clone(&shutdown_for_thread),
+                    )
+                }));
+                phase_for_thread.store(ROUTER_EXITED, Ordering::Release);
+                match result {
+                    Ok(result) => report_router_exit(
+                        &health_for_thread,
+                        shutdown_for_thread.load(Ordering::Acquire),
+                        result,
+                    ),
+                    Err(_) => {
+                        health_for_thread.poison("libpipewire router thread panicked");
+                    }
                 }
             })
             .context("failed to spawn libpipewire router thread")?;
@@ -832,19 +899,46 @@ impl StreamRouter {
         Ok((
             Self {
                 cmd_tx,
-                thread: Some(thread),
+                thread: OwnedThread::new("libpipewire router thread", thread, health),
+                phase,
             },
             event_rx,
         ))
     }
 
-    fn shutdown(mut self) {
+    async fn shutdown(mut self) -> bool {
         // Best-effort: if the send fails the thread is already gone.
         let _ = self.cmd_tx.send(Cmd::Shutdown);
-        if let Some(t) = self.thread.take()
-            && let Err(e) = t.join()
-        {
-            tracing::warn!("audio routing: pw thread join failed: {e:?}");
+        let budget = router_shutdown_budget(self.phase.load(Ordering::Acquire));
+        self.thread.join_within(budget).await
+    }
+}
+
+impl Drop for StreamRouter {
+    fn drop(&mut self) {
+        // If async shutdown is cancelled, wake the MainLoop before OwnedThread's
+        // Drop poisons/quarantines the still-owned handle.
+        let _ = self.cmd_tx.send(Cmd::Shutdown);
+    }
+}
+
+fn router_shutdown_budget(phase: u8) -> Duration {
+    if phase == ROUTER_STARTING {
+        ROUTER_STARTING_STOP_BUDGET
+    } else {
+        ROUTER_RUNNING_STOP_BUDGET
+    }
+}
+
+fn report_router_exit(health: &health::Reporter, shutdown_observed: bool, result: Result<()>) {
+    match result {
+        Ok(()) if shutdown_observed => {}
+        Ok(()) => {
+            health.poison("libpipewire router thread exited without a shutdown command");
+        }
+        Err(e) => {
+            tracing::warn!("audio routing: libpipewire thread exited with error: {e:#}");
+            health.poison(format!("libpipewire router thread failed: {e:#}"));
         }
     }
 }
@@ -856,6 +950,8 @@ fn run_router(
     sink_name: String,
     cmd_rx: pipewire::channel::Receiver<Cmd>,
     event_tx: tokio::sync::mpsc::UnboundedSender<Event>,
+    phase: Arc<AtomicU8>,
+    shutdown_observed: Arc<AtomicBool>,
 ) -> Result<()> {
     use pipewire::{self as pw, types::ObjectType};
 
@@ -878,8 +974,10 @@ fn run_router(
     // Cmd handler: clear metadata for routed streams, then quit.
     let main_loop_for_cmd = main_loop.clone();
     let state_for_cmd = Rc::clone(&state);
+    let shutdown_for_cmd = Arc::clone(&shutdown_observed);
     let _cmd_recv = cmd_rx.attach(main_loop.loop_(), move |cmd| match cmd {
         Cmd::Shutdown => {
+            shutdown_for_cmd.store(true, Ordering::Release);
             let s = state_for_cmd.borrow();
             if let Some(meta) = &s.default_metadata {
                 for &nid in &s.routed_node_ids {
@@ -978,6 +1076,7 @@ fn run_router(
         .register();
 
     tracing::info!(filter = %filter_name, "audio routing: pw thread running");
+    phase.store(ROUTER_RUNNING, Ordering::Release);
     main_loop.run();
     tracing::info!("audio routing: pw thread exiting");
     Ok(())
@@ -1074,6 +1173,7 @@ mod tests {
     use super::*;
     use crate::host::ledger::SlotState;
     use crate::repair::plan::{ModuleObservation, classify};
+    use std::sync::mpsc;
 
     /// Whole-desktop routing: no app filter, so no PipeWire thread and no event
     /// task — just the null-sink and its default-sink loopback.
@@ -1094,6 +1194,86 @@ mod tests {
         }
     }
 
+    #[test]
+    fn router_exit_is_healthy_only_after_the_thread_observed_shutdown() {
+        let (clean, _) = health::channel();
+        report_router_exit(&clean, true, Ok(()));
+        assert!(clean.fault().is_none());
+
+        let (unexpected, _) = health::channel();
+        report_router_exit(&unexpected, false, Ok(()));
+        assert_eq!(
+            unexpected.fault().as_deref(),
+            Some("libpipewire router thread exited without a shutdown command")
+        );
+
+        let (failed, _) = health::channel();
+        report_router_exit(&failed, false, Err(anyhow::anyhow!("fixture failure")));
+        assert!(
+            failed
+                .fault()
+                .as_deref()
+                .is_some_and(|reason| reason.contains("fixture failure"))
+        );
+    }
+
+    #[test]
+    fn router_shutdown_has_distinct_startup_and_running_budgets() {
+        assert_eq!(
+            router_shutdown_budget(ROUTER_STARTING),
+            ROUTER_STARTING_STOP_BUDGET
+        );
+        assert_eq!(
+            router_shutdown_budget(ROUTER_RUNNING),
+            ROUTER_RUNNING_STOP_BUDGET
+        );
+        assert!(ROUTER_STARTING_STOP_BUDGET > ROUTER_RUNNING_STOP_BUDGET);
+    }
+
+    #[tokio::test]
+    async fn cancelling_router_shutdown_keeps_the_thread_owned_and_poisons() {
+        let (cmd_tx, _cmd_rx) = pipewire::channel::channel::<Cmd>();
+        let (release_tx, release_rx) = mpsc::channel();
+        let (health, _) = health::channel();
+        let thread = std::thread::spawn(move || {
+            let _ = release_rx.recv();
+        });
+        let router = StreamRouter {
+            cmd_tx,
+            thread: OwnedThread::new("cancellation fixture", thread, health.clone()),
+            phase: Arc::new(AtomicU8::new(ROUTER_STARTING)),
+        };
+
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), router.shutdown())
+                .await
+                .is_err(),
+            "the outer timeout must cancel shutdown before its policy deadline"
+        );
+        assert!(
+            health.fault().is_some(),
+            "cancellation must poison instead of detaching the OS handle"
+        );
+        release_tx.send(()).expect("release quarantined fixture");
+    }
+
+    #[test]
+    fn bounded_module_worker_uses_the_contained_spawn_path() {
+        let mut command = Command::new("sh");
+        command.args([
+            "-c",
+            "read pid comm state ppid pgrp rest < /proc/self/stat; printf '%s %s' \"$pid\" \"$pgrp\"",
+        ]);
+        let output = bounded_output(&mut command, Duration::from_secs(1))
+            .expect("run contained module-worker fixture");
+        assert!(output.status.success());
+        let ids = String::from_utf8(output.stdout).expect("ascii pid/pgid");
+        let mut ids = ids.split_whitespace();
+        let pid = ids.next().expect("child pid");
+        let pgid = ids.next().expect("child process group");
+        assert_eq!(pid, pgid, "module worker must lead its own process group");
+    }
+
     /// The module table exactly as `--repair` observes it.
     fn module_snapshot() -> Vec<(u32, String, String)> {
         let mut session =
@@ -1148,7 +1328,8 @@ mod tests {
     #[ignore = "loads real Pulse modules; run with --ignored --test-threads=1"]
     async fn live_teardown_leaves_the_module_table_as_it_found_it() {
         let before = module_snapshot();
-        let routing = Routing::start(&whole_desktop_opts())
+        let (health, _) = health::channel();
+        let routing = Routing::start(&whole_desktop_opts(), health)
             .await
             .expect("routing starts");
 
@@ -1180,6 +1361,39 @@ mod tests {
         );
     }
 
+    /// Exercise the real libpipewire mainloop command path, not only the
+    /// whole-desktop module path above. The deliberately unmatched app filter
+    /// is enough to start the router without moving an unrelated live stream.
+    #[tokio::test]
+    #[ignore = "uses the real Pulse/PipeWire graph; run with --ignored --test-threads=1"]
+    async fn live_stream_router_stops_within_its_policy_budget() {
+        let before = module_snapshot();
+        let mut opts = whole_desktop_opts();
+        opts.app = Some("__pixelpass_s3b_no_matching_application__".to_string());
+        opts.strict_audio = true;
+        let (health, _) = health::channel();
+        let routing = Routing::start(&opts, health.clone())
+            .await
+            .expect("per-app routing starts");
+
+        // Let the OS thread reach its normal running phase so this covers the
+        // tighter steady-state budget rather than only startup containment.
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        tokio::time::timeout(Duration::from_secs(10), routing.shutdown())
+            .await
+            .expect("router and graph teardown stay globally bounded");
+
+        assert!(
+            health.fault().is_none(),
+            "an observed shutdown and successful join must remain healthy"
+        );
+        assert_eq!(
+            module_snapshot(),
+            before,
+            "per-app teardown must leave the module table byte-identical"
+        );
+    }
+
     /// The orphan race, staged against a real server: a load cancelled while
     /// `pactl` is in flight must still be findable and removable.
     ///