diff --git a/src/cli.rs b/src/cli.rs index 680b9bc..6949d50 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -1,3 +1,4 @@ +use anyhow::{Result, bail}; use clap::{Parser, ValueEnum}; #[derive(Parser, Debug)] @@ -38,6 +39,11 @@ pub struct Cli { #[arg(long)] pub strict_audio: bool, + /// Internal phase-0d trigger for the not-yet-public desktop-excluding + /// capture plan. Phase 7 replaces this with the versioned public selector. + #[arg(long, hide = true, conflicts_with = "app")] + pub internal_desktop_excluding: bool, + /// Override display server autodetection. #[arg(long, value_enum)] pub display_server: Option, @@ -174,6 +180,35 @@ pub enum Quality { Auto, } +/// Internal input to the typed audio-capture planner. `DesktopExcluding` is +/// deliberately reachable only through a hidden phase-0d trigger until the +/// public selector and capability contract land in phase 7. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(crate) enum CaptureMode { + #[default] + Legacy, + DesktopExcluding, +} + +impl CaptureMode { + fn validate_inputs(self, app: Option<&str>, legacy_null_sink: bool) -> Result<()> { + if self != Self::DesktopExcluding { + return Ok(()); + } + if app.is_some() { + bail!( + "the internal desktop-excluding mode conflicts with --app; refusing to fall back to legacy per-app routing" + ); + } + if legacy_null_sink { + bail!( + "the internal desktop-excluding mode conflicts with PIXELPASS_AUDIO_VIA_NULL_SINK; refusing to load the legacy default-monitor loopback" + ); + } + Ok(()) + } +} + #[derive(Debug, Clone)] pub struct HostOpts { pub window: bool, @@ -194,6 +229,13 @@ pub struct HostOpts { pub no_hwencode: bool, pub max_viewers: Option, pub interactive: bool, + /// Phase-0d internal selector. This is not a public capability or protocol + /// surface; phase 7 owns that promotion. + pub(crate) capture_mode: CaptureMode, + /// Snapshot the legacy dogfood override during CLI resolution. Capture is + /// lazy, so reading the process environment later would let it change modes + /// between ticket creation and the first viewer. + pub(crate) legacy_null_sink: bool, /// Relay override (resolved from `--relay` / `PIXELPASS_RELAY`); None = defaults. pub relay: Option, } @@ -207,8 +249,24 @@ pub struct ViewerOpts { } impl Cli { - pub fn into_host_opts(self, interactive: bool) -> HostOpts { - HostOpts { + pub fn into_host_opts(self, interactive: bool) -> Result { + let legacy_null_sink = std::env::var_os("PIXELPASS_AUDIO_VIA_NULL_SINK").is_some(); + self.into_host_opts_with_legacy_override(interactive, legacy_null_sink) + } + + fn into_host_opts_with_legacy_override( + self, + interactive: bool, + legacy_null_sink: bool, + ) -> Result { + let capture_mode = if self.internal_desktop_excluding { + CaptureMode::DesktopExcluding + } else { + CaptureMode::Legacy + }; + capture_mode.validate_inputs(self.app.as_deref(), legacy_null_sink)?; + + Ok(HostOpts { window: self.window, app: self.app, strict_audio: self.strict_audio, @@ -222,8 +280,10 @@ impl Cli { no_hwencode: self.no_hwencode, max_viewers: self.max_viewers, interactive, + capture_mode, + legacy_null_sink, relay: crate::common::endpoint::relay_override(self.relay.as_deref()), - } + }) } pub fn into_viewer_opts(self, interactive: bool) -> ViewerOpts { @@ -234,3 +294,49 @@ impl Cli { } } } + +#[cfg(test)] +mod tests { + use super::*; + use clap::CommandFactory; + + #[test] + fn phase_0d_trigger_is_hidden_from_help() { + let help = Cli::command().render_long_help().to_string(); + assert!(!help.contains("internal-desktop-excluding")); + } + + #[test] + fn phase_0d_trigger_conflicts_with_app_during_clap_parsing() { + let error = Cli::try_parse_from([ + "pixelpass", + "--host", + "--internal-desktop-excluding", + "--app", + "Firefox", + ]) + .expect_err("clap must reject the hidden mode combined with --app"); + assert_eq!(error.kind(), clap::error::ErrorKind::ArgumentConflict); + } + + #[test] + fn phase_0d_trigger_conflicts_with_legacy_env_override_during_option_resolution() { + let cli = Cli::try_parse_from(["pixelpass", "--host", "--internal-desktop-excluding"]) + .expect("hidden trigger parses"); + let error = cli + .into_host_opts_with_legacy_override(false, true) + .expect_err("legacy override must be rejected before host startup"); + assert!(error.to_string().contains("PIXELPASS_AUDIO_VIA_NULL_SINK")); + } + + #[test] + fn phase_0d_trigger_reaches_the_internal_host_mode() { + let cli = Cli::try_parse_from(["pixelpass", "--host", "--internal-desktop-excluding"]) + .expect("hidden trigger parses"); + let opts = cli + .into_host_opts_with_legacy_override(false, false) + .expect("non-conflicting hidden mode resolves"); + assert_eq!(opts.capture_mode, CaptureMode::DesktopExcluding); + assert!(!opts.legacy_null_sink); + } +} diff --git a/src/host/audio.rs b/src/host/audio.rs index 488d899..675a3f7 100644 --- a/src/host/audio.rs +++ b/src/host/audio.rs @@ -1,21 +1,18 @@ -//! Per-app audio routing. +//! Connection-owned capture-sink and per-app audio routing. //! //! Two cooperating layers: //! -//! - **Null-sink + loopback** (pactl shell-out): a per-PID null-sink -//! `pixelpass_capture_` plus a `module-loopback` that mirrors the -//! default sink's monitor into it. gst captures from the null-sink's -//! monitor, so the viewer hears whatever the user hears — by default. +//! - **Native graph actor** (libpipewire on a dedicated OS thread): owns a +//! non-lingering per-PID sink named `pixelpass_capture_`. When +//! [`HostOpts::app`] is set, the same actor finds matching +//! `Stream/Output/Audio` nodes and writes `target.object` so WirePlumber +//! reroutes them to that sink. The sink disappears with the actor's PipeWire +//! connection, including after SIGKILL. //! -//! - **Per-stream rerouting** (libpipewire on a dedicated OS thread): -//! when [`HostOpts::app`] is set, a [`StreamRouter`] subscribes to the -//! PipeWire registry, finds `Stream/Output/Audio` nodes whose -//! `application.name` matches the filter, and writes -//! `target.object` to the "default" metadata so WirePlumber reroutes -//! them to our null-sink. Once at least one stream is actually routed, -//! the loopback is unloaded — otherwise the viewer would hear the -//! filtered audio twice (once via the routed stream, once via the -//! default-sink monitor loopback). +//! - **Pulse loopbacks** (bounded pactl shell-outs): by default one loopback +//! mirrors the default sink's monitor into the native capture sink. Once at +//! least one selected app stream is routed, that loopback is unloaded so the +//! viewer does not hear the app twice. //! //! - **Local monitor** (pactl shell-out, app mode only): rerouting *moves* //! the chosen app off the sharer's speakers into the null-sink, so without @@ -26,25 +23,21 @@ //! the first routed stream (after the default-sink loopback is gone, so the //! two never coexist and feed back) and unloaded when the app stops. //! -//! pactl is the right tool for the one-shot null-sink/loopback graph -//! mutations. libpipewire is dragged in only when per-stream filtering -//! is requested, because that needs registry-event subscription. +//! Shutdown quiesces route writes, unloads every dependent Pulse loopback, and +//! only then releases the actor connection and native sink. use anyhow::{Context, Result, bail}; -use std::cell::RefCell; use std::collections::BTreeMap; use std::io::{self, Read}; use std::process::{Child, Command, ExitStatus, Stdio}; -use std::rc::Rc; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::time::{Duration, Instant}; use crate::cli::HostOpts; use crate::common::contained; +use crate::host::graph::{AudioGraphOwner, CaptureSinkSpec, GraphEvent, QuiesceOutcome}; 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. @@ -57,13 +50,11 @@ 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; -/// prefer [`Routing::shutdown`] explicitly, which is the only path that can -/// reconcile a load whose outcome was never observed. +/// Owns the native graph actor plus its pactl-loaded dependent modules. Drop +/// unloads modules as a backstop; prefer [`Routing::shutdown`] explicitly, +/// which is the only path that can reconcile a load whose outcome was never +/// observed and acknowledge route restoration. pub struct Routing { /// Every module this host has loaded, is loading, or must ask the server /// about. Shared with the event task, which loads and unloads the two @@ -74,41 +65,42 @@ pub struct Routing { /// exactly like no load at all and its module was left behind. ledger: Arc, sink_name: String, - stream_router: Option, + graph_owner: Option, event_task: Option>, 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. + /// Create the per-PID native sink and graph actor, plus the default-monitor + /// loopback when the selected routing mode permits it. pub(super) async fn start(opts: &HostOpts, health: health::Reporter) -> Result { let pid = std::process::id(); let sink_name = repair_plan::sink_name_for(pid); let ledger = ModuleLedger::new(); - // Construct the owner before the first mutation. Any error or cancellation - // below now drops a real `Routing`, whose backstop closes, quiesces, - // reconciles, and unloads this ledger. Previously the owner did not exist - // until both initial modules had loaded, so constructor failure leaked - // everything loaded up to that point. + // Construct Routing before either ownership layer mutates the graph. Any + // error or cancellation below drops a real owner whose backstop closes, + // reconciles, and unloads the ledger before releasing the native sink. let mut routing = Self { ledger: Arc::clone(&ledger), sink_name: sink_name.clone(), - stream_router: None, + graph_owner: None, event_task: None, health: health.clone(), }; - // Every module this host loads carries an ownership token, minted per - // load, so `--repair` can tell whose pid the name refers to instead of - // assuming the number means the same thing everywhere. Without it a repair - // run in another pid namespace can unload a live host's audio; see - // `repair::plan::OwnerToken`. That same per-load nonce is what lets - // reconciliation identify a module whose load was interrupted before its - // index was ever read. - load_module(&ledger, Shape::LegacyCaptureSink, pid) - .await - .context("failed to load module-null-sink")?; + // S4: the capture sink is a native, non-lingering PipeWire object owned + // by this actor connection, not a module owned by pipewire-pulse. The + // actor also absorbs the per-app router so every sink-owning mode has one + // graph lifetime and one readiness handshake. + let (graph_owner, mut event_rx) = AudioGraphOwner::start( + opts.app.clone(), + CaptureSinkSpec::for_pid(pid), + health.clone(), + ) + .await + .context("failed to start the connection-owned audio graph")?; + debug_assert_eq!(graph_owner.identity().name, sink_name); + routing.graph_owner = Some(graph_owner); // In strict per-app mode we never mirror the default sink: the viewer // must hear *only* the chosen app, never the whole desktop (which would @@ -119,28 +111,31 @@ impl Routing { // 20ms loopback latency keeps the mirrored audio tight; pactl's // default of 200ms is enough to be perceptible. let strict_app = opts.app.is_some() && opts.strict_audio; - if !strict_app { - load_module(&ledger, Shape::LoopbackIntoCapture, pid) - .await - .context("failed to load module-loopback (null-sink cleaned up on Drop)")?; + if !strict_app + && let Err(error) = load_module(&ledger, Shape::LoopbackIntoCapture, pid).await + { + // Once the actor exists, an ordinary constructor error gets a full + // async teardown rather than falling through the narrower + // synchronous Drop backstop and quarantining a responsive thread. + routing.shutdown().await; + return Err(error) + .context("failed to load module-loopback (connection-owned sink cleaned up)"); } tracing::info!( strict_app, %sink_name, - "audio routing: null-sink ready (loopback skipped in strict app mode)" + "audio routing: connection-owned sink ready (loopback skipped in strict app mode)" ); - if let Some(app) = &opts.app { - let (router, mut event_rx) = - StreamRouter::spawn(app.clone(), sink_name.clone(), health.clone())?; + if opts.app.is_some() { let ledger_for_task = Arc::clone(&ledger); let strict = opts.strict_audio; let event_task = tokio::spawn(async move { use crate::common::output::{self, AppAudioState}; while let Some(ev) = event_rx.recv().await { match ev { - Event::FirstRoutedStream => { + GraphEvent::FirstRoutedStream => { tracing::info!( "audio routing: first stream routed → unloading default-sink loopback" ); @@ -167,7 +162,7 @@ impl Routing { state: AppAudioState::Routed, }); } - Event::LastRoutedStreamGone => { + GraphEvent::LastRoutedStreamGone => { // Routed app exited/paused mid-session. Notify the // front-end either way; the recovery differs by mode. output::emit(output::Event::AppAudio { @@ -207,7 +202,6 @@ impl Routing { } } }); - routing.stream_router = Some(router); routing.event_task = Some(event_task); } @@ -228,10 +222,10 @@ impl Routing { &self.sink_name } - /// Stop the stream router and the event task, settle anything the ledger is - /// unsure about, then unload every module in shape order — the loopbacks - /// before the sink they reference, because PipeWire can leave zombie links if - /// a sink is destroyed with active inputs. + /// Quiesce graph mutations, stop the event task, settle anything the ledger + /// is unsure about, unload every dependent loopback, then release the native + /// sink. PipeWire can leave zombie links if a sink is destroyed with active + /// inputs, so that final ordering is load-bearing. /// /// The event task is **awaited, not merely aborted**. Aborting and walking /// away is what left orphans behind: the task's load is an await point now, @@ -241,28 +235,24 @@ impl Routing { pub async fn shutdown(mut self) { // Closing is synchronous and happens first: after this point the event // task cannot register another mutation even if it receives one last - // router event while shutdown is in progress. + // graph event while shutdown is in progress. self.ledger.close(); - let router_stopped = if let Some(router) = self.stream_router.take() { - router.shutdown().await + let graph_quiesced = if let Some(graph_owner) = self.graph_owner.as_mut() { + graph_owner.quiesce().await == QuiesceOutcome::Confirmed } else { true }; if let Some(mut task) = self.event_task.take() { - 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. + if !graph_quiesced { + // An unresponsive graph actor may still own its event sender. + // Cancel and await the task before ledger reconciliation. 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. + // Quiesce closes the event sender while retaining the native + // sink. Abort is the fallback and is awaited through `&mut + // JoinHandle`, so any in-flight affine permit is dropped before + // reconciliation reads the ledger. match tokio::time::timeout(PACTL_BUDGET, &mut task).await { Ok(Ok(())) => {} Ok(Err(e)) => { @@ -299,6 +289,15 @@ impl Routing { } cleanup_modules(&self.ledger).await; + // Loopbacks are gone before the actor connection is released. This is + // the S4 ordering invariant: dependent Pulse modules never outlive the + // native sink they reference during an ordinary shutdown. + if let Some(graph_owner) = self.graph_owner.take() + && !graph_owner.shutdown().await + { + tracing::warn!("audio routing: AudioGraphOwner shutdown was not confirmed"); + } + if !self.ledger.is_clean() { tracing::warn!( settled = self.ledger.is_settled(), @@ -321,14 +320,16 @@ impl Drop for Routing { /// After a completed `shutdown` the ledger holds nothing and this does nothing. fn drop(&mut self) { self.ledger.close(); - if let Some(router) = self.stream_router.take() { - drop(router); - } if let Some(task) = self.event_task.take() { task.abort(); } self.ledger.close_and_wait(); cleanup_modules_blocking(&self.ledger); + // Keep the actor/sink alive until dependent modules have been handled, + // even on this synchronous error/unwind backstop. + if let Some(graph_owner) = self.graph_owner.take() { + drop(graph_owner); + } if !self.ledger.is_clean() { tracing::warn!( settled = self.ledger.is_settled(), @@ -817,271 +818,6 @@ fn cleanup_modules_blocking(ledger: &Arc) { } } -// ────────────────────────────────────────────────────────────────────── -// Per-stream routing (libpipewire thread) -// ────────────────────────────────────────────────────────────────────── - -/// Command from tokio → libpipewire thread. -enum Cmd { - /// Clear `target.object` for everything we routed, then quit the - /// MainLoop so the thread joins. - Shutdown, -} - -/// Event from libpipewire thread → tokio. The pair drives loopback -/// oscillation: unload on `FirstRoutedStream`, re-load on -/// `LastRoutedStreamGone`. Both fire on count-transitions (0→N and N→0 -/// respectively), not on every change. -enum Event { - /// At least one stream is now routed to our sink. Receiver unloads - /// the default-sink loopback so the filtered audio isn't doubled. - FirstRoutedStream, - /// The last routed stream just disappeared (app closed, paused, - /// switched output). Receiver re-loads the default-sink loopback so - /// the viewer doesn't go silent. - LastRoutedStreamGone, -} - -/// Handle to the libpipewire stream-router thread. -pub struct StreamRouter { - cmd_tx: pipewire::channel::Sender, - thread: OwnedThread, - phase: Arc, -} - -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)> { - let (cmd_tx, cmd_rx) = pipewire::channel::channel::(); - let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel::(); - 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 || { - 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")?; - - Ok(( - Self { - cmd_tx, - thread: OwnedThread::new("libpipewire router thread", thread, health), - phase, - }, - event_rx, - )) - } - - async fn shutdown(mut self) -> bool { - // Best-effort: if the send fails the thread is already gone. - let _ = self.cmd_tx.send(Cmd::Shutdown); - 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:#}")); - } - } -} - -/// Body of the libpipewire thread. Owns MainLoop, registry listener, and -/// all PipeWire proxies for the duration of the routing session. -fn run_router( - filter_name: String, - sink_name: String, - cmd_rx: pipewire::channel::Receiver, - event_tx: tokio::sync::mpsc::UnboundedSender, - phase: Arc, - shutdown_observed: Arc, -) -> Result<()> { - use pipewire::{self as pw, types::ObjectType}; - - let main_loop = - pw::main_loop::MainLoopRc::new(None).context("pw main loop construction failed")?; - let context = - pw::context::ContextRc::new(&main_loop, None).context("pw context construction failed")?; - let core = context - .connect_rc(None) - .context("pw core connect failed (is the daemon running?)")?; - let registry = core.get_registry_rc().context("pw get_registry failed")?; - - let state = Rc::new(RefCell::new(RouterState { - sink_serial: None, - default_metadata: None, - routed_node_ids: Vec::new(), - pending: Vec::new(), - })); - - // 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 { - meta.set_property(nid, "target.object", None, None); - } - if !s.routed_node_ids.is_empty() { - tracing::info!( - n = s.routed_node_ids.len(), - "audio routing: cleared target.object on routed streams before quitting" - ); - } - } - main_loop_for_cmd.quit(); - } - }); - - let filter_lower = filter_name.to_ascii_lowercase(); - let sink_name_owned = sink_name.clone(); - let registry_weak = registry.downgrade(); - let state_for_reg = Rc::clone(&state); - let event_tx_for_reg = event_tx.clone(); - let state_for_remove = Rc::clone(&state); - let event_tx_for_remove = event_tx.clone(); - - let _reg_listener = registry - .add_listener_local() - .global(move |obj| { - let Some(reg) = registry_weak.upgrade() else { - return; - }; - match obj.type_ { - ObjectType::Node => { - let Some(props) = obj.props.as_ref() else { - return; - }; - if props.get("node.name") == Some(sink_name_owned.as_str()) { - match props.get("object.serial").and_then(parse_object_serial) { - Some(serial) => { - state_for_reg.borrow_mut().sink_serial = Some(serial); - tracing::info!(serial, "audio routing: pixelpass sink registered"); - try_flush(&state_for_reg, &event_tx_for_reg); - } - // Never silently: without a serial `try_flush` can - // never route anything, so the whole app-filter mode - // is dead and the only symptom is missing audio. - None => tracing::warn!( - node_id = obj.id, - serial = props.get("object.serial").unwrap_or(""), - "audio routing: pixelpass sink has no usable object.serial; \ - stream rerouting disabled" - ), - } - return; - } - if props.get("media.class") != Some("Stream/Output/Audio") { - return; - } - let Some(app) = props.get("application.name") else { - return; - }; - if !app.eq_ignore_ascii_case(&filter_lower) { - return; - } - tracing::info!( - node_id = obj.id, - %app, - "audio routing: matched stream, queued for route" - ); - state_for_reg.borrow_mut().pending.push(obj.id); - try_flush(&state_for_reg, &event_tx_for_reg); - } - ObjectType::Metadata => { - let Some(props) = obj.props.as_ref() else { - return; - }; - if props.get("metadata.name") != Some("default") { - return; - } - let metadata: pw::metadata::Metadata = match reg.bind(obj) { - Ok(m) => m, - Err(e) => { - tracing::warn!("audio routing: bind default metadata failed: {e}"); - return; - } - }; - state_for_reg.borrow_mut().default_metadata = Some(metadata); - tracing::info!("audio routing: default metadata bound"); - try_flush(&state_for_reg, &event_tx_for_reg); - } - _ => {} - } - }) - .global_remove(move |id| { - handle_global_remove(&state_for_remove, &event_tx_for_remove, id); - }) - .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(()) -} - /// Parse a PipeWire `object.serial` property value. /// /// `object.serial` is a **64-bit** monotonically-increasing counter @@ -1102,81 +838,14 @@ pub(crate) fn parse_object_serial(raw: &str) -> Option { raw.parse::().ok() } -struct RouterState { - /// See [`parse_object_serial`] — 64-bit, and not interchangeable with - /// the `u32` node ids in `routed_node_ids` / `pending`. - sink_serial: Option, - default_metadata: Option, - routed_node_ids: Vec, - pending: Vec, -} - -/// Drop the vanished node from `routed_node_ids` and `pending`. If it -/// was the last routed stream, emit `LastRoutedStreamGone` so the -/// tokio side restores the default-sink loopback. -fn handle_global_remove( - state: &Rc>, - event_tx: &tokio::sync::mpsc::UnboundedSender, - id: u32, -) { - let mut s = state.borrow_mut(); - let was_routed = !s.routed_node_ids.is_empty(); - s.routed_node_ids.retain(|&x| x != id); - s.pending.retain(|&x| x != id); - if was_routed && s.routed_node_ids.is_empty() { - tracing::info!( - node_id = id, - "audio routing: last routed stream disappeared" - ); - let _ = event_tx.send(Event::LastRoutedStreamGone); - } -} - -/// Drain pending streams to the sink, but only once both prerequisites -/// (sink serial known + default metadata bound) are in place. Emits -/// `FirstRoutedStream` when routed count crosses 0→N (so it fires -/// each time the count comes back up from zero, not just the first -/// time — pairs with `LastRoutedStreamGone` to oscillate the loopback). -fn try_flush( - state: &Rc>, - event_tx: &tokio::sync::mpsc::UnboundedSender, -) { - let mut s = state.borrow_mut(); - let Some(serial) = s.sink_serial else { return }; - if s.default_metadata.is_none() { - return; - } - if s.pending.is_empty() { - return; - } - let was_empty = s.routed_node_ids.is_empty(); - let serial_str = serial.to_string(); - let pending = std::mem::take(&mut s.pending); - if let Some(meta) = &s.default_metadata { - for nid in &pending { - meta.set_property(*nid, "target.object", Some("Spa:Id"), Some(&serial_str)); - tracing::info!( - node_id = *nid, - sink_serial = serial, - "audio routing: stream routed to pixelpass sink" - ); - } - } - s.routed_node_ids.extend(pending); - if was_empty && !s.routed_node_ids.is_empty() { - let _ = event_tx.send(Event::FirstRoutedStream); - } -} - #[cfg(test)] 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. + /// Whole-desktop routing: the graph actor owns the native sink, while the + /// ledger owns only its default-monitor loopback. fn whole_desktop_opts() -> HostOpts { HostOpts { window: false, @@ -1190,73 +859,12 @@ mod tests { no_hwencode: false, max_viewers: None, interactive: false, + capture_mode: crate::cli::CaptureMode::Legacy, + legacy_null_sink: false, relay: None, } } - #[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::(); - 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"); @@ -1340,8 +948,8 @@ mod tests { .collect(); assert_eq!( ours.len(), - 2, - "the null-sink and its default-sink loopback must both be loaded" + 1, + "only the default-monitor loopback is a Pulse module; the sink is native" ); for (id, name, args) in ours { let fp = classify(&ModuleObservation::new(*id, name, args)) @@ -1361,12 +969,11 @@ 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. + /// Exercise the real graph-actor command path with app routing enabled. The + /// deliberately unmatched filter avoids 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() { + async fn live_audio_graph_owner_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()); @@ -1381,7 +988,7 @@ mod tests { 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"); + .expect("actor and graph teardown stay globally bounded"); assert!( health.fault().is_none(), diff --git a/src/host/audio_plan.rs b/src/host/audio_plan.rs new file mode 100644 index 0000000..21b3989 --- /dev/null +++ b/src/host/audio_plan.rs @@ -0,0 +1,326 @@ +//! Typed selection and ownership for the audio branch of a capture. +//! +//! The enum is deliberately shaped so the desktop-excluding mode cannot carry +//! either unsafe legacy input: +//! +//! - only [`CapturePlan::LegacyDesktop`] can contain [`DefaultMonitor`]; +//! - only [`CapturePlan::PerApp`] can contain [`Routing`], whose legacy mode may +//! load the default-monitor Pulse loopback; +//! - [`CapturePlan::DesktopExcluding`] contains only [`BareCaptureSink`], which +//! has no module ledger or loopback constructor. +//! +//! Phase 6 will add retained native PipeWire links to the last variant. Until +//! then it intentionally captures silence through a real, connection-owned +//! sink reached by the hidden phase-0d CLI trigger. + +use anyhow::{Context, Result, bail}; +use tokio::process::Command; + +use super::audio::Routing; +use super::graph::BareCaptureSink; +use super::health; +use crate::cli::{CaptureMode, HostOpts}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum CapturePlanKind { + LegacyDesktop, + /// Also covers the legacy `PIXELPASS_AUDIO_VIA_NULL_SINK` dogfood path + /// when no app is selected. The odd name is retained from the design of + /// record so the three safety variants stay mechanically recognisable. + PerApp, + DesktopExcluding, +} + +impl CapturePlanKind { + pub(super) fn resolve(opts: &HostOpts) -> Result { + if opts.capture_mode == CaptureMode::DesktopExcluding { + if opts.app.is_some() { + bail!( + "desktop-excluding capture cannot be combined with --app; legacy Routing is forbidden in this mode" + ); + } + if opts.legacy_null_sink { + bail!( + "desktop-excluding capture cannot be combined with PIXELPASS_AUDIO_VIA_NULL_SINK; the default-monitor loopback is forbidden in this mode" + ); + } + return Ok(Self::DesktopExcluding); + } + + if opts.app.is_some() || opts.legacy_null_sink { + Ok(Self::PerApp) + } else { + Ok(Self::LegacyDesktop) + } + } +} + +/// The real default sink's monitor. Construction is private to this module and +/// the value can only inhabit `LegacyDesktop`. +pub(super) struct DefaultMonitor(String); + +impl DefaultMonitor { + async fn resolve() -> Result { + let output = Command::new("pactl") + .arg("get-default-sink") + .output() + .await + .context( + "failed to run `pactl get-default-sink` (install pulseaudio-utils or pipewire-pulse)", + )?; + if !output.status.success() { + bail!( + "pactl get-default-sink failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + ); + } + let sink = String::from_utf8(output.stdout) + .context("default sink name was not UTF-8")? + .trim() + .to_string(); + if sink.is_empty() { + bail!("pactl get-default-sink returned no name (is a sound server running?)"); + } + Ok(Self(format!("{sink}.monitor"))) + } + + fn gst_device_arg(&self) -> String { + format!("device={}", self.0) + } +} + +/// Owns both the typed source choice and every graph object needed to keep that +/// choice alive for the capture lifetime. +pub(super) enum CapturePlan { + LegacyDesktop { source: DefaultMonitor }, + PerApp { routing: Routing }, + DesktopExcluding { capture_sink: BareCaptureSink }, +} + +impl CapturePlan { + pub(super) async fn start(opts: &HostOpts, health: health::Reporter) -> Result { + match CapturePlanKind::resolve(opts)? { + CapturePlanKind::LegacyDesktop => Ok(Self::LegacyDesktop { + source: DefaultMonitor::resolve().await?, + }), + CapturePlanKind::PerApp => Ok(Self::PerApp { + routing: Routing::start(opts, health) + .await + .context("audio routing setup failed")?, + }), + CapturePlanKind::DesktopExcluding => Ok(Self::DesktopExcluding { + capture_sink: BareCaptureSink::start(health) + .await + .context("desktop-excluding capture-sink setup failed")?, + }), + } + } + + /// The only conversion to GStreamer's stringly `pulsesrc device=...` + /// boundary. Callers cannot supply a free-form source string. + pub(super) fn gst_device_arg(&self) -> String { + match self { + Self::LegacyDesktop { source } => source.gst_device_arg(), + Self::PerApp { routing } => format!("device={}.monitor", routing.sink_name()), + Self::DesktopExcluding { capture_sink } => { + format!("device={}", capture_sink.monitor_name()) + } + } + } + + #[cfg(test)] + pub(super) fn legacy_fixture(monitor_name: &str) -> Self { + Self::LegacyDesktop { + source: DefaultMonitor(monitor_name.to_string()), + } + } + + pub(super) async fn shutdown(self) { + match self { + Self::LegacyDesktop { .. } => {} + Self::PerApp { routing } => routing.shutdown().await, + Self::DesktopExcluding { capture_sink } => capture_sink.shutdown().await, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::cli::Quality; + use std::time::{Duration, Instant}; + + fn opts( + app: bool, + strict_audio: bool, + capture_mode: CaptureMode, + legacy_null_sink: bool, + ) -> HostOpts { + HostOpts { + window: false, + app: app.then(|| "Firefox".to_string()), + strict_audio, + display_server: None, + quality: Quality::Auto, + bitrate: None, + framerate: None, + max_height: None, + no_hwencode: false, + max_viewers: None, + interactive: false, + capture_mode, + legacy_null_sink, + relay: None, + } + } + + #[test] + fn complete_phase_0d_mode_matrix() { + let mut rows = 0; + for app in [false, true] { + for strict_audio in [false, true] { + for capture_mode in [CaptureMode::Legacy, CaptureMode::DesktopExcluding] { + for legacy_null_sink in [false, true] { + rows += 1; + let opts = opts(app, strict_audio, capture_mode, legacy_null_sink); + let actual = CapturePlanKind::resolve(&opts); + let conflicts = capture_mode == CaptureMode::DesktopExcluding + && (app || legacy_null_sink); + if conflicts { + assert!( + actual.is_err(), + "conflicting row unexpectedly resolved: {opts:?}" + ); + continue; + } + let expected = if capture_mode == CaptureMode::DesktopExcluding { + CapturePlanKind::DesktopExcluding + } else if app || legacy_null_sink { + CapturePlanKind::PerApp + } else { + CapturePlanKind::LegacyDesktop + }; + assert_eq!(actual.unwrap(), expected, "wrong plan for {opts:?}"); + } + } + } + } + assert_eq!(rows, 16, "the full 2×2×2×2 mode matrix must run"); + } + + #[test] + fn default_monitor_is_confined_to_the_legacy_variant() { + let plan = CapturePlan::LegacyDesktop { + source: DefaultMonitor("alsa_output.fixture.monitor".to_string()), + }; + assert_eq!(plan.gst_device_arg(), "device=alsa_output.fixture.monitor"); + assert!(matches!(plan, CapturePlan::LegacyDesktop { .. })); + } + + fn pulse_source_exists(name: &str) -> bool { + std::process::Command::new("pactl") + .args(["get-source-volume", name]) + .output() + .is_ok_and(|output| output.status.success()) + } + + fn incoming_links(sink_name: &str) -> Result> { + let output = std::process::Command::new("pw-dump") + .output() + .context("run pw-dump for the phase-0d graph assertion")?; + if !output.status.success() { + bail!( + "pw-dump failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + ); + } + let objects: serde_json::Value = + serde_json::from_slice(&output.stdout).context("parse pw-dump JSON")?; + let objects = objects + .as_array() + .context("pw-dump root was not an array")?; + let sink_id = objects + .iter() + .find(|object| { + object.get("type").and_then(serde_json::Value::as_str) + == Some("PipeWire:Interface:Node") + && object + .pointer("/info/props/node.name") + .and_then(serde_json::Value::as_str) + == Some(sink_name) + }) + .and_then(|object| object.get("id")) + .and_then(serde_json::Value::as_u64) + .context("bare capture sink was absent from pw-dump")?; + + Ok(objects + .iter() + .filter(|object| { + if object.get("type").and_then(serde_json::Value::as_str) + != Some("PipeWire:Interface:Link") + { + return false; + } + let Some(input_node) = object.pointer("/info/props/link.input.node") else { + return false; + }; + input_node.as_u64().or_else(|| { + input_node + .as_str() + .and_then(|value| value.parse::().ok()) + }) == Some(sink_id) + }) + .cloned() + .collect()) + } + + /// Phase-0d graph assertion: the hidden mode reaches the real native sink, + /// but neither a Pulse default-monitor module nor any native link feeds it. + #[tokio::test] + #[ignore = "live: mutates the shared PipeWire graph; run alone with --test-threads=1"] + async fn live_desktop_excluding_sink_has_no_legacy_feed() { + let opts = opts(false, false, CaptureMode::DesktopExcluding, false); + let sink_name = crate::repair::plan::sink_name_for(std::process::id()); + assert!(!pulse_source_exists(&format!("{sink_name}.monitor"))); + + let (health, _) = health::channel(); + let plan = CapturePlan::start(&opts, health.clone()) + .await + .expect("start the hidden desktop-excluding plan"); + let capture_sink = match &plan { + CapturePlan::DesktopExcluding { capture_sink } => capture_sink, + _ => panic!("hidden mode constructed the wrong capture-plan variant"), + }; + assert_eq!(capture_sink.sink_name(), sink_name); + assert!(pulse_source_exists(capture_sink.monitor_name())); + + let mut pulse = crate::repair::introspect::PulseSession::connect() + .expect("connect to the local Pulse server"); + let modules = pulse.list_modules().expect("read the Pulse module table"); + let feeding_modules: Vec<_> = modules + .into_iter() + .filter(|module| { + module.name == "module-loopback" + && module.args.contains(&format!("sink={sink_name}")) + }) + .collect(); + assert!( + feeding_modules.is_empty(), + "DesktopExcluding must not construct a Pulse loopback into its sink: {feeding_modules:?}" + ); + assert!( + incoming_links(&sink_name) + .expect("inspect incoming native links") + .is_empty(), + "phase 0d must leave the bare sink unfed until phase 6 owns native links" + ); + + plan.shutdown().await; + let deadline = Instant::now() + Duration::from_secs(2); + while pulse_source_exists(&format!("{sink_name}.monitor")) && Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(!pulse_source_exists(&format!("{sink_name}.monitor"))); + assert!(health.fault().is_none()); + } +} diff --git a/src/host/fanout.rs b/src/host/fanout.rs new file mode 100644 index 0000000..00caf51 --- /dev/null +++ b/src/host/fanout.rs @@ -0,0 +1,333 @@ +//! Pure Phase-6 fan-out planning. +//! +//! This module converts one coherent taint decision into exact per-port link +//! specifications. It deliberately cannot create PipeWire objects: the +//! mutation edge will consume this plan on the observer thread, re-evaluate in +//! that same callback, then retain the resulting non-lingering link proxies. +//! +//! Port identity is `audio.channel`, never registry enumeration order or a +//! recyclable global id. An unknown or incompatible layout makes that stream +//! unsupported; it never produces a guessed partial capture. + +#![allow(dead_code)] + +use std::collections::{BTreeMap, BTreeSet}; + +use super::taint::snapshot::{ + GlobalId, GraphSnapshot, MediaRole, PortDirection, PortSnapshot, Serial, +}; +use super::taint::{Decisions, Reason}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub(super) struct DesiredLink { + pub(super) output_node: GlobalId, + pub(super) output_port: GlobalId, + pub(super) input_node: GlobalId, + pub(super) input_port: GlobalId, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum PlanIssue { + CaptureSinkMissing, + CaptureSinkNotASink, + MissingOutputPorts, + MissingCapturePorts, + UnidentifiedChannel, + AmbiguousCaptureChannel, + DuplicateOutputChannel, + IncompatibleChannelLayout, + ExclusiveCapturePort, +} + +impl PlanIssue { + pub(super) fn code(self) -> &'static str { + match self { + Self::CaptureSinkMissing => "capture-sink-missing", + Self::CaptureSinkNotASink => "capture-sink-not-a-sink", + Self::MissingOutputPorts => "missing-output-ports", + Self::MissingCapturePorts => "missing-capture-ports", + Self::UnidentifiedChannel => "unidentified-channel", + Self::AmbiguousCaptureChannel => "ambiguous-capture-channel", + Self::DuplicateOutputChannel => "duplicate-output-channel", + Self::IncompatibleChannelLayout => "incompatible-channel-layout", + Self::ExclusiveCapturePort => "exclusive-capture-port", + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) enum StreamPlan { + Excluded { reason: Reason }, + Unsupported { issue: PlanIssue }, + Capture { links: BTreeSet }, +} + +/// Plan every candidate in the decision universe against one capture sink. +/// +/// The caller must pass `decisions` produced from this exact `snapshot`. The +/// mutation edge upholds that condition structurally by evaluating and planning +/// inline in one observer callback; keeping this function pure makes the port +/// matrix independently falsifiable. +pub(super) fn plan( + snapshot: &GraphSnapshot, + decisions: &Decisions, + capture_sink: Serial, +) -> BTreeMap { + let sink = snapshot.node(capture_sink); + let sink_issue = match sink { + None => Some(PlanIssue::CaptureSinkMissing), + Some(node) if node.role != MediaRole::Sink => Some(PlanIssue::CaptureSinkNotASink), + Some(_) => None, + }; + + decisions + .candidates + .iter() + .map(|(&serial, decision)| { + let stream_plan = if let Some(reason) = decision.reason() { + StreamPlan::Excluded { reason } + } else if let Some(issue) = sink_issue { + StreamPlan::Unsupported { issue } + } else { + let node = snapshot + .node(serial) + .expect("a decision from this snapshot names a live node"); + match pair_ports( + snapshot.ports_of(node.id), + snapshot.ports_of(sink.expect("validated above").id), + node.id, + sink.expect("validated above").id, + ) { + Ok(links) => StreamPlan::Capture { links }, + Err(issue) => StreamPlan::Unsupported { issue }, + } + }; + (serial, stream_plan) + }) + .collect() +} + +fn pair_ports<'a>( + candidate_ports: impl Iterator, + capture_ports: impl Iterator, + output_node: GlobalId, + input_node: GlobalId, +) -> Result, PlanIssue> { + let outputs: Vec<&PortSnapshot> = candidate_ports + .filter(|port| port.direction == PortDirection::Out && !port.monitor) + .collect(); + if outputs.is_empty() { + return Err(PlanIssue::MissingOutputPorts); + } + let inputs: Vec<&PortSnapshot> = capture_ports + .filter(|port| port.direction == PortDirection::In) + .collect(); + if inputs.is_empty() { + return Err(PlanIssue::MissingCapturePorts); + } + if inputs.iter().any(|port| port.exclusive) { + return Err(PlanIssue::ExclusiveCapturePort); + } + + let mut inputs_by_channel: BTreeMap<&str, GlobalId> = BTreeMap::new(); + for port in &inputs { + let channel = port + .channel + .as_deref() + .ok_or(PlanIssue::UnidentifiedChannel)?; + if inputs_by_channel.insert(channel, port.id).is_some() { + return Err(PlanIssue::AmbiguousCaptureChannel); + } + } + + let mut output_channels = BTreeSet::new(); + let mut links = BTreeSet::new(); + for output in outputs { + let channel = output + .channel + .as_deref() + .ok_or(PlanIssue::UnidentifiedChannel)?; + if !output_channels.insert(channel) { + return Err(PlanIssue::DuplicateOutputChannel); + } + if channel == "MONO" { + for &input_port in inputs_by_channel.values() { + links.insert(DesiredLink { + output_node, + output_port: output.id, + input_node, + input_port, + }); + } + continue; + } + let Some(&input_port) = inputs_by_channel.get(channel) else { + return Err(PlanIssue::IncompatibleChannelLayout); + }; + links.insert(DesiredLink { + output_node, + output_port: output.id, + input_node, + input_port, + }); + } + Ok(links) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::host::taint::fixture::{Graph, PULSE_PID}; + use crate::host::taint::snapshot::{MediaRole, PortDirection}; + use crate::host::taint::{ExclusionCtx, StickyState, evaluate}; + + fn decisions(graph: &Graph) -> (GraphSnapshot, Decisions) { + let snapshot = graph.build(); + let (decisions, _) = evaluate( + &snapshot, + &ExclusionCtx { + pipewire_pulse_pid: Some(PULSE_PID), + graph_ready: true, + ..ExclusionCtx::default() + }, + &StickyState::default(), + ); + (snapshot, decisions) + } + + fn stereo_graph() -> (Graph, Serial, Serial, [GlobalId; 4]) { + let mut graph = Graph::new(); + let app = graph.app_node("music", MediaRole::StreamOutput, 42); + let sink = graph.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43); + let app_fl = graph.port_on_channel(app, PortDirection::Out, false, Some("FL")); + let app_fr = graph.port_on_channel(app, PortDirection::Out, false, Some("FR")); + let sink_fl = graph.port_on_channel(sink, PortDirection::In, false, Some("FL")); + let sink_fr = graph.port_on_channel(sink, PortDirection::In, false, Some("FR")); + ( + graph, + app.serial, + sink.serial, + [app_fl, app_fr, sink_fl, sink_fr], + ) + } + + #[test] + fn stereo_ports_pair_by_channel_not_enumeration_order() { + let (graph, app, sink, [app_fl, app_fr, sink_fl, sink_fr]) = stereo_graph(); + let (snapshot, decisions) = decisions(&graph); + assert_eq!( + plan(&snapshot, &decisions, sink).get(&app), + Some(&StreamPlan::Capture { + links: BTreeSet::from([ + DesiredLink { + output_node: snapshot.node(app).unwrap().id, + output_port: app_fl, + input_node: snapshot.node(sink).unwrap().id, + input_port: sink_fl, + }, + DesiredLink { + output_node: snapshot.node(app).unwrap().id, + output_port: app_fr, + input_node: snapshot.node(sink).unwrap().id, + input_port: sink_fr, + }, + ]), + }) + ); + } + + #[test] + fn mono_fans_to_both_stereo_inputs() { + let mut graph = Graph::new(); + let app = graph.app_node("mono", MediaRole::StreamOutput, 42); + let sink = graph.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43); + let mono = graph.port_on_channel(app, PortDirection::Out, false, Some("MONO")); + let left = graph.port_on_channel(sink, PortDirection::In, false, Some("FL")); + let right = graph.port_on_channel(sink, PortDirection::In, false, Some("FR")); + let (snapshot, decisions) = decisions(&graph); + let StreamPlan::Capture { links } = + plan(&snapshot, &decisions, sink.serial)[&app.serial].clone() + else { + panic!("mono stream must be plannable"); + }; + assert_eq!(links.len(), 2); + assert_eq!( + links + .iter() + .map(|link| link.output_port) + .collect::>(), + BTreeSet::from([mono]) + ); + assert_eq!( + links + .iter() + .map(|link| link.input_port) + .collect::>(), + BTreeSet::from([left, right]) + ); + } + + #[test] + fn unknown_or_incompatible_layout_is_unsupported_not_guessed() { + let (mut graph, app, sink, _) = stereo_graph(); + let surround = graph.app_node("surround", MediaRole::StreamOutput, 44); + graph.port_on_channel(surround, PortDirection::Out, false, Some("FC")); + let unknown = graph.app_node("unknown", MediaRole::StreamOutput, 45); + graph.port_on_channel(unknown, PortDirection::Out, false, None); + let (snapshot, decisions) = decisions(&graph); + let plans = plan(&snapshot, &decisions, sink); + assert!(matches!(plans[&app], StreamPlan::Capture { .. })); + assert_eq!( + plans[&surround.serial], + StreamPlan::Unsupported { + issue: PlanIssue::IncompatibleChannelLayout + } + ); + assert_eq!( + plans[&unknown.serial], + StreamPlan::Unsupported { + issue: PlanIssue::UnidentifiedChannel + } + ); + } + + #[test] + fn excluded_candidate_never_produces_a_link_plan() { + let (mut graph, app, sink, _) = stereo_graph(); + let call = graph.peerspeak_node("call", 7); + graph.port_on_channel(call, PortDirection::Out, false, Some("FL")); + let (snapshot, decisions) = decisions(&graph); + let plans = plan(&snapshot, &decisions, sink); + assert!(matches!(plans[&app], StreamPlan::Capture { .. })); + assert_eq!( + plans[&call.serial], + StreamPlan::Excluded { + reason: Reason::PeerspeakOwned + } + ); + } + + #[test] + fn replacing_the_capture_sink_changes_every_target_port() { + let (mut graph, app, old_sink, _) = stereo_graph(); + let new_sink = graph.native_virtual_node("pixelpass_capture_new", MediaRole::Sink, 46); + let new_fl = graph.port_on_channel(new_sink, PortDirection::In, false, Some("FL")); + let new_fr = graph.port_on_channel(new_sink, PortDirection::In, false, Some("FR")); + let (snapshot, decisions) = decisions(&graph); + let old = plan(&snapshot, &decisions, old_sink); + let new = plan(&snapshot, &decisions, new_sink.serial); + let StreamPlan::Capture { links: old } = &old[&app] else { + panic!("old sink plan"); + }; + let StreamPlan::Capture { links: new } = &new[&app] else { + panic!("new sink plan"); + }; + assert_ne!(old, new); + assert_eq!( + new.iter() + .map(|link| link.input_port) + .collect::>(), + BTreeSet::from([new_fl, new_fr]) + ); + } +} diff --git a/src/host/graph.rs b/src/host/graph.rs new file mode 100644 index 0000000..5db2784 --- /dev/null +++ b/src/host/graph.rs @@ -0,0 +1,1447 @@ +//! Connection-owned PipeWire graph actor. +//! +//! The actor owns one non-lingering native capture sink plus the existing +//! per-application `target.object` router. The sink is therefore scoped to the +//! actor's PipeWire connection: closing or losing that connection removes it, +//! including after SIGKILL. Tokio never receives a bare PipeWire global id to +//! mutate; matching, serial validation, and metadata writes stay ordered on the +//! PipeWire main-loop thread. + +use super::audio::parse_object_serial; +use super::health; +use super::owned_thread::OwnedThread; +use crate::repair::plan as repair_plan; +use anyhow::{Context, Result, bail}; +use pipewire::proxy::ProxyT; +use pipewire::{self as pw, types::ObjectType}; +use std::cell::RefCell; +use std::collections::BTreeMap; +use std::io; +use std::process::Stdio; +use std::rc::Rc; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use std::time::{Duration, Instant}; +use tokio::sync::{mpsc, oneshot}; + +/// A running main loop normally stops in a few milliseconds. Two seconds is +/// deliberately generous while still bounding host teardown. +const GRAPH_RUNNING_STOP_BUDGET: Duration = Duration::from_secs(2); +/// A thread stuck inside PipeWire initialisation cannot process `Stop`. Keep a +/// separate, longer policy deadline for that different failure mode. +const GRAPH_STARTING_STOP_BUDGET: Duration = Duration::from_secs(5); +const GRAPH_NATIVE_READY_BUDGET: Duration = Duration::from_secs(5); +const GRAPH_QUIESCE_BUDGET: Duration = Duration::from_secs(2); +/// Warm measurements on this host are single-digit milliseconds. This wider +/// bound covers a lagging pipewire-pulse bridge without borrowing the actor's +/// thread or turning a missing Pulse namespace into an unbounded wait. +const PULSE_MONITOR_READY_BUDGET: Duration = Duration::from_secs(3); +const PULSE_PROBE_BUDGET: Duration = Duration::from_millis(500); +const PULSE_PROBE_INTERVAL: Duration = Duration::from_millis(20); + +const GRAPH_STARTING: u8 = 0; +const GRAPH_RUNNING: u8 = 1; +const GRAPH_EXITED: u8 = 2; + +/// Exact native capture-sink shape used by S4. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct CaptureSinkSpec { + name: String, + monitor_name: String, +} + +impl CaptureSinkSpec { + pub(super) fn for_pid(pid: u32) -> Self { + let name = repair_plan::sink_name_for(pid); + let monitor_name = format!("{name}.monitor"); + Self { name, monitor_name } + } + + pub(super) fn name(&self) -> &str { + &self.name + } + + pub(super) fn monitor_name(&self) -> &str { + &self.monitor_name + } + + fn properties(&self) -> pw::properties::PropertiesBox { + let mut props = pw::properties::properties! { + "factory.name" => "support.null-audio-sink", + "media.class" => "Audio/Sink", + "audio.channels" => "2", + "audio.position" => "[FL,FR]", + "node.virtual" => "true", + "monitor.channel-volumes" => "true", + // Load-bearing: the remote object must die with this connection. + "object.linger" => "false" + }; + props.insert("node.name", self.name.as_str()); + props + } +} + +/// Identity established from the created proxy's bound id and that exact +/// registry global's serial. `node.name` is intentionally not an identity +/// input: PipeWire accepts duplicate names and Pulse selects the older one. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct SinkIdentity { + pub(super) name: String, + pub(super) monitor_name: String, + pub(super) global_id: u32, + pub(super) serial: u64, +} + +#[derive(Debug)] +pub(super) enum GraphEvent { + FirstRoutedStream, + LastRoutedStreamGone, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum QuiesceOutcome { + Confirmed, + Unconfirmed, +} + +enum Cmd { + /// Stop producing route events, restore routes still owned by PixelPass, + /// and acknowledge the writes with a core round-trip. The sink remains + /// alive so Tokio can unload every dependent Pulse loopback first. + Quiesce { ack: oneshot::Sender }, + /// Restore any still-owned routes, wait for a core round-trip, then quit + /// the main loop. Dropping the connection removes the native non-lingering + /// sink only after those restoration writes have reached the server. + Stop, +} + +/// Tokio-side owner of the PipeWire actor thread. +pub(super) struct AudioGraphOwner { + cmd_tx: pw::channel::Sender, + thread: OwnedThread, + phase: Arc, + identity: SinkIdentity, +} + +/// A connection-owned capture sink with no Pulse-module ledger and no API for +/// constructing the legacy default-monitor loopback. +/// +/// This is the load-bearing type boundary for the phase-0d +/// `DesktopExcluding` plan: owning this value proves the sink exists, but the +/// only operations available are reading its monitor name and shutting down +/// its actor connection. The link manager added in phase 6 will feed it through +/// retained native PipeWire links; legacy `Routing` remains a separate type in +/// `host::audio`. +pub(super) struct BareCaptureSink { + graph_owner: Option, + monitor_name: String, +} + +impl BareCaptureSink { + pub(super) async fn start(health: health::Reporter) -> Result { + let spec = CaptureSinkSpec::for_pid(std::process::id()); + let (graph_owner, _event_rx) = AudioGraphOwner::start(None, spec, health) + .await + .context("failed to start the bare connection-owned capture sink")?; + let identity = graph_owner.identity().clone(); + Ok(Self { + graph_owner: Some(graph_owner), + monitor_name: identity.monitor_name, + }) + } + + #[cfg(test)] + pub(super) fn sink_name(&self) -> &str { + self.monitor_name + .strip_suffix(".monitor") + .expect("capture-sink monitor names always end in .monitor") + } + + pub(super) fn monitor_name(&self) -> &str { + &self.monitor_name + } + + pub(super) async fn shutdown(mut self) { + if let Some(graph_owner) = self.graph_owner.take() + && !graph_owner.shutdown().await + { + tracing::warn!("audio capture: bare AudioGraphOwner shutdown was not confirmed"); + } + } +} + +impl AudioGraphOwner { + pub(super) async fn start( + filter_name: Option, + spec: CaptureSinkSpec, + health: health::Reporter, + ) -> Result<(Self, mpsc::UnboundedReceiver)> { + let (cmd_tx, cmd_rx) = pw::channel::channel::(); + let (event_tx, event_rx) = mpsc::unbounded_channel::(); + let (ready_tx, ready_rx) = oneshot::channel::(); + let phase = Arc::new(AtomicU8::new(GRAPH_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 spec_for_thread = spec.clone(); + + let thread = std::thread::Builder::new() + .name("pixelpass-audio-graph".to_string()) + .spawn(move || { + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + run_graph( + filter_name, + spec_for_thread, + cmd_rx, + event_tx, + ready_tx, + Arc::clone(&phase_for_thread), + Arc::clone(&shutdown_for_thread), + health_for_thread.clone(), + ) + })); + phase_for_thread.store(GRAPH_EXITED, Ordering::Release); + match result { + Ok(result) => report_graph_exit( + &health_for_thread, + shutdown_for_thread.load(Ordering::Acquire), + result, + ), + Err(_) => { + health_for_thread.poison("AudioGraphOwner thread panicked"); + } + } + }) + .context("failed to spawn AudioGraphOwner thread")?; + + // Construct the owner before the first await. Cancellation therefore + // keeps the OS handle owned and sends Stop from Drop. + let mut owner = Self { + cmd_tx, + thread: OwnedThread::new("AudioGraphOwner thread", thread, health.clone()), + phase, + // Replaced after the native handshake; this value is never exposed. + identity: SinkIdentity { + name: spec.name.clone(), + monitor_name: spec.monitor_name.clone(), + global_id: u32::MAX, + serial: 0, + }, + }; + + let identity = match tokio::time::timeout(GRAPH_NATIVE_READY_BUDGET, ready_rx).await { + Ok(Ok(identity)) => identity, + Ok(Err(_)) => { + owner.shutdown().await; + bail!("AudioGraphOwner exited before the native sink became ready"); + } + Err(_) => { + owner.shutdown().await; + bail!( + "AudioGraphOwner did not establish the native sink within {GRAPH_NATIVE_READY_BUDGET:?}" + ); + } + }; + owner.identity = identity.clone(); + + // A PipeWire core round-trip proves only this client's namespace. The + // GStreamer `pulsesrc` path uses pipewire-pulse, so readiness requires a + // separate Pulse lookup from Tokio, never a blocking subprocess on the + // actor thread. + let pulse_ready = wait_for_pulse_monitor(&identity.monitor_name).await; + if let Err(error) = pulse_ready { + owner.shutdown().await; + return Err(error).context("native capture sink was not exported to Pulse"); + } + + tracing::info!( + global_id = identity.global_id, + serial = identity.serial, + sink = %identity.name, + monitor = %identity.monitor_name, + "audio graph: connection-owned capture sink ready" + ); + Ok((owner, event_rx)) + } + + pub(super) fn identity(&self) -> &SinkIdentity { + &self.identity + } + + pub(super) async fn quiesce(&mut self) -> QuiesceOutcome { + let (ack_tx, ack_rx) = oneshot::channel(); + if self.cmd_tx.send(Cmd::Quiesce { ack: ack_tx }).is_err() { + tracing::warn!("audio graph: actor exited before routes could be quiesced"); + return QuiesceOutcome::Unconfirmed; + } + match tokio::time::timeout(GRAPH_QUIESCE_BUDGET, ack_rx).await { + Ok(Ok(true)) => QuiesceOutcome::Confirmed, + Ok(Ok(false) | Err(_)) => { + tracing::warn!("audio graph: route restoration was not acknowledged"); + QuiesceOutcome::Unconfirmed + } + Err(_) => { + tracing::warn!( + "audio graph: route restoration was not acknowledged within {GRAPH_QUIESCE_BUDGET:?}" + ); + QuiesceOutcome::Unconfirmed + } + } + } + + pub(super) async fn shutdown(mut self) -> bool { + let _ = self.cmd_tx.send(Cmd::Stop); + let budget = graph_shutdown_budget(self.phase.load(Ordering::Acquire)); + self.thread.join_within(budget).await + } +} + +impl Drop for AudioGraphOwner { + fn drop(&mut self) { + // If async construction or shutdown is cancelled, wake the MainLoop + // before OwnedThread poisons/quarantines the still-owned handle. + let _ = self.cmd_tx.send(Cmd::Stop); + } +} + +fn graph_shutdown_budget(phase: u8) -> Duration { + if phase == GRAPH_STARTING { + GRAPH_STARTING_STOP_BUDGET + } else { + GRAPH_RUNNING_STOP_BUDGET + } +} + +fn report_graph_exit(health: &health::Reporter, shutdown_observed: bool, result: Result<()>) { + match result { + Ok(()) if shutdown_observed => {} + Ok(()) => { + health.poison("AudioGraphOwner exited without a Stop command"); + } + Err(error) => { + tracing::warn!("audio graph: actor thread exited with error: {error:#}"); + health.poison(format!("AudioGraphOwner failed: {error:#}")); + } + } +} + +async fn wait_for_pulse_monitor(monitor_name: &str) -> Result { + let started = Instant::now(); + let deadline = started + PULSE_MONITOR_READY_BUDGET; + + loop { + let mut command = tokio::process::Command::new("pactl"); + command + .args(["get-source-volume", monitor_name]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + let last_error = match tokio::time::timeout(PULSE_PROBE_BUDGET, command.output()).await { + Ok(Ok(output)) if output.status.success() => return Ok(started.elapsed()), + Ok(Ok(output)) => String::from_utf8_lossy(&output.stderr).trim().to_string(), + Ok(Err(error)) if error.kind() == io::ErrorKind::NotFound => { + return Err(error).context("failed to run `pactl get-source-volume`"); + } + Ok(Err(error)) => error.to_string(), + Err(_) => format!("pactl probe exceeded {PULSE_PROBE_BUDGET:?}"), + }; + + if Instant::now() >= deadline { + bail!( + "Pulse monitor {monitor_name:?} did not become ready within {PULSE_MONITOR_READY_BUDGET:?}: {last_error}" + ); + } + tokio::time::sleep(PULSE_PROBE_INTERVAL).await; + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct ObservedNode { + global_id: u32, + serial: u64, + /// Diagnostic only. Serial equality is the identity gate; epochs do not + /// invalidate an observation merely because unrelated graph traffic moved. + epoch: u64, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +struct MetadataValue { + type_: Option, + value: Option, +} + +#[derive(Clone, Debug)] +struct RouteRecord { + observed: ObservedNode, + prior_target: MetadataValue, +} + +struct BoundMetadata { + // Listener first: its Drop unhooks callbacks that reference the proxy. + _listener: pw::metadata::MetadataListener, + metadata: pw::metadata::Metadata, +} + +struct OwnedSink { + // Listener first for the same teardown-order invariant as BoundMetadata. + _listener: pw::proxy::ProxyListener, + _proxy: pw::node::Node, +} + +#[derive(Default)] +struct NativeReadiness { + bound_global_id: Option, + globals: BTreeMap, + core_synced: bool, +} + +impl NativeReadiness { + fn identity(&self, spec: &CaptureSinkSpec) -> Option { + let global_id = self.bound_global_id?; + let serial = *self.globals.get(&global_id)?; + self.core_synced.then(|| SinkIdentity { + name: spec.name().to_string(), + monitor_name: spec.monitor_name().to_string(), + global_id, + serial, + }) + } +} + +struct ActorState { + spec: CaptureSinkSpec, + filter_lower: Option, + readiness: NativeReadiness, + ready_tx: Option>, + identity: Option, + epoch: u64, + pending: BTreeMap, + routed: BTreeMap, + current_targets: BTreeMap, + default_metadata: Option>, + event_tx: Option>, + initial_sync: Option, + quiesce_sync: Option<(pw::spa::utils::result::AsyncSeq, oneshot::Sender)>, + stop_sync: Option, + closing: bool, +} + +impl ActorState { + fn publish_ready_if_complete(&mut self) { + if self.identity.is_some() { + return; + } + let Some(identity) = self.readiness.identity(&self.spec) else { + return; + }; + self.identity = Some(identity.clone()); + if let Some(tx) = self.ready_tx.take() { + let _ = tx.send(identity); + } + } + + fn observe_global(&mut self, id: u32, serial: u64) { + self.epoch = self.epoch.wrapping_add(1); + self.readiness.globals.insert(id, serial); + self.publish_ready_if_complete(); + } + + fn observe_bound(&mut self, id: u32) { + self.readiness.bound_global_id = Some(id); + self.publish_ready_if_complete(); + } + + fn observe_initial_sync(&mut self) { + self.readiness.core_synced = true; + self.publish_ready_if_complete(); + } + + fn note_matching_stream(&mut self, id: u32, serial: u64) { + if self.closing { + return; + } + self.pending.insert( + id, + ObservedNode { + global_id: id, + serial, + epoch: self.epoch, + }, + ); + } + + fn remove_global(&mut self, id: u32) -> bool { + self.epoch = self.epoch.wrapping_add(1); + self.readiness.globals.remove(&id); + self.pending.remove(&id); + self.current_targets.remove(&id); + let was_routed = !self.routed.is_empty(); + self.routed.remove(&id); + if was_routed + && self.routed.is_empty() + && let Some(tx) = &self.event_tx + { + let _ = tx.send(GraphEvent::LastRoutedStreamGone); + } + self.readiness.bound_global_id == Some(id) + } + + fn update_metadata( + &mut self, + subject: u32, + key: Option<&str>, + type_: Option<&str>, + value: Option<&str>, + ) { + match key { + Some("target.object") => { + self.current_targets.insert( + subject, + MetadataValue { + type_: type_.map(str::to_string), + value: value.map(str::to_string), + }, + ); + } + None => self.current_targets.clear(), + _ => {} + } + } +} + +fn observation_is_current(globals: &BTreeMap, observed: &ObservedNode) -> bool { + globals.get(&observed.global_id) == Some(&observed.serial) +} + +fn try_flush(state: &Rc>) { + let (metadata, actions, notify_first) = { + let mut state = state.borrow_mut(); + if state.closing || !state.readiness.core_synced { + return; + } + let Some(identity) = state.identity.clone() else { + return; + }; + let Some(metadata) = state.default_metadata.clone() else { + return; + }; + let owned_target = identity.serial.to_string(); + let was_empty = state.routed.is_empty(); + let pending = std::mem::take(&mut state.pending); + let mut actions = Vec::new(); + + for observed in pending.into_values() { + if !observation_is_current(&state.readiness.globals, &observed) + || state.routed.contains_key(&observed.global_id) + { + continue; + } + let prior_target = state + .current_targets + .get(&observed.global_id) + .cloned() + .unwrap_or_default(); + state.current_targets.insert( + observed.global_id, + MetadataValue { + type_: Some("Spa:Id".to_string()), + value: Some(owned_target.clone()), + }, + ); + state.routed.insert( + observed.global_id, + RouteRecord { + observed: observed.clone(), + prior_target, + }, + ); + actions.push(observed.global_id); + } + let notify_first = was_empty && !actions.is_empty(); + (metadata, actions, notify_first) + }; + + let owned_target = state + .borrow() + .identity + .as_ref() + .expect("identity existed while staging routes") + .serial + .to_string(); + for id in &actions { + metadata + .metadata + .set_property(*id, "target.object", Some("Spa:Id"), Some(&owned_target)); + tracing::info!(node_id = *id, sink_serial = %owned_target, "audio graph: stream routed"); + } + if notify_first && let Some(tx) = state.borrow().event_tx.as_ref() { + let _ = tx.send(GraphEvent::FirstRoutedStream); + } +} + +fn restoration_for( + record: &RouteRecord, + current: Option<&MetadataValue>, + owned_target: &str, +) -> Option { + let current = current?; + (current.value.as_deref() == Some(owned_target)).then(|| record.prior_target.clone()) +} + +fn prepare_quiesce( + state: &Rc>, +) -> (Option>, Vec<(u32, MetadataValue)>) { + let mut state = state.borrow_mut(); + state.closing = true; + state.event_tx.take(); + state.pending.clear(); + let metadata = state.default_metadata.clone(); + let owned_target = state + .identity + .as_ref() + .map(|identity| identity.serial.to_string()) + .unwrap_or_default(); + let actions = state + .routed + .values() + .filter_map(|record| { + restoration_for( + record, + state.current_targets.get(&record.observed.global_id), + &owned_target, + ) + .map(|value| (record.observed.global_id, value)) + }) + .collect(); + state.routed.clear(); + (metadata, actions) +} + +#[allow(clippy::too_many_arguments)] +fn run_graph( + filter_name: Option, + spec: CaptureSinkSpec, + cmd_rx: pw::channel::Receiver, + event_tx: mpsc::UnboundedSender, + ready_tx: oneshot::Sender, + phase: Arc, + shutdown_observed: Arc, + health: health::Reporter, +) -> Result<()> { + let main_loop = + pw::main_loop::MainLoopRc::new(None).context("pw main loop construction failed")?; + let context = + pw::context::ContextRc::new(&main_loop, None).context("pw context construction failed")?; + let core = context + .connect_rc(None) + .context("pw core connect failed (is the daemon running?)")?; + let registry = core.get_registry_rc().context("pw get_registry failed")?; + + let state = Rc::new(RefCell::new(ActorState { + spec: spec.clone(), + filter_lower: filter_name.map(|name| name.to_ascii_lowercase()), + readiness: NativeReadiness::default(), + ready_tx: Some(ready_tx), + identity: None, + epoch: 0, + pending: BTreeMap::new(), + routed: BTreeMap::new(), + current_targets: BTreeMap::new(), + default_metadata: None, + event_tx: Some(event_tx), + initial_sync: None, + quiesce_sync: None, + stop_sync: None, + closing: false, + })); + + let main_loop_for_cmd = main_loop.clone(); + let core_for_cmd = core.clone(); + let state_for_cmd = Rc::clone(&state); + let shutdown_for_cmd = Arc::clone(&shutdown_observed); + let _cmd_receiver = cmd_rx.attach(main_loop.loop_(), move |cmd| match cmd { + Cmd::Quiesce { ack } => { + let (metadata, restorations) = prepare_quiesce(&state_for_cmd); + if let Some(metadata) = metadata { + for (id, prior) in restorations { + metadata.metadata.set_property( + id, + "target.object", + prior.type_.as_deref(), + prior.value.as_deref(), + ); + } + } + match core_for_cmd.sync(0) { + Ok(seq) => state_for_cmd.borrow_mut().quiesce_sync = Some((seq, ack)), + Err(error) => { + tracing::warn!("audio graph: quiesce core.sync failed: {error}"); + let _ = ack.send(false); + } + } + } + Cmd::Stop => { + shutdown_for_cmd.store(true, Ordering::Release); + let (metadata, restorations) = prepare_quiesce(&state_for_cmd); + if let Some(metadata) = metadata { + for (id, prior) in restorations { + metadata.metadata.set_property( + id, + "target.object", + prior.type_.as_deref(), + prior.value.as_deref(), + ); + } + } + if let Some((_, ack)) = state_for_cmd.borrow_mut().quiesce_sync.take() { + let _ = ack.send(false); + } + match core_for_cmd.sync(0) { + Ok(seq) => state_for_cmd.borrow_mut().stop_sync = Some(seq), + Err(error) => { + tracing::warn!( + "audio graph: final route-restoration core.sync failed: {error}" + ); + main_loop_for_cmd.quit(); + } + } + } + }); + + let state_for_done = Rc::clone(&state); + let main_loop_for_done = main_loop.clone(); + let state_for_core_error = Rc::clone(&state); + let main_loop_for_core_error = main_loop.clone(); + let health_for_core_error = health.clone(); + let _core_listener = core + .add_listener_local() + .done(move |id, seq| { + if id != pw::core::PW_ID_CORE { + return; + } + let (initial_completed, stop_completed) = { + let mut state = state_for_done.borrow_mut(); + let initial_completed = state.initial_sync == Some(seq); + if initial_completed { + state.initial_sync = None; + state.observe_initial_sync(); + } + if state + .quiesce_sync + .as_ref() + .is_some_and(|(pending, _)| *pending == seq) + && let Some((_, ack)) = state.quiesce_sync.take() + { + let _ = ack.send(true); + } + let stop_completed = state.stop_sync == Some(seq); + if stop_completed { + state.stop_sync = None; + } + (initial_completed, stop_completed) + }; + if initial_completed { + // Existing matching streams may have been observed before the + // initial round-trip. Readiness becoming true is itself the + // event that makes those observations routable. + try_flush(&state_for_done); + } + if stop_completed { + main_loop_for_done.quit(); + } + }) + .error(move |id, seq, res, message| { + tracing::warn!(id, seq, result = res, %message, "audio graph: PipeWire core error"); + if let Some((_, ack)) = state_for_core_error.borrow_mut().quiesce_sync.take() { + let _ = ack.send(false); + } + health_for_core_error.poison(format!("PipeWire core error {res}: {message}")); + main_loop_for_core_error.quit(); + }) + .register(); + + let registry_weak = registry.downgrade(); + let state_for_global = Rc::clone(&state); + let state_for_remove = Rc::clone(&state); + let main_loop_for_remove = main_loop.clone(); + let health_for_remove = health.clone(); + let shutdown_for_remove = Arc::clone(&shutdown_observed); + let _registry_listener = registry + .add_listener_local() + .global(move |obj| { + match obj.type_ { + ObjectType::Node => { + let Some(props) = obj.props.as_ref() else { + return; + }; + let Some(serial) = props.get("object.serial").and_then(parse_object_serial) + else { + return; + }; + { + let mut state = state_for_global.borrow_mut(); + state.observe_global(obj.id, serial); + let matches_filter = state.filter_lower.as_ref().is_some_and(|filter| { + props.get("media.class") == Some("Stream/Output/Audio") + && props + .get("application.name") + .is_some_and(|app| app.eq_ignore_ascii_case(filter)) + }); + if matches_filter && state.readiness.bound_global_id != Some(obj.id) { + state.note_matching_stream(obj.id, serial); + } + } + try_flush(&state_for_global); + } + ObjectType::Metadata => { + let Some(props) = obj.props.as_ref() else { + return; + }; + if props.get("metadata.name") != Some("default") { + return; + } + let Some(registry) = registry_weak.upgrade() else { + return; + }; + let metadata: pw::metadata::Metadata = match registry.bind(obj) { + Ok(metadata) => metadata, + Err(error) => { + tracing::warn!("audio graph: bind default metadata failed: {error}"); + return; + } + }; + let weak_state = Rc::downgrade(&state_for_global); + let listener = metadata + .add_listener_local() + .property(move |subject, key, type_, value| { + if let Some(state) = weak_state.upgrade() { + state + .borrow_mut() + .update_metadata(subject, key, type_, value); + // A property callback may be the last missing + // observation for a stream queued earlier. + try_flush(&state); + } + 0 + }) + .register(); + state_for_global.borrow_mut().default_metadata = Some(Rc::new(BoundMetadata { + _listener: listener, + metadata, + })); + try_flush(&state_for_global); + } + _ => {} + } + }) + .global_remove(move |id| { + let own_sink_removed = state_for_remove.borrow_mut().remove_global(id); + if own_sink_removed && !shutdown_for_remove.load(Ordering::Acquire) { + health_for_remove.poison("connection-owned capture sink disappeared unexpectedly"); + main_loop_for_remove.quit(); + } + }) + .register(); + + let sink = core + .create_object::("adapter", &spec.properties()) + .context("could not create the connection-owned capture sink")?; + let state_for_bound = Rc::clone(&state); + let main_loop_for_proxy_remove = main_loop.clone(); + let health_for_proxy_remove = health.clone(); + let shutdown_for_proxy_remove = Arc::clone(&shutdown_observed); + let main_loop_for_proxy_error = main_loop.clone(); + let health_for_proxy_error = health.clone(); + let shutdown_for_proxy_error = Arc::clone(&shutdown_observed); + let listener = sink + .upcast_ref() + .add_listener_local() + .bound(move |global_id| { + state_for_bound.borrow_mut().observe_bound(global_id); + }) + .removed(move || { + if !shutdown_for_proxy_remove.load(Ordering::Acquire) { + health_for_proxy_remove.poison("connection-owned capture-sink proxy was removed"); + main_loop_for_proxy_remove.quit(); + } + }) + .error(move |seq, res, message| { + tracing::warn!(seq, result = res, %message, "audio graph: capture-sink proxy error"); + if !shutdown_for_proxy_error.load(Ordering::Acquire) { + health_for_proxy_error.poison(format!( + "connection-owned capture-sink proxy error {res}: {message}" + )); + main_loop_for_proxy_error.quit(); + } + }) + .register(); + let _owned_sink = OwnedSink { + _listener: listener, + _proxy: sink, + }; + + let initial_sync = core + .sync(0) + .context("AudioGraphOwner initial core.sync failed")?; + state.borrow_mut().initial_sync = Some(initial_sync); + + phase.store(GRAPH_RUNNING, Ordering::Release); + tracing::info!(sink = %spec.name, "audio graph: actor main loop running"); + main_loop.run(); + tracing::info!(sink = %spec.name, "audio graph: actor main loop exiting"); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::cli::{HostOpts, Quality}; + use crate::repair::plan::{ModuleObservation, Shape, classify}; + use nix::sys::signal::Signal; + use std::io::{BufRead, BufReader, Write}; + use std::process::{Child, Command, Stdio}; + use std::sync::mpsc as std_mpsc; + + const S5_HOST_HELPER: &str = "PIXELPASS_S5_HOST_HELPER"; + const S5_READY_PREFIX: &str = "PIXELPASS_S5_READY="; + + fn spec(name: &str) -> CaptureSinkSpec { + CaptureSinkSpec { + name: name.to_string(), + monitor_name: format!("{name}.monitor"), + } + } + + #[test] + fn capture_sink_spec_is_non_lingering_and_pulse_compatible() { + let spec = CaptureSinkSpec::for_pid(4242); + assert_eq!(spec.name(), "pixelpass_capture_4242"); + assert_eq!(spec.monitor_name(), "pixelpass_capture_4242.monitor"); + let props = spec.properties(); + let props = props.dict(); + assert_eq!(props.get("factory.name"), Some("support.null-audio-sink")); + assert_eq!(props.get("node.name"), Some("pixelpass_capture_4242")); + assert_eq!(props.get("media.class"), Some("Audio/Sink")); + assert_eq!(props.get("audio.channels"), Some("2")); + assert_eq!(props.get("audio.position"), Some("[FL,FR]")); + assert_eq!(props.get("node.virtual"), Some("true")); + assert_eq!(props.get("monitor.channel-volumes"), Some("true")); + assert_eq!(props.get("object.linger"), Some("false")); + } + + #[test] + fn readiness_correlates_the_proxy_bound_id_not_a_duplicate_name() { + let spec = spec("duplicate_name_is_not_identity"); + let mut readiness = NativeReadiness::default(); + // A different global could carry the same node.name. Readiness has no + // name input at all and therefore cannot accidentally select it. + readiness.globals.insert(11, 101); + readiness.bound_global_id = Some(12); + readiness.core_synced = true; + assert!(readiness.identity(&spec).is_none()); + + readiness.globals.insert(12, 202); + assert_eq!( + readiness.identity(&spec), + Some(SinkIdentity { + name: spec.name.clone(), + monitor_name: spec.monitor_name.clone(), + global_id: 12, + serial: 202, + }) + ); + } + + #[test] + fn readiness_requires_the_core_round_trip_as_well_as_identity() { + let spec = spec("roundtrip_gate"); + let mut readiness = NativeReadiness::default(); + readiness.globals.insert(7, 70); + readiness.bound_global_id = Some(7); + assert!(readiness.identity(&spec).is_none()); + readiness.core_synced = true; + assert_eq!(readiness.identity(&spec).unwrap().serial, 70); + } + + #[test] + fn route_gate_revalidates_serial_but_not_unrelated_epoch_churn() { + let observed = ObservedNode { + global_id: 9, + serial: 90, + epoch: 1, + }; + let mut globals = BTreeMap::from([(9, 90)]); + assert!(observation_is_current(&globals, &observed)); + + // Epoch is diagnostic only: unrelated graph traffic must not invalidate + // a still-live identity. + let later_epoch = ObservedNode { + epoch: 500, + ..observed.clone() + }; + assert!(observation_is_current(&globals, &later_epoch)); + + // A recycled global id is a different object and must never be routed + // from the stale observation. + globals.insert(9, 91); + assert!(!observation_is_current(&globals, &observed)); + } + + #[test] + fn route_restoration_never_overwrites_a_later_owner() { + let record = RouteRecord { + observed: ObservedNode { + global_id: 9, + serial: 90, + epoch: 1, + }, + prior_target: MetadataValue { + type_: Some("Spa:Id".to_string()), + value: Some("44".to_string()), + }, + }; + let ours = MetadataValue { + type_: Some("Spa:Id".to_string()), + value: Some("55".to_string()), + }; + assert_eq!( + restoration_for(&record, Some(&ours), "55"), + Some(record.prior_target.clone()) + ); + let user_override = MetadataValue { + type_: Some("Spa:Id".to_string()), + value: Some("66".to_string()), + }; + assert_eq!(restoration_for(&record, Some(&user_override), "55"), None); + assert_eq!(restoration_for(&record, None, "55"), None); + } + + #[test] + fn shutdown_has_distinct_starting_and_running_budgets() { + assert_eq!( + graph_shutdown_budget(GRAPH_STARTING), + GRAPH_STARTING_STOP_BUDGET + ); + assert_eq!( + graph_shutdown_budget(GRAPH_RUNNING), + GRAPH_RUNNING_STOP_BUDGET + ); + assert!(GRAPH_STARTING_STOP_BUDGET > GRAPH_RUNNING_STOP_BUDGET); + } + + #[test] + fn graph_exit_is_healthy_only_after_stop_was_observed() { + let (clean, _) = health::channel(); + report_graph_exit(&clean, true, Ok(())); + assert!(clean.fault().is_none()); + + let (unexpected, _) = health::channel(); + report_graph_exit(&unexpected, false, Ok(())); + assert_eq!( + unexpected.fault().as_deref(), + Some("AudioGraphOwner exited without a Stop command") + ); + + let (failed, _) = health::channel(); + report_graph_exit(&failed, false, Err(anyhow::anyhow!("fixture failure"))); + assert!( + failed + .fault() + .as_deref() + .is_some_and(|reason| reason.contains("fixture failure")) + ); + } + + #[tokio::test] + async fn cancelling_shutdown_keeps_the_graph_thread_owned_and_poisons() { + let (cmd_tx, _cmd_rx) = pw::channel::channel::(); + let (release_tx, release_rx) = std_mpsc::channel(); + let (health, _) = health::channel(); + let thread = std::thread::spawn(move || { + let _ = release_rx.recv(); + }); + let owner = AudioGraphOwner { + cmd_tx, + thread: OwnedThread::new("graph cancellation fixture", thread, health.clone()), + phase: Arc::new(AtomicU8::new(GRAPH_STARTING)), + identity: SinkIdentity { + name: "fixture".to_string(), + monitor_name: "fixture.monitor".to_string(), + global_id: 1, + serial: 1, + }, + }; + + assert!( + tokio::time::timeout(Duration::from_millis(20), owner.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"); + } + + async fn pulse_source_exists(name: &str) -> bool { + tokio::process::Command::new("pactl") + .args(["get-source-volume", name]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .await + .is_ok_and(|status| status.success()) + } + + fn whole_desktop_opts() -> HostOpts { + HostOpts { + window: false, + app: None, + strict_audio: false, + display_server: None, + quality: Quality::Auto, + bitrate: None, + framerate: None, + max_height: None, + no_hwencode: false, + max_viewers: None, + interactive: false, + capture_mode: crate::cli::CaptureMode::Legacy, + legacy_null_sink: false, + relay: None, + } + } + + /// Subprocess half of the S5 two-host gate. The outer test invokes this + /// exact test twice, so each helper owns a distinct PipeWire connection, + /// process id, native sink, and ownership-tagged Pulse loopback. SIGINT + /// requests an ordinary teardown; SIGKILL deliberately skips it. + #[tokio::test] + async fn s5_connection_owned_host_helper() { + if std::env::var_os(S5_HOST_HELPER).is_none() { + return; + } + + let (health, _) = health::channel(); + let routing = super::super::audio::Routing::start(&whole_desktop_opts(), health.clone()) + .await + .expect("S5 helper routing starts"); + assert!(health.fault().is_none()); + println!( + "{S5_READY_PREFIX}{} {}", + std::process::id(), + routing.sink_name() + ); + std::io::stdout().flush().expect("flush S5 readiness"); + + tokio::signal::ctrl_c() + .await + .expect("S5 parent requests a clean SIGINT stop"); + + routing.shutdown().await; + assert!(health.fault().is_none()); + } + + struct S5Host { + child: Option, + pid: u32, + sink_name: String, + } + + impl S5Host { + fn spawn() -> Result { + let executable = std::env::current_exe().context("locate the test executable")?; + let mut command = Command::new(executable); + command + .args([ + "--exact", + "host::graph::tests::s5_connection_owned_host_helper", + "--nocapture", + "--test-threads=1", + ]) + .env(S5_HOST_HELPER, "1") + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()); + let child = crate::common::contained::spawn(&mut command) + .context("spawn a contained S5 host helper")?; + let child_pid = child.id(); + // From this point every early return owns a kill-and-reap fallback. + // A readiness timeout must not strand a graph-mutating helper. + let mut host = Self { + child: Some(child), + pid: child_pid, + sink_name: String::new(), + }; + let stdout = host + .child + .as_mut() + .expect("S5 host was just constructed") + .stdout + .take() + .context("S5 helper stdout was not piped")?; + let (line_tx, line_rx) = std_mpsc::sync_channel(1); + std::thread::spawn(move || { + let mut readiness_sent = false; + for line in BufReader::new(stdout) + .lines() + .map_while(std::result::Result::ok) + { + if !readiness_sent && line.contains(S5_READY_PREFIX) { + readiness_sent = true; + let _ = line_tx.send(Some(line)); + } + // Keep draining after readiness. Dropping this pipe while the + // helper still owns it makes libtest fail its final result + // write with BrokenPipe, masking a clean graph shutdown. + } + if !readiness_sent { + let _ = line_tx.send(None); + } + }); + + let ready_line = line_rx + .recv_timeout(Duration::from_secs(10)) + .context("S5 helper did not report readiness within 10 seconds")? + .context("S5 helper exited before reporting readiness")?; + let marker = ready_line + .split_once(S5_READY_PREFIX) + .map(|(_, value)| value) + .context("malformed S5 readiness marker")?; + let mut fields = marker.split_whitespace(); + let reported_pid: u32 = fields + .next() + .context("S5 readiness omitted the pid")? + .parse() + .context("S5 readiness pid was not numeric")?; + let sink_name = fields + .next() + .context("S5 readiness omitted the sink name")? + .to_string(); + if reported_pid != child_pid { + bail!("S5 helper reported pid {reported_pid}, expected {child_pid}"); + } + + host.sink_name = sink_name; + Ok(host) + } + + fn kill_and_reap(&mut self) -> Result<()> { + if self.child.is_none() { + return Ok(()); + } + crate::common::contained::signal_group(self.pid, Signal::SIGKILL) + .context("SIGKILL the first S5 host")?; + let mut child = self.child.take().expect("S5 child stayed owned"); + child.wait().context("reap the first S5 host")?; + Ok(()) + } + + fn interrupt_and_reap_within(&mut self, budget: Duration) -> Result { + if self.child.is_none() { + return Ok(Duration::ZERO); + } + let started = Instant::now(); + crate::common::contained::signal_group(self.pid, Signal::SIGINT) + .context("SIGINT the surviving S5 host")?; + let deadline = started + budget; + loop { + let status = self + .child + .as_mut() + .expect("S5 child stayed owned across the wait") + .try_wait() + .context("poll the S5 helper")?; + if let Some(status) = status { + if status.success() { + self.child.take(); + return Ok(started.elapsed()); + } + bail!("S5 helper SIGINT stop exited with {status}"); + } + if Instant::now() >= deadline { + let _ = crate::common::contained::signal_group(self.pid, Signal::SIGKILL); + if let Some(mut child) = self.child.take() { + let _ = child.wait(); + } + bail!("S5 helper did not stop cleanly within {budget:?}"); + } + std::thread::sleep(Duration::from_millis(20)); + } + } + } + + impl Drop for S5Host { + fn drop(&mut self) { + if let Some(mut child) = self.child.take() { + let _ = crate::common::contained::signal_group(self.pid, Signal::SIGKILL); + let _ = child.wait(); + } + } + } + + fn capture_sink_serial(name: &str) -> Result> { + let output = Command::new("pw-dump") + .output() + .context("run pw-dump for the S5 ownership gate")?; + if !output.status.success() { + bail!( + "pw-dump failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + ); + } + let objects: serde_json::Value = + serde_json::from_slice(&output.stdout).context("parse pw-dump JSON")?; + let Some(objects) = objects.as_array() else { + bail!("pw-dump root was not an array"); + }; + for object in objects { + if object.get("type").and_then(serde_json::Value::as_str) + != Some("PipeWire:Interface:Node") + { + continue; + } + let Some(props) = object.pointer("/info/props") else { + continue; + }; + if props.get("node.name").and_then(serde_json::Value::as_str) != Some(name) { + continue; + } + let Some(serial) = props.get("object.serial") else { + bail!("capture sink {name:?} had no object.serial"); + }; + let serial = serial + .as_u64() + .or_else(|| serial.as_str().and_then(|value| value.parse::().ok())); + return serial + .map(Some) + .context("capture sink object.serial was not an integer"); + } + Ok(None) + } + + fn owned_loopbacks(pid: u32) -> Result> { + let mut pulse = crate::repair::introspect::PulseSession::connect() + .context("connect to the local Pulse server")?; + let modules = pulse.list_modules().context("list Pulse modules")?; + Ok(modules + .into_iter() + .filter_map(|module| { + classify(&ModuleObservation::new( + module.id, + &module.name, + &module.args, + )) + }) + .filter(|fingerprint| { + fingerprint.pid == pid && fingerprint.shape == Shape::LoopbackIntoCapture + }) + .map(|fingerprint| fingerprint.id) + .collect()) + } + + fn wait_for_sink_state(name: &str, expected_serial: Option) -> Result<()> { + let deadline = Instant::now() + Duration::from_secs(3); + loop { + if capture_sink_serial(name)? == expected_serial { + return Ok(()); + } + if Instant::now() >= deadline { + bail!("capture sink {name:?} did not reach serial state {expected_serial:?}"); + } + std::thread::sleep(Duration::from_millis(20)); + } + } + + /// S5 ownership exit gate: two real processes own two independent native + /// sinks. SIGKILL must remove exactly the killed process's sink, and + /// ownership-aware repair must remove its orphaned Pulse loopback without + /// touching the still-live host. The survivor then exercises ordinary + /// teardown so the gate itself leaves no graph residue. + #[tokio::test] + #[ignore = "live: starts two graph-owning processes, SIGKILLs one, and runs --repair"] + async fn live_two_host_sigkill_and_repair_preserve_the_survivor() { + let mut first = S5Host::spawn().expect("start the first S5 host"); + let mut second = S5Host::spawn().expect("start the second S5 host"); + + let first_serial = capture_sink_serial(&first.sink_name) + .expect("observe the first S5 sink") + .expect("the first S5 sink is present"); + let second_serial = capture_sink_serial(&second.sink_name) + .expect("observe the second S5 sink") + .expect("the second S5 sink is present"); + assert_ne!(first.sink_name, second.sink_name); + assert_ne!(first_serial, second_serial); + assert_eq!(owned_loopbacks(first.pid).unwrap().len(), 1); + let second_modules = owned_loopbacks(second.pid).unwrap(); + assert_eq!(second_modules.len(), 1); + + first.kill_and_reap().expect("SIGKILL the first S5 host"); + wait_for_sink_state(&first.sink_name, None) + .expect("the killed host's connection-owned sink disappears"); + wait_for_sink_state(&second.sink_name, Some(second_serial)) + .expect("the live host's connection-owned sink remains"); + + crate::repair::run(false) + .await + .expect("ownership-aware repair succeeds"); + assert!(owned_loopbacks(first.pid).unwrap().is_empty()); + assert_eq!( + owned_loopbacks(second.pid).unwrap(), + second_modules, + "repair must not unload the live host's module" + ); + wait_for_sink_state(&second.sink_name, Some(second_serial)) + .expect("repair must not disturb the live host's native sink"); + + let stop_elapsed = second + .interrupt_and_reap_within(Duration::from_secs(2)) + .expect("the surviving S5 host honours SIGINT inside PeerSpeak's grace"); + println!("S5_ACTIVE_SIGINT_ELAPSED_MS={}", stop_elapsed.as_millis()); + assert!( + stop_elapsed < Duration::from_secs(2), + "active graph teardown reached PeerSpeak's SIGKILL fallback boundary: {stop_elapsed:?}" + ); + wait_for_sink_state(&second.sink_name, None) + .expect("the surviving host's sink disappears after clean shutdown"); + assert!(owned_loopbacks(second.pid).unwrap().is_empty()); + } + + #[tokio::test] + #[ignore = "creates a real connection-owned PipeWire sink; serialize live audio tests"] + async fn live_sink_survives_quiesce_and_dies_with_its_owner() { + let spec = CaptureSinkSpec::for_pid(std::process::id()); + assert!( + !pulse_source_exists(spec.monitor_name()).await, + "live fixture name is already present" + ); + let (health, _) = health::channel(); + let (mut owner, _events) = AudioGraphOwner::start(None, spec.clone(), health.clone()) + .await + .expect("connection-owned sink starts"); + assert_ne!(owner.identity().global_id, u32::MAX); + assert!(owner.identity().serial > 0); + assert!(pulse_source_exists(spec.monitor_name()).await); + + assert_eq!(owner.quiesce().await, QuiesceOutcome::Confirmed); + assert!( + pulse_source_exists(spec.monitor_name()).await, + "quiesce must retain the sink until dependent modules are removed" + ); + assert!(owner.shutdown().await); + + let deadline = Instant::now() + Duration::from_secs(2); + while pulse_source_exists(spec.monitor_name()).await && Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!( + !pulse_source_exists(spec.monitor_name()).await, + "sink must disappear with the actor connection" + ); + assert!(health.fault().is_none()); + } +} diff --git a/src/host/mod.rs b/src/host/mod.rs index 087ee06..78e43fc 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -1,7 +1,10 @@ pub mod aec; pub mod audio; +mod audio_plan; pub mod audit; mod capture; +mod fanout; +mod graph; mod health; pub mod ledger; mod observer; @@ -564,7 +567,9 @@ fn copy_to_clipboard(text: &str) -> bool { fn capture_summary(opts: &HostOpts) -> String { let mut bits = vec![if opts.window { "window" } else { "fullscreen" }.to_string()]; - if let Some(app) = &opts.app { + if opts.capture_mode == crate::cli::CaptureMode::DesktopExcluding { + bits.push("desktop-excluding-audio (internal)".to_string()); + } else if let Some(app) = &opts.app { if opts.strict_audio { bits.push(format!("app-audio={app} (strict)")); } else { @@ -594,6 +599,8 @@ mod tests { no_hwencode: false, max_viewers: None, interactive: false, + capture_mode: crate::cli::CaptureMode::Legacy, + legacy_null_sink: false, relay: None, } } @@ -617,6 +624,13 @@ mod tests { capture_summary(&opts(None, true)), "fullscreen + system-audio" ); + + let mut desktop_excluding = opts(None, false); + desktop_excluding.capture_mode = crate::cli::CaptureMode::DesktopExcluding; + assert_eq!( + capture_summary(&desktop_excluding), + "fullscreen + desktop-excluding-audio (internal)" + ); } #[test] diff --git a/src/host/observer/adapter.rs b/src/host/observer/adapter.rs index 7efcc5e..b375291 100644 --- a/src/host/observer/adapter.rs +++ b/src/host/observer/adapter.rs @@ -443,6 +443,7 @@ fn run_observer( id, node, direction, + channel: props.get("audio.channel").map(str::to_string), exclusive: truthy(props.get("port.exclusive")), monitor: truthy(props.get("port.monitor")), }), @@ -689,6 +690,10 @@ fn peerspeak_owned(value: Option<&str>) -> bool { } fn node_observation_from_props(props: &pw::spa::utils::dict::DictRef) -> NodeObservation { + let device_id = props + .get("device.id") + .and_then(|value| value.parse::().ok()) + .map(GlobalId); NodeObservation { name: props.get("node.name").map(str::to_string), role: MediaRole::parse(props.get("media.class")), @@ -710,13 +715,11 @@ fn node_observation_from_props(props: &pw::spa::utils::dict::DictRef) -> NodeObs .get("application.process.id") .and_then(|value| value.parse::().ok()), passthrough: truthy(props.get("node.passthrough")), + device_id, session_device: false, }, device_claim: DeviceClaim { - device_id: props - .get("device.id") - .and_then(|value| value.parse::().ok()) - .map(GlobalId), + device_id, device_api: props.get("device.api").map(str::to_string), factory_name: props.get("factory.name").map(str::to_string), alsa_driver_name: props.get("alsa.driver_name").map(str::to_string), @@ -832,6 +835,7 @@ mod tests { props.insert("media.class", "Stream/Output/Audio"); props.insert("node.name", "probe"); props.insert("client.id", "42"); + props.insert("device.id", "77"); if let Some(value) = value { props.insert(PEERSPEAK_OWNED_PROP, value); } @@ -846,6 +850,8 @@ mod tests { assert_eq!(observation.role, MediaRole::StreamOutput); assert_eq!(observation.name.as_deref(), Some("probe")); assert_eq!(observation.props.client_id, Some(GlobalId(42))); + assert_eq!(observation.props.device_id, Some(GlobalId(77))); + assert_eq!(observation.device_claim.device_id, Some(GlobalId(77))); } } @@ -1168,9 +1174,19 @@ mod tests { .is_some_and(|name| name.contains("alsa")) || matches!(node.role, MediaRole::Sink | MediaRole::Source)) }); + let session_device = session_device.expect( + "a named ALSA or Audio/Sink/Audio/Source node must classify as a session device", + ); assert!( - session_device.is_some(), - "a named ALSA or Audio/Sink/Audio/Source node must classify as a session device" + session_device.props.device_id.is_some(), + "a live session device must retain the device.id used by the hardware bridge" + ); + assert!( + projection + .snapshot + .ports() + .any(|port| port.channel.is_some()), + "at least one live audio port must retain audio.channel for Phase-6 pairing" ); assert!(projection.graph_ready); diff --git a/src/host/observer/tests.rs b/src/host/observer/tests.rs index e594f47..e5c8afe 100644 --- a/src/host/observer/tests.rs +++ b/src/host/observer/tests.rs @@ -74,10 +74,14 @@ fn device_with(api: Option<&str>, driver: Option<&str>) -> DeviceProps { } fn obs(name: &str, role: MediaRole, claim: DeviceClaim) -> NodeObservation { + let device_id = claim.device_id; NodeObservation { name: Some(name.to_string()), role, - props: NodeProps::default(), + props: NodeProps { + device_id, + ..NodeProps::default() + }, device_claim: claim, } } @@ -148,6 +152,7 @@ fn port(serial: u64, id: u32, node_id: u32, dir: PortDirection) -> RegEvent { id: gid(id), node: gid(node_id), direction: dir, + channel: None, exclusive: false, monitor: false, }) @@ -911,6 +916,13 @@ fn model_readiness_does_not_release_with_obligation_outstanding() { }); assert_eq!(m.readiness(), Readiness::Complete); assert!(m.graph_ready()); + let projected = m.project(); + let device_node = projected + .snapshot + .node(ser(100)) + .expect("resolved device node is projected"); + assert!(device_node.props.session_device); + assert_eq!(device_node.props.device_id, Some(gid(42))); } #[test] diff --git a/src/host/pipeline.rs b/src/host/pipeline.rs index 61cc163..5700fcb 100644 --- a/src/host/pipeline.rs +++ b/src/host/pipeline.rs @@ -5,7 +5,7 @@ //! the [`Serve`] fanout binding, and the [`CaptureHandle`] lifecycle — is shared //! and lives here. Backends call [`spawn`] with just their source-element args. -use anyhow::{Context, Result, bail}; +use anyhow::{Context, Result}; use nix::sys::signal::Signal; use std::process::Stdio; use std::sync::Arc; @@ -14,7 +14,7 @@ use std::time::Duration; use tokio::process::{Child, Command}; use tokio::time::timeout; -use super::audio::Routing; +use super::audio_plan::CapturePlan; use super::health; use super::quality::EffectiveQuality; use super::serve::Serve; @@ -119,7 +119,7 @@ impl Drop for CaptureProcess { pub(super) struct CaptureHandle { gst: Option, - audio: Option, + audio: Option, serve: Option, stopping: Arc, } @@ -142,8 +142,8 @@ impl CaptureHandle { if let Some(mut gst) = self.gst.take() { gst.shutdown().await; } - if let Some(audio) = self.audio.take() { - audio.shutdown().await; + if let Some(audio_plan) = self.audio.take() { + audio_plan.shutdown().await; } if let Some(serve) = self.serve.take() { serve.shutdown().await; @@ -155,7 +155,7 @@ impl Drop for CaptureHandle { fn drop(&mut self) { self.stopping.store(true, Ordering::Release); // CaptureProcess kills the whole process group and poisons the host; - // Routing's and Serve's own Drop impls handle their respective layers. + // the typed plan's inner owner and Serve handle their own Drop layers. } } @@ -176,8 +176,8 @@ pub(super) async fn spawn( health: health::Reporter, after_spawn: impl FnOnce(), ) -> Result { - let (audio_routing, audio_device) = setup_audio(opts, health.clone()).await?; - let args = build_args(&source_args, &audio_device, opts, quality, source_dims); + let audio_plan = CapturePlan::start(opts, health.clone()).await?; + let args = build_args(&source_args, &audio_plan, opts, quality, source_dims); let mut gst_cmd = Command::new("gst-launch-1.0"); gst_cmd @@ -207,42 +207,12 @@ pub(super) async fn spawn( Ok(CaptureHandle { gst: Some(gst), - audio: audio_routing, + audio: Some(audio_plan), serve: Some(serve), stopping, }) } -/// Decide whether per-app audio routing is active and produce the `device=…` -/// argument for `pulsesrc`. Routing activates when either `--app` is set -/// (per-stream rerouting to a per-PID null-sink) or `PIXELPASS_AUDIO_VIA_NULL_SINK=1` -/// is set (no app filter — captures everything via the null-sink, used for -/// dogfooding the loopback path). Otherwise we capture the default sink's -/// monitor (system audio out), not the default source (the mic). -async fn setup_audio( - opts: &HostOpts, - health: health::Reporter, -) -> Result<(Option, String)> { - let routing_requested = - opts.app.is_some() || std::env::var_os("PIXELPASS_AUDIO_VIA_NULL_SINK").is_some(); - let audio_routing = if routing_requested { - Some( - Routing::start(opts, health) - .await - .context("audio routing setup failed")?, - ) - } else { - None - }; - let audio_device = if let Some(r) = &audio_routing { - format!("device={}.monitor", r.sink_name()) - } else { - let default = default_audio_monitor().await?; - format!("device={default}") - }; - Ok((audio_routing, audio_device)) -} - /// Build the full gst-launch argument vector: MPEG-TS mux + fdsink, then the /// video branch (caller's `source` → videorate cap → optional downscale → /// encoder → h264parse → mux.), then the audio branch (pulsesrc → AAC → mux.). @@ -252,7 +222,7 @@ async fn setup_audio( /// wants I420). fn build_args( source: &[String], - audio_device: &str, + audio_plan: &CapturePlan, opts: &HostOpts, quality: &EffectiveQuality, source_dims: Option<(u32, u32)>, @@ -406,7 +376,7 @@ fn build_args( // not the default source (which is the mic). args.extend([ "pulsesrc".into(), - audio_device.to_string(), + audio_plan.gst_device_arg(), "do-timestamp=true".into(), "!".into(), "queue".into(), @@ -428,35 +398,66 @@ fn build_args( args } -async fn default_audio_monitor() -> Result { - let output = Command::new("pactl") - .arg("get-default-sink") - .output() - .await - .context( - "failed to run `pactl get-default-sink` (install pulseaudio-utils or pipewire-pulse)", - )?; - if !output.status.success() { - bail!( - "pactl get-default-sink failed: {}", - String::from_utf8_lossy(&output.stderr).trim() - ); - } - let sink = String::from_utf8(output.stdout) - .context("default sink name was not UTF-8")? - .trim() - .to_string(); - if sink.is_empty() { - bail!("pactl get-default-sink returned no name (is a sound server running?)"); - } - Ok(format!("{sink}.monitor")) -} - #[cfg(test)] mod tests { use super::*; + use crate::cli::{CaptureMode, Quality}; use std::time::{Duration, Instant}; + fn legacy_opts() -> HostOpts { + HostOpts { + window: false, + app: None, + strict_audio: false, + display_server: None, + quality: Quality::Source, + bitrate: None, + framerate: None, + max_height: None, + no_hwencode: false, + max_viewers: None, + interactive: false, + capture_mode: CaptureMode::Legacy, + legacy_null_sink: false, + relay: None, + } + } + + #[test] + fn legacy_desktop_audio_tail_is_byte_identical() { + let opts = legacy_opts(); + let quality = super::super::quality::resolve(&opts, 1); + let plan = CapturePlan::legacy_fixture("alsa_output.fixture.monitor"); + let args = build_args(&["ximagesrc".to_string()], &plan, &opts, &quality, None); + let audio_start = args + .iter() + .position(|arg| arg == "pulsesrc") + .expect("pipeline has an audio branch"); + assert_eq!( + &args[audio_start..], + [ + "pulsesrc", + "device=alsa_output.fixture.monitor", + "do-timestamp=true", + "!", + "queue", + "!", + "audioconvert", + "!", + "audioresample", + "!", + "audio/x-raw,rate=48000,channels=2", + "!", + "avenc_aac", + "bitrate=128000", + "!", + "aacparse", + "!", + "mux.", + ] + ); + } + fn process_is_running(pid: u32) -> bool { let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else { return false; diff --git a/src/host/quality.rs b/src/host/quality.rs index a744d7f..d360c43 100644 --- a/src/host/quality.rs +++ b/src/host/quality.rs @@ -214,6 +214,8 @@ mod tests { no_hwencode: false, max_viewers, interactive: false, + capture_mode: crate::cli::CaptureMode::Legacy, + legacy_null_sink: false, relay: None, } } diff --git a/src/host/taint/fixture.rs b/src/host/taint/fixture.rs index d0c2647..7f0ed71 100644 --- a/src/host/taint/fixture.rs +++ b/src/host/taint/fixture.rs @@ -155,7 +155,17 @@ impl Graph { /// A device node as the session manager creates it: no strong key, /// WirePlumber's client and PID — shared with every other device — and /// a `device.id`, which is what marks it as session-manager-exported. + /// Each call models a distinct physical device; use [`Self::device_node_on`] + /// when two terminals belong to the same card. pub fn device_node(&mut self, name: &str, role: MediaRole) -> NodeRef { + let device_id = self.id(); + self.device_node_on(name, role, device_id) + } + + /// A passive terminal exported by a particular physical Device. Sink + /// and source nodes given the same id model the hidden playback-to-capture + /// path that an ALSA/USB device may expose outside PipeWire's Link graph. + pub fn device_node_on(&mut self, name: &str, role: MediaRole, device_id: GlobalId) -> NodeRef { let session = match self.session_client { Some(id) => id, None => { @@ -164,7 +174,7 @@ impl Graph { id } }; - self.node(name, role, device(session, SESSION_PID)) + self.node(name, role, device(session, SESSION_PID, device_id)) } /// A node that *belongs to* a Device but is not a passive device node — @@ -247,6 +257,18 @@ impl Graph { } pub fn port(&mut self, node: NodeRef, direction: PortDirection, exclusive: bool) { + self.port_on_channel(node, direction, exclusive, None); + } + + /// A port with the channel identity Phase 6 uses for deterministic link + /// pairing. Returns its snapshot-local id for exact plan assertions. + pub fn port_on_channel( + &mut self, + node: NodeRef, + direction: PortDirection, + exclusive: bool, + channel: Option<&str>, + ) -> GlobalId { let serial = self.serial(); let id = self.id(); self.ports.push(PortSnapshot { @@ -254,9 +276,11 @@ impl Graph { id, node: node.id, direction, + channel: channel.map(str::to_string), exclusive, monitor: false, }); + id } /// A signal edge: audio flows `from → to`. @@ -386,10 +410,11 @@ pub fn link_group(group: &str, client: GlobalId, pid: u32) -> NodeProps { /// here is deliberately *more* pessimistic than reality — it hands the /// engine a second coarse key it could fuse devices on, so a test that /// passes here also passes against the real props. -pub fn device(session_client: GlobalId, session_pid: u32) -> NodeProps { +pub fn device(session_client: GlobalId, session_pid: u32, device_id: GlobalId) -> NodeProps { NodeProps { client_id: Some(session_client), process_id: Some(session_pid), + device_id: Some(device_id), session_device: true, ..NodeProps::default() } diff --git a/src/host/taint/mod.rs b/src/host/taint/mod.rs index a4a79d4..45dee3c 100644 --- a/src/host/taint/mod.rs +++ b/src/host/taint/mod.rs @@ -34,7 +34,12 @@ //! *is* a real Link whose output node is the sink node itself (measured). //! A port-granular walk would need a synthetic edge; a node-granular one //! does not. -//! 3. **Owner bridges** — the intra-process hop the graph cannot see. See +//! 3. **Hardware-device bridges** — a passive sink can feed a passive source +//! on the same physical Device through a mixer/loopback path that PipeWire +//! does not expose as a Link. The observer positively classifies both +//! terminals and retains their shared `device.id`; the walk conservatively +//! adds `sink → source` for that one Device. +//! 4. **Owner bridges** — the intra-process hop the graph cannot see. See //! [`owner`]; this is the hard one. //! //! ## Stickiness @@ -784,6 +789,40 @@ fn downstream_edges(snapshot: &GraphSnapshot, unresolved_input: &mut BTreeSet {} } } + + // A physical sound device may route playback back into capture in its + // own mixer/firmware without publishing a PipeWire Link (HDA "Stereo + // Mix", USB loopback channels, vendor DSPs). Control-name inspection is + // neither portable nor proof of absence, so v1 fails closed: once a + // passive hardware sink is tainted, passive capture terminals exported + // by the same Device are downstream too. + // + // Both guards are load-bearing. `session_device` limits this to the + // observer's positive hardware-terminal allowlist, so an app-associated + // filter cannot invent a bridge. `device_id` limits it to one physical + // Device, so the shared WirePlumber client does not fuse every card. + let hardware_outputs: Vec<(Serial, snapshot::GlobalId)> = snapshot + .nodes() + .filter(|node| { + node.props.session_device && matches!(node.role, MediaRole::Sink | MediaRole::Duplex) + }) + .filter_map(|node| node.props.device_id.map(|id| (node.serial, id))) + .collect(); + let hardware_inputs: Vec<(Serial, snapshot::GlobalId)> = snapshot + .nodes() + .filter(|node| { + node.props.session_device && matches!(node.role, MediaRole::Source | MediaRole::Duplex) + }) + .filter_map(|node| node.props.device_id.map(|id| (node.serial, id))) + .collect(); + for (from, output_device) in hardware_outputs { + for &(to, input_device) in &hardware_inputs { + if from != to && output_device == input_device { + edges.entry(from).or_default().push(to); + receivers.insert(to); + } + } + } for targets in edges.values_mut() { targets.sort_unstable(); targets.dedup(); diff --git a/src/host/taint/snapshot.rs b/src/host/taint/snapshot.rs index 61d75ad..8709ba8 100644 --- a/src/host/taint/snapshot.rs +++ b/src/host/taint/snapshot.rs @@ -139,6 +139,15 @@ pub struct NodeProps { /// The stream negotiated an encoded/passthrough format; a second link /// would refuse or corrupt it (v3.4 §6.2). pub passthrough: bool, + /// `device.id` — the snapshot-local PipeWire Device this node belongs + /// to. This is retained separately from [`Self::session_device`]: the + /// latter says the node is a positively-classified passive hardware + /// terminal, while this id lets the taint walk relate the playback and + /// capture terminals exported by that *same* device. + /// + /// Like every [`GlobalId`], this is valid only within this snapshot. It + /// must never enter sticky identity or survive a recompute. + pub device_id: Option, /// This node is a **passive device node exported by the session /// manager** — a real sound card's sink or source, not something that /// forwards audio. @@ -219,6 +228,9 @@ pub struct PortSnapshot { /// Owning node, by snapshot-local id. pub node: GlobalId, pub direction: PortDirection, + /// `audio.channel` (for example `FL`, `FR`, `MONO`). Phase 6 pairs + /// ports by channel, never by global-id or enumeration order. + pub channel: Option, /// `port.exclusive` — fan-out will be refused (v3.4 §6.2). pub exclusive: bool, /// `port.monitor`. Recorded for phase 6 link creation; taint does not diff --git a/src/host/taint/tests.rs b/src/host/taint/tests.rs index afc854e..d7999da 100644 --- a/src/host/taint/tests.rs +++ b/src/host/taint/tests.rs @@ -16,7 +16,7 @@ use std::collections::BTreeSet; use super::fixture::{Graph, NodeRef, PULSE_PID, app}; use super::owner::{OwnerCtx, OwnerKey, strongest_shared_key}; -use super::snapshot::{MediaRole, NodeProps, PortDirection, Serial}; +use super::snapshot::{GlobalId, MediaRole, NodeProps, PortDirection, Serial}; use super::{Decisions, Eligibility, ExclusionCtx, ObjectRef, Reason, StickyState, evaluate}; fn ctx() -> ExclusionCtx { @@ -176,6 +176,88 @@ fn peerspeak_tagged_nodes_are_excluded_and_plain_apps_are_not() { assert_tainted(&decisions, sink, "tainted-upstream"); } +// ────────────────────────────────────────────────────────────────────── +// Hidden hardware playback-to-capture paths — same Device only +// ───────────────────────────────────────────────────────────────────── + +#[test] +fn same_hardware_device_closes_an_unpublished_playback_to_capture_hop() { + let mut graph = Graph::new(); + let card = GlobalId(700); + let sink = graph.device_node_on("card-playback", MediaRole::Sink, card); + let source = graph.device_node_on("card-capture", MediaRole::Source, card); + let call = graph.peerspeak_node("peerspeak-call", 7); + let music = graph.app_node("music", MediaRole::StreamOutput, 8); + let recorder_in = graph.app_node("recorder-in", MediaRole::StreamInput, 9); + let recorder_out = graph.app_node("recorder-out", MediaRole::StreamOutput, 9); + + graph.link(call, sink); + graph.link(music, sink); + // There is deliberately no sink → source Link: the hardware bridge is + // the route being modeled. + graph.link(source, recorder_in); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[("music", music)], + &[ + ("call", call, "peerspeak-owned"), + ("recorder-out", recorder_out, "tainted-owner-bridge"), + ], + ); + assert_tainted(&decisions, sink, "tainted-upstream"); + assert_tainted(&decisions, source, "tainted-upstream"); + assert_tainted(&decisions, recorder_in, "tainted-upstream"); +} + +#[test] +fn different_hardware_devices_do_not_invent_a_capture_path() { + let mut graph = Graph::new(); + let sink = graph.device_node_on("speaker", MediaRole::Sink, GlobalId(700)); + let source = graph.device_node_on("usb-mic", MediaRole::Source, GlobalId(701)); + let call = graph.peerspeak_node("peerspeak-call", 7); + let recorder_in = graph.app_node("recorder-in", MediaRole::StreamInput, 9); + let recorder_out = graph.app_node("recorder-out", MediaRole::StreamOutput, 9); + + graph.link(call, sink); + graph.link(source, recorder_in); + + let decisions = run(&graph, &ctx()); + assert_partition( + &decisions, + &[("recorder-out", recorder_out)], + &[("call", call, "peerspeak-owned")], + ); + assert_untainted(&decisions, source); + assert_untainted(&decisions, recorder_in); +} + +#[test] +fn same_device_microphone_use_is_intentionally_over_excluded() { + // The hardware's private mixer/firmware path is not observable in the + // PipeWire graph. If an app captures the same device receiving the call, + // v1 cannot prove that its capture is clean, so its playback is excluded. + let mut graph = Graph::new(); + let card = GlobalId(700); + let sink = graph.device_node_on("headset-output", MediaRole::Sink, card); + let mic = graph.device_node_on("headset-mic", MediaRole::Source, card); + let call = graph.peerspeak_node("peerspeak-call", 7); + let firefox_in = graph.app_node("firefox-mic", MediaRole::StreamInput, 11_114); + let firefox_out = graph.app_node("firefox-audio", MediaRole::StreamOutput, 11_114); + graph.link(call, sink); + graph.link(mic, firefox_in); + + assert_partition( + &run(&graph, &ctx()), + &[], + &[ + ("call", call, "peerspeak-owned"), + ("firefox-out", firefox_out, "tainted-owner-bridge"), + ], + ); +} + /// Each ownership carrier must work **alone** (v3.5 §5.1). /// /// ⚠️ The phase-3r lesson, applied deliberately: a gate that asserts a value diff --git a/src/interactive.rs b/src/interactive.rs index 72f7d15..a2d7afe 100644 --- a/src/interactive.rs +++ b/src/interactive.rs @@ -30,7 +30,7 @@ pub async fn run(cli: Cli) -> Result<()> { if cli.quality.is_none() { cli.quality = Some(pick_quality(&theme)?); } - host::run(cli.into_host_opts(true)).await + host::run(cli.into_host_opts(true)?).await } _ => { let ticket = prompt_ticket(&theme)?; diff --git a/src/main.rs b/src/main.rs index 1b9ac65..01e8d99 100644 --- a/src/main.rs +++ b/src/main.rs @@ -70,7 +70,7 @@ async fn main() -> Result<()> { screen, a ticket views someone else's." ); } - return host::run(cli.into_host_opts(false)).await; + return host::run(cli.into_host_opts(false)?).await; } match cli.ticket.as_deref() {