host/audio: emit initial app_audio "lost" at strict capture start (A23 P2/F1)

In strict per-app mode the default-sink loopback is suppressed, so until the
chosen app's first stream routes the viewer hears silence. Previously no event
fired for an app that never routed (`lost` only fires on an N→0 transition
after a prior route), so peerspeak couldn't warn — the share looked normal but
was silent. Emit a `lost` at capture start (lazy, on first viewer) when, and
only when, `--app` + `--strict-audio` are both set; whole-desktop and
best-effort modes keep audio flowing via the loopback and emit nothing.

Factored the emit decision into the pure, unit-tested `initial_app_audio_state`;
derive Debug/PartialEq/Eq on AppAudioState so it can be asserted on.

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

Cargo.toml

@@ -6,6 +6,20 @@ description = "P2P screen sharing CLI over iroh"
 license = "MIT OR Apache-2.0"
 publish = false
 
+# Debian/Ubuntu packaging (cargo-deb). Headless default build (no `gui` feature) —
+# that is exactly what peerspeak spawns as a child. Runtime shared-lib deps
+# (libpipewire, libc, …) are resolved by dpkg-shlibdeps via `depends = "$auto"`.
+# Build inside a Debian/Ubuntu distrobox, then `cargo deb --no-build`.
+[package.metadata.deb]
+maintainer = "mollusk <jitty+lc1iz0dc@protonmail.com>"
+section = "net"
+priority = "optional"
+depends = "$auto"
+extended-description = "Peer-to-peer screen sharing over iroh (QUIC). Companion to peerspeak: shares a window or screen directly to a peer with no central server, driven via the CLI and its JSON event stream."
+assets = [
+    ["target/release/pixelpass", "usr/bin/", "755"],
+]
+
 [[bin]]
 name = "pixelpass"
 path = "src/main.rs"

src/cli.rs

@@ -27,6 +27,17 @@ pub struct Cli {
     #[arg(long, value_name = "NAME")]
     pub app: Option<String>,
 
+    /// With `--app`, never fall back to whole-desktop audio. By default an
+    /// app-filtered host mirrors the default sink's monitor until (and again
+    /// after) the chosen app's streams route, so the viewer isn't left in
+    /// silence. That fallback also captures everything else playing — including
+    /// a voice call the sharer is in — so a caller can hear themselves echoed.
+    /// `--strict-audio` suppresses the fallback entirely: the viewer hears only
+    /// the chosen app, and silence when it isn't producing audio. Ignored
+    /// without `--app`.
+    #[arg(long)]
+    pub strict_audio: bool,
+
     /// Override display server autodetection.
     #[arg(long, value_enum)]
     pub display_server: Option<DisplayServerArg>,
@@ -135,6 +146,10 @@ pub enum Quality {
 pub struct HostOpts {
     pub window: bool,
     pub app: Option<String>,
+    /// With `app` set, suppress the whole-desktop loopback fallback so the
+    /// viewer only ever hears the chosen app (silence when it's quiet). No
+    /// effect when `app` is None.
+    pub strict_audio: bool,
     pub display_server: Option<DisplayServerArg>,
     /// Chosen preset (Auto = derive at startup). Defaults to Auto.
     pub quality: Quality,
@@ -164,6 +179,7 @@ impl Cli {
         HostOpts {
             window: self.window,
             app: self.app,
+            strict_audio: self.strict_audio,
             display_server: self.display_server,
             // No `--quality` and nothing picked interactively → the documented
             // default, Auto.

src/common/control.rs

@@ -166,13 +166,16 @@ async fn handle(incoming: Incoming, tx: &mpsc::Sender<Inbound>) -> Result<()> {
         .await
         .context("timed out reading control message")??;
 
-    // Wait (briefly) for the sender's close so our ACK flushes before the
-    // connection is dropped at the end of this scope.
-    let _ = tokio::time::timeout(IO_TIMEOUT, conn.closed()).await;
-
+    // Hand the message up first, so it reaches the UI promptly even when the
+    // sender is slow to close (a degraded link could otherwise delay a friend
+    // request / pushed code by up to IO_TIMEOUT).
     tx.send(Inbound { from, msg })
         .await
         .map_err(|_| anyhow::anyhow!("control: receiver dropped"))?;
+
+    // Then wait (briefly) for the sender's close so our ACK has flushed before
+    // the connection is dropped at the end of this scope.
+    let _ = tokio::time::timeout(IO_TIMEOUT, conn.closed()).await;
     Ok(())
 }
 

src/common/friends.rs

@@ -147,35 +147,45 @@ impl FriendStore {
         self.friends.len() != before
     }
 
-    /// Apply an inbound friend request. Returns `true` if it *completes a mutual
-    /// match* — we'd already sent them one, so they're now [`Accepted`] and the
-    /// caller should reply with a `FriendAccept`. Otherwise it's recorded as
+    /// Apply an inbound friend request. Returns `true` if the friendship is now
+    /// settled at [`Accepted`] and the caller should reply with a `FriendAccept`
+    /// — either because we'd already sent them a request (a mutual match) or
+    /// because they're an existing friend re-announcing (we never downgrade an
+    /// [`Accepted`] friend back to pending; a peer who lost their store and
+    /// re-adds us just gets re-confirmed). Otherwise it's recorded as
     /// [`PendingIncoming`] for the user to act on and `false` is returned.
     ///
     /// [`Accepted`]: FriendState::Accepted
     /// [`PendingIncoming`]: FriendState::PendingIncoming
     pub fn on_friend_request(&mut self, id: EndpointId, name: String) -> bool {
+        match self.find(&id).map(|f| f.state) {
+            Some(FriendState::PendingOutgoing | FriendState::Accepted) => {
+                self.upsert(id, name, FriendState::Accepted);
+                true
+            }
+            _ => {
+                self.upsert(id, name, FriendState::PendingIncoming);
+                false
+            }
+        }
+    }
+
+    /// Apply an inbound acceptance of a request we sent. Returns `true` only if
+    /// it advanced one of *our* outgoing requests to [`Accepted`]. An accept for
+    /// any other state is ignored: a stranger's, or one for a peer still in
+    /// [`PendingIncoming`] (their request, awaiting our decision) — honouring the
+    /// latter would let a peer mark itself accepted without the local user's
+    /// consent.
+    ///
+    /// [`Accepted`]: FriendState::Accepted
+    /// [`PendingIncoming`]: FriendState::PendingIncoming
+    pub fn on_friend_accept(&mut self, id: EndpointId, name: String) -> bool {
         if matches!(
             self.find(&id).map(|f| f.state),
             Some(FriendState::PendingOutgoing)
         ) {
             self.upsert(id, name, FriendState::Accepted);
             true
-        } else {
-            self.upsert(id, name, FriendState::PendingIncoming);
-            false
-        }
-    }
-
-    /// Apply an inbound acceptance of a request we sent. Returns `true` if it
-    /// advanced a friendship to [`Accepted`] (i.e. we actually knew this peer);
-    /// an accept from a stranger is ignored.
-    ///
-    /// [`Accepted`]: FriendState::Accepted
-    pub fn on_friend_accept(&mut self, id: EndpointId, name: String) -> bool {
-        if self.find(&id).is_some() {
-            self.upsert(id, name, FriendState::Accepted);
-            true
         } else {
             false
         }
@@ -293,4 +303,29 @@ mod tests {
         assert!(!store.on_friend_accept(stranger, "Nope".into()));
         assert!(store.find(&stranger).is_none());
     }
+
+    #[test]
+    fn accept_does_not_advance_a_pending_incoming_peer() {
+        // They asked us and we haven't decided yet; an unsolicited FriendAccept
+        // from them must not auto-accept on our behalf (consent bypass).
+        let mut store = FriendStore::default();
+        let id = sample_id();
+        store.upsert(id, "Theirs".into(), FriendState::PendingIncoming);
+        assert!(!store.on_friend_accept(id, "Theirs".into()));
+        assert_eq!(store.find(&id).unwrap().state, FriendState::PendingIncoming);
+    }
+
+    #[test]
+    fn request_does_not_downgrade_an_accepted_friend() {
+        // A current friend re-sending a request (e.g. after losing their store)
+        // must stay accepted; the call signals a re-confirm rather than a
+        // downgrade to pending.
+        let mut store = FriendStore::default();
+        let id = sample_id();
+        store.upsert(id, "Pal".into(), FriendState::Accepted);
+        let settled = store.on_friend_request(id, "Pal (reinstalled)".into());
+        assert!(settled);
+        assert_eq!(store.find(&id).unwrap().state, FriendState::Accepted);
+        assert_eq!(store.find(&id).unwrap().name, "Pal (reinstalled)");
+    }
 }

src/common/output.rs

@@ -58,6 +58,11 @@ pub enum Event<'a> {
     ViewerRefused { reason: &'a str },
     /// Viewer-side: the local player URL is ready to open.
     Connected { url: &'a str },
+    /// Per-app audio routing state (only emitted when `--app` is set). `routed`
+    /// = the chosen app's audio is now reaching viewers; `lost` = its last
+    /// stream went away. Under `--strict-audio`, `lost` means viewers currently
+    /// hear silence; without it, viewers fall back to whole-desktop audio.
+    AppAudio { state: AppAudioState },
 }
 
 #[derive(Serialize)]
@@ -67,6 +72,13 @@ pub enum CaptureState {
     Stopped,
 }
 
+#[derive(Serialize, Clone, Copy, Debug, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum AppAudioState {
+    Routed,
+    Lost,
+}
+
 /// Emit one event as a JSON line on stdout, flushed. No-op unless JSON
 /// output was enabled with [`set_json`], so call sites can sprinkle these
 /// unconditionally without branching.
@@ -85,3 +97,25 @@ pub fn emit(event: Event) {
         Err(e) => tracing::warn!("failed to serialize event: {e}"),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // The app_audio event is the wire contract peerspeak parses to drive its
+    // echo warning; pin the exact shape so a rename here is caught here.
+    #[test]
+    fn app_audio_event_wire_shape() {
+        let routed = serde_json::to_string(&Event::AppAudio {
+            state: AppAudioState::Routed,
+        })
+        .unwrap();
+        assert_eq!(routed, r#"{"event":"app_audio","state":"routed"}"#);
+
+        let lost = serde_json::to_string(&Event::AppAudio {
+            state: AppAudioState::Lost,
+        })
+        .unwrap();
+        assert_eq!(lost, r#"{"event":"app_audio","state":"lost"}"#);
+    }
+}

src/common/process.rs

@@ -6,10 +6,19 @@ use std::process::{Command, Stdio};
 ///
 /// The child gets its own session via `setsid(2)` and null stdio, so it
 /// survives the parent exiting and doesn't take a SIGKILL cascade when
-/// pixelpass dies. The `Child` is dropped immediately — `std::process::Child::drop`
-/// does not kill the process on Unix.
+/// pixelpass dies.
+///
+/// A detached reaper thread `wait()`s the child so it doesn't linger as a
+/// `<defunct>` zombie under a long-lived parent — the `--gui` front-end launches
+/// players itself and lives for the whole session, and `std::process::Child`
+/// (unlike tokio's) has no orphan reaping, so simply dropping the handle would
+/// leak a zombie per closed player. If the parent exits while the player is
+/// still up, the reaper thread dies with it but the `setsid`'d player survives
+/// and is reaped by init. (A double-fork would also avoid the zombie, but
+/// `fork(2)` followed by non-trivial work in this multithreaded process is
+/// unsound — the reaper thread is the safe equivalent.)
 pub fn spawn_detached(prog: &str, args: &[&str]) -> io::Result<()> {
-    unsafe {
+    let child = unsafe {
         Command::new(prog)
             .args(args)
             .stdin(Stdio::null())
@@ -19,7 +28,11 @@ pub fn spawn_detached(prog: &str, args: &[&str]) -> io::Result<()> {
                 nix::unistd::setsid().ok();
                 Ok(())
             })
-            .spawn()?;
-    }
+            .spawn()?
+    };
+    std::thread::spawn(move || {
+        let mut child = child;
+        let _ = child.wait();
+    });
     Ok(())
 }

src/gui/mod.rs

@@ -1145,11 +1145,14 @@ impl PixelPassApp {
                             f.name = name.clone();
                             store_changed = true;
                         }
-                        self.push_notice(from, name.clone(), ticket);
-                        notify(
-                            "PixelPass — a friend is sharing",
-                            format!("{name} is sharing their screen. Open PixelPass to watch."),
-                        );
+                        // Only toast for a new/changed code — an ACK-loss retry
+                        // redelivers the same code and shouldn't fire again.
+                        if self.push_notice(from, name.clone(), ticket) {
+                            notify(
+                                "PixelPass — a friend is sharing",
+                                format!("{name} is sharing their screen. Open PixelPass to watch."),
+                            );
+                        }
                     } else {
                         tracing::warn!(from = %from, "presence: ignoring ShareCode from a non-friend");
                     }
@@ -1197,13 +1200,20 @@ impl PixelPassApp {
     }
 
     /// Record a share code a friend pushed us, replacing any prior notice from
-    /// the same friend (their previous code is stale once they re-host).
-    fn push_notice(&mut self, from: iroh::EndpointId, name: String, code: String) {
+    /// the same friend (their previous code is stale once they re-host). Returns
+    /// `true` if this is a new notice or a *different* code than we already had
+    /// from them — i.e. worth a fresh desktop notification. A duplicate delivery
+    /// (an ACK-loss retry redelivering the same code) updates in place and
+    /// returns `false`, so it doesn't fire a second toast.
+    fn push_notice(&mut self, from: iroh::EndpointId, name: String, code: String) -> bool {
         if let Some(n) = self.notices.iter_mut().find(|n| n.from == from) {
+            let changed = n.code != code;
             n.name = name;
             n.code = code;
+            changed
         } else {
             self.notices.push(ShareNotice { from, name, code });
+            true
         }
     }
 
@@ -2318,19 +2328,29 @@ impl PixelPassApp {
             self.apply_host_event(ev);
         }
 
-        if let Some(p) = &mut self.host.proc
-            && !p.is_alive()
-        {
-            if self.host.ticket.is_none() {
-                let tail = p.stderr_tail();
-                self.host.error = Some(if tail.trim().is_empty() {
+        let dead = self.host.proc.as_mut().is_some_and(|p| !p.is_alive());
+        if dead {
+            // If it never reached a ticket, capture why (from the stderr tail)
+            // before tearing down. Then run the *full* Stop cleanup — most
+            // importantly stop_share, so a host that died on its own stops
+            // pushing its now-dead code to friends. Without this the campaign
+            // would keep retrying offline friends with a stale ticket for the
+            // life of the GUI, and share_status/met/share_code would leak.
+            let error = self.host.ticket.is_none().then(|| {
+                let tail = self
+                    .host
+                    .proc
+                    .as_mut()
+                    .map(|p| p.stderr_tail())
+                    .unwrap_or_default();
+                if tail.trim().is_empty() {
                     "Host exited before it could start.".to_string()
                 } else {
                     format!("Host exited before it could start:\n{tail}")
-                });
-            }
-            self.host.proc = None;
-            self.host.capturing = false;
+                }
+            });
+            self.stop_host();
+            self.host.error = error;
         }
     }
 

src/host/audio.rs

@@ -55,24 +55,38 @@ impl Routing {
         let sink_module = load_module(&["module-null-sink", &format!("sink_name={sink_name}")])
             .context("failed to load module-null-sink")?;
 
+        // 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
+        // leak e.g. a voice call the sharer is in back to viewers — the echo
+        // bug A23). Without strict mode (whole-desktop share, or best-effort
+        // app filtering) we load the monitor loopback so the viewer hears
+        // system audio immediately and during any gap before the app routes.
         // 20ms loopback latency keeps the mirrored audio tight; pactl's
         // default of 200ms is enough to be perceptible.
-        let loopback_module = load_module(&[
-            "module-loopback",
-            "source=@DEFAULT_SINK@.monitor",
-            &format!("sink={sink_name}"),
-            "latency_msec=20",
-        ])
-        .context("failed to load module-loopback (null-sink will be cleaned up on Drop)")?;
+        let strict_app = opts.app.is_some() && opts.strict_audio;
+        let loopback_module = if strict_app {
+            None
+        } else {
+            Some(
+                load_module(&[
+                    "module-loopback",
+                    "source=@DEFAULT_SINK@.monitor",
+                    &format!("sink={sink_name}"),
+                    "latency_msec=20",
+                ])
+                .context("failed to load module-loopback (null-sink cleaned up on Drop)")?,
+            )
+        };
 
         tracing::info!(
             sink_module,
-            loopback_module,
+            ?loopback_module,
+            strict_app,
             %sink_name,
-            "audio routing: null-sink + loopback ready"
+            "audio routing: null-sink ready (loopback skipped in strict app mode)"
         );
 
-        let loopback_arc = Arc::new(Mutex::new(Some(loopback_module)));
+        let loopback_arc = Arc::new(Mutex::new(loopback_module));
         let mut routing = Self {
             sink_module: Some(sink_module),
             loopback_module: Arc::clone(&loopback_arc),
@@ -85,7 +99,9 @@ impl Routing {
             let (router, mut event_rx) = StreamRouter::spawn(app.clone(), sink_name.clone())?;
             let loopback_for_task = Arc::clone(&loopback_arc);
             let sink_name_for_task = sink_name.clone();
+            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 => {
@@ -96,11 +112,30 @@ impl Routing {
                                 );
                                 unload_module(id);
                             }
+                            // Tell the front-end the chosen app's audio is live.
+                            output::emit(output::Event::AppAudio {
+                                state: AppAudioState::Routed,
+                            });
                         }
                         Event::LastRoutedStreamGone => {
-                            // Routed app exited mid-session. Restore the
-                            // default-sink loopback so the viewer hears
-                            // system audio again instead of silence.
+                            // Routed app exited/paused mid-session. Notify the
+                            // front-end either way; the recovery differs by mode.
+                            output::emit(output::Event::AppAudio {
+                                state: AppAudioState::Lost,
+                            });
+                            if strict {
+                                // Strict mode: do NOT restore the whole-desktop
+                                // loopback. Viewers hear silence until the app
+                                // produces audio again — never the rest of the
+                                // desktop (call included).
+                                tracing::info!(
+                                    "audio routing: strict mode — last routed stream gone, leaving viewers silent"
+                                );
+                                continue;
+                            }
+                            // Best-effort mode: restore the default-sink loopback
+                            // so the viewer hears system audio again instead of
+                            // silence.
                             if loopback_for_task.lock().unwrap().is_some() {
                                 continue;
                             }
@@ -130,6 +165,16 @@ impl Routing {
             routing.event_task = Some(event_task);
         }
 
+        // Strict per-app mode suppresses the default-sink loopback, so until the
+        // chosen app's first stream routes the viewer hears *silence*. Emit an
+        // initial `lost` at capture start (capture is lazy — this runs on the
+        // first viewer) so the front-end can warn from the outset rather than
+        // only after an app that *was* routed later stops (audit A23 P2/F1):
+        // `LastRoutedStreamGone`→`lost` never fires for an app that never routed.
+        if let Some(state) = initial_app_audio_state(opts) {
+            crate::common::output::emit(crate::common::output::Event::AppAudio { state });
+        }
+
         Ok(routing)
     }
 
@@ -171,6 +216,16 @@ impl Drop for Routing {
     }
 }
 
+/// The app-audio state to announce at capture start, if any. Only strict per-app
+/// mode warrants one: there the loopback is suppressed, so the viewer hears
+/// silence until the chosen app's first stream routes — surface that as an
+/// initial `lost`. In every other mode (whole-desktop, or best-effort app
+/// filtering) the loopback keeps audio flowing from the outset, so there is no
+/// initial gap to report. Pure: no I/O, so the emit decision is unit-testable.
+pub(super) fn initial_app_audio_state(opts: &HostOpts) -> Option<crate::common::output::AppAudioState> {
+    (opts.app.is_some() && opts.strict_audio).then_some(crate::common::output::AppAudioState::Lost)
+}
+
 // ──────────────────────────────────────────────────────────────────────
 // App enumeration (interactive picker source)
 // ──────────────────────────────────────────────────────────────────────

src/host/mod.rs

@@ -488,9 +488,67 @@ 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 {
-        bits.push(format!("app-audio={app}"));
+        if opts.strict_audio {
+            bits.push(format!("app-audio={app} (strict)"));
+        } else {
+            bits.push(format!("app-audio={app}"));
+        }
     } else {
         bits.push("system-audio".to_string());
     }
     bits.join(" + ")
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::cli::Quality;
+
+    fn opts(app: Option<&str>, strict_audio: bool) -> HostOpts {
+        HostOpts {
+            window: false,
+            app: app.map(str::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,
+            relay: None,
+        }
+    }
+
+    #[test]
+    fn capture_summary_reflects_audio_mode() {
+        assert_eq!(capture_summary(&opts(None, false)), "fullscreen + system-audio");
+        assert_eq!(
+            capture_summary(&opts(Some("Firefox"), false)),
+            "fullscreen + app-audio=Firefox"
+        );
+        // strict only shows when an app is selected.
+        assert_eq!(
+            capture_summary(&opts(Some("Firefox"), true)),
+            "fullscreen + app-audio=Firefox (strict)"
+        );
+        assert_eq!(capture_summary(&opts(None, true)), "fullscreen + system-audio");
+    }
+
+    #[test]
+    fn initial_app_audio_is_lost_only_in_strict_app_mode() {
+        use crate::common::output::AppAudioState;
+        use crate::host::audio::initial_app_audio_state;
+        // Strict + app: announce silence up front (loopback suppressed).
+        assert_eq!(
+            initial_app_audio_state(&opts(Some("Firefox"), true)),
+            Some(AppAudioState::Lost)
+        );
+        // Best-effort app (no strict): loopback covers the gap → no initial event.
+        assert_eq!(initial_app_audio_state(&opts(Some("Firefox"), false)), None);
+        // Whole-desktop (strict is ignored without --app): no per-app events.
+        assert_eq!(initial_app_audio_state(&opts(None, true)), None);
+        assert_eq!(initial_app_audio_state(&opts(None, false)), None);
+    }
+}

src/host/quality.rs

@@ -205,6 +205,7 @@ mod tests {
         HostOpts {
             window: false,
             app: None,
+            strict_audio: false,
             display_server: None::<DisplayServerArg>,
             quality,
             bitrate: None,

src/host/wayland.rs

@@ -12,8 +12,7 @@ use ashpd::{
     },
 };
 use nix::fcntl::{FcntlArg, FdFlag, fcntl};
-use nix::unistd::close;
-use std::os::fd::{AsFd, IntoRawFd, OwnedFd, RawFd};
+use std::os::fd::{AsFd, AsRawFd, OwnedFd, RawFd};
 
 use super::pipeline::{self, CaptureHandle};
 use super::quality::EffectiveQuality;
@@ -61,11 +60,14 @@ pub async fn start(opts: &HostOpts, quality: &EffectiveQuality) -> Result<Captur
     let pw_fd: OwnedFd = proxy.open_pipe_wire_remote(&session).await?;
     tracing::info!(node_id, width = w, height = h, "portal handshake complete");
     // The fd is CLOEXEC by default; the gst child needs to inherit it across
-    // exec. We then leak it via into_raw_fd so its lifetime spans the spawn,
-    // and close the parent's copy once gst is running (the pipeline's
-    // after_spawn hook below).
+    // exec, so clear CLOEXEC. We keep the OwnedFd alive across the spawn (gst
+    // inherits its own copy at exec) by moving it into the after_spawn hook,
+    // which drops — and so closes — the parent's copy once gst is running. If
+    // pipeline::spawn errors *before* calling the hook (e.g. audio setup or the
+    // gst spawn fails), the unused closure is dropped, dropping the fd just the
+    // same — so the portal fd never leaks on the error path.
     clear_cloexec(&pw_fd)?;
-    let raw_fd: RawFd = pw_fd.into_raw_fd();
+    let raw_fd: RawFd = pw_fd.as_raw_fd();
 
     let source_args = vec![
         "pipewiresrc".to_string(),
@@ -81,7 +83,7 @@ pub async fn start(opts: &HostOpts, quality: &EffectiveQuality) -> Result<Captur
         source_args,
         move || {
             // Parent no longer needs the pipewire fd — gst inherited its own copy.
-            let _ = close(raw_fd);
+            drop(pw_fd);
         },
     )
     .await

src/viewer/mod.rs

@@ -71,7 +71,18 @@ pub async fn run(ticket: EndpointTicket, opts: ViewerOpts) -> Result<()> {
             accepted = listener.accept() => {
                 let (tcp, peer) = accepted?;
                 tracing::info!(%peer, "local viewer connected");
-                crate::common::tunnel::bridge(quic_send, quic_recv, tcp).await
+                // Race the bridge against ctrl-c so a disconnect lands promptly
+                // mid-stream (mirrors the host's handle_peer). Without this, the
+                // cancel token is set but nothing checks it once the player has
+                // connected — ctrl-c is ignored until a second press, and a GUI
+                // "Disconnect" only takes effect via the child's SIGKILL backstop.
+                tokio::select! {
+                    res = crate::common::tunnel::bridge(quic_send, quic_recv, tcp) => res,
+                    _ = cancel.cancelled() => {
+                        tracing::info!("ctrl-c received during stream — disconnecting");
+                        Ok(())
+                    }
+                }
             }
             _ = cancel.cancelled() => {
                 tracing::info!("ctrl-c received before local viewer connected");