feat(audio/linux): host-owned stream sink decouples capture from hardware-sink churn

The Linux desktop-audio capture stream now registers itself as an Audio/Sink
node ("Punktfunk Stream Speaker", default-on, PUNKTFUNK_STREAM_SINK=0 =
legacy escape hatch) and claims the configured default sink for the duration
of a session (saved and restored around it, refcounted across concurrent
sessions, crash-stale claims degrade to automatic election). Host apps play
directly into the capture stream, so the capture link no longer depends on
any hardware sink.

Root cause this fixes (live-diagnosed on a bazzite/LG-TV host): gamescope
modesets drop the NVIDIA HDMI audio endpoint, WirePlumber ping-pongs the
default sink HDMI<->auto_null ~8x/s, and the old monitor-follower relinked
its capture on every flip - Paused/renegotiate/Streaming storms (~1300 log
lines/min) heard as crackle on the client.

Also fixes a latent liveness bug in both modes: the capture thread had no
core-error listener, so a PipeWire daemon restart mid-session left a zombie
thread returning quiet-sink empty chunks forever (same class as the historic
virtual-mic death bug). Now the thread exits and sessions reopen with backoff.

Bonus: the sink advertises the session's true channel count, so games can
produce real 5.1/7.1 even when local hardware is stereo. New AudioCapturer::
idle() hook releases the routing claim when a capturer is parked between
sessions; drain() re-claims on reuse.

Verified: cargo clippy -D warnings + 295 tests green on Linux (.21).
On-glass validation on the bazzite host pending.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

crates/punktfunk-host/src/audio.rs

@@ -1,5 +1,7 @@
-//! Desktop audio capture for the GameStream audio stream. On Linux: a PipeWire stream that
+//! Desktop audio capture for the GameStream audio stream. On Linux: a PipeWire stream that by
-//! records the default sink's monitor (i.e. everything playing out of the system), delivered
+//! default registers a host-owned **stream sink** claimed as the session's default output
+//! (apps play into it directly — immune to hardware-sink churn; `PUNKTFUNK_STREAM_SINK=0`
+//! falls back to recording the default sink's monitor). Either way the capture is delivered
 //! as interleaved `f32` PCM at 48 kHz in the requested channel count (stereo, 5.1 or 7.1 —
 //! GameStream surround order FL FR FC LFE RL RR [SL SR]). The audio data plane
 //! (`gamestream::audio`) reframes this into fixed Opus frames, encodes, and sends it.
@@ -28,13 +30,24 @@ pub trait AudioCapturer: Send {
     }
 
     /// Discard any buffered chunks (called when a persistent capturer is reused for a new
-    /// stream, so the client doesn't hear stale audio captured while idle). Default: no-op.
+    /// stream, so the client doesn't hear stale audio captured while idle). On Linux this is
+    /// also the session-start hook: the stream-sink capturer re-claims the default sink here
+    /// (see [`idle`](Self::idle)). Default: no-op.
     fn drain(&mut self) {}
+
+    /// Called when a session parks the capturer back into the persistent slot: release
+    /// session-scoped routing side effects while keeping the capture backend alive. On Linux
+    /// the stream-sink capturer restores the user's default sink here, so host apps play to
+    /// the real output again between streams; the claim returns with the next
+    /// [`drain`](Self::drain) (reuse) or a fresh open. Default: no-op.
+    fn idle(&mut self) {}
 }
 
-/// Open a live capturer for the default sink monitor (system output) via PipeWire, asking
+/// Open a live capturer for system output via PipeWire, asking for `channels` interleaved
-/// for `channels` interleaved channels. If the sink has fewer channels than requested,
+/// channels. Default: a host-owned stream sink claimed as the default output (the sink
-/// PipeWire's channel-mixer fills the missing positions with silence (zero upmix).
+/// advertises exactly `channels`, so apps can produce real surround); with
+/// `PUNKTFUNK_STREAM_SINK=0`, the default sink's monitor, where a sink with fewer channels
+/// gets the missing positions filled with silence (zero upmix).
 #[cfg(target_os = "linux")]
 pub fn open_audio_capture(channels: u32) -> Result<Box<dyn AudioCapturer>> {
     linux::PwAudioCapturer::open(channels).map(|c| Box::new(c) as Box<dyn AudioCapturer>)

crates/punktfunk-host/src/audio/linux/mod.rs

@@ -1,17 +1,33 @@
-//! PipeWire audio capture of the default sink's monitor (system output).
+//! PipeWire desktop-audio capture — via a **host-owned stream sink** (default), or the legacy
+//! default-sink-monitor follower (`PUNKTFUNK_STREAM_SINK=0`).
 //!
-//! Connects to the user's PipeWire daemon (via `XDG_RUNTIME_DIR`, inherited from the Sway
+//! **Stream-sink mode.** The capture stream registers itself as an `Audio/Sink` node
-//! session) and opens an input stream with `stream.capture.sink=true`, which routes the
+//! ("Punktfunk Stream Speaker", unique `node.name` per capturer): host apps play *into* it,
-//! default sink's monitor into us — no portal needed (unlike screen capture). The (`!Send`)
+//! PipeWire mixes them, and our `process()` callback receives the mix directly — the same
-//! MainLoop/Stream live on a dedicated thread; interleaved `f32` chunks leave over a bounded
+//! stream-node architecture as [`PwMicSource`] below (inverted), and the documented
-//! channel (dropped if the encoder falls behind, never blocking the PipeWire loop).
+//! `pw-loopback --capture-props='media.class=Audio/Sink'` virtual-sink recipe. A session-scoped
+//! [`stream_sink`] claim makes it the *default* sink so apps route to it (and back) with the
+//! session. Why: capture no longer depends on any hardware sink, whose availability is display
+//! hardware state — live-diagnosed 2026-07-14 on a bazzite/TV host, every gamescope modeset
+//! dropped the HDMI audio endpoint, WirePlumber ping-ponged the default HDMI↔auto_null ~8×/s,
+//! and the old monitor-follower relinked on every flip (Paused→renegotiate→Streaming storms =
+//! client crackle). Bonus: the sink advertises the session's true channel count, so games can
+//! produce real 5.1/7.1 even when the local hardware is stereo.
 //!
-//! The stream is opened at the *session's* channel count (2/6/8). If the sink has fewer
+//! **Legacy mode** connects an input stream with `stream.capture.sink=true`, which routes the
-//! channels than requested, PipeWire's channel-mixer fills the extra positions with silence
+//! *default* sink's monitor into us — no portal needed (unlike screen capture), but coupled to
-//! (zero upmix), so a stereo desktop still produces a valid 5.1/7.1 capture. Dropping the
+//! hardware-default churn as above.
-//! capturer quits the loop thread (via a `pipewire::channel` Terminate message), tearing the
+//!
-//! stream down promptly — required so a surround session can replace a stereo capturer
+//! In both modes the (`!Send`) MainLoop/Stream live on a dedicated thread; interleaved `f32`
-//! without leaking a PipeWire consumer (see CLAUDE.md: a wedged link head-blocks the daemon).
+//! chunks leave over a bounded channel (dropped if the encoder falls behind, never blocking
+//! the PipeWire loop). The stream is opened at the *session's* channel count (2/6/8); in
+//! legacy mode PipeWire's channel-mixer fills missing positions with silence (zero upmix).
+//! Dropping the capturer quits the loop thread (via a `pipewire::channel` Terminate message),
+//! tearing the stream — and in stream-sink mode the sink node itself — down promptly, so a
+//! surround session can replace a stereo capturer without leaking a PipeWire consumer (see
+//! CLAUDE.md: a wedged link head-blocks the daemon).
+
+mod stream_sink;
 
 use super::{AudioCapturer, VirtualMic, SAMPLE_RATE};
 use anyhow::{anyhow, Context, Result};
@@ -25,10 +41,26 @@ use std::time::Duration;
 /// Message asking the PipeWire loop thread to quit (sent from `Drop`).
 struct Terminate;
 
+/// Whether the host-owned stream sink is active. **Default ON** — decouples capture (and app
+/// routing) from hardware-sink availability; see the module docs for the live-diagnosed
+/// crackle this fixes. `PUNKTFUNK_STREAM_SINK=0` (also `false`/`no`/`off`) is the escape hatch
+/// back to capturing the default sink's monitor.
+fn stream_sink_enabled() -> bool {
+    std::env::var("PUNKTFUNK_STREAM_SINK")
+        .map(|v| !matches!(v.trim(), "0" | "false" | "no" | "off"))
+        .unwrap_or(true)
+}
+
 pub struct PwAudioCapturer {
     chunks: Receiver<Vec<f32>>,
     channels: u32,
     quit: pipewire::channel::Sender<Terminate>,
+    /// `Some(node.name)` in stream-sink mode; `None` = legacy monitor follower.
+    sink_name: Option<String>,
+    /// Whether this capturer currently holds a [`stream_sink`] default-sink claim (session
+    /// active). Toggled by open/[`drain`](AudioCapturer::drain) (claim) and
+    /// [`idle`](AudioCapturer::idle)/Drop (release).
+    claimed: bool,
 }
 
 impl PwAudioCapturer {
@@ -37,26 +69,64 @@ impl PwAudioCapturer {
             matches!(channels, 1 | 2 | 6 | 8),
             "unsupported audio channel count {channels} (want 2, 6 or 8)"
         );
+        // Unique per capturer: overlapping instances (mid-session reopen, concurrent sessions)
+        // must never alias in metadata claims, and a fresh name gets fresh (unity) WirePlumber
+        // volume state instead of whatever a previous run left behind.
+        let sink_name = stream_sink_enabled().then(|| {
+            use std::sync::atomic::AtomicU64;
+            static SEQ: AtomicU64 = AtomicU64::new(0);
+            format!(
+                "{}-{}-{}",
+                stream_sink::SINK_NAME_PREFIX,
+                std::process::id(),
+                SEQ.fetch_add(1, Ordering::Relaxed)
+            )
+        });
         let (tx, rx) = sync_channel::<Vec<f32>>(64);
         let (quit_tx, quit_rx) = pipewire::channel::channel::<Terminate>();
+        // Bring-up handshake (mirrors the virtual mic): a PipeWire that isn't running must
+        // surface as an open ERROR — engaging the callers' reopen backoff — and in stream-sink
+        // mode the sink node must exist before we claim the default to its name.
+        let (ready_tx, ready_rx) = sync_channel::<Result<()>>(1);
+        let thread_sink_name = sink_name.clone();
         thread::Builder::new()
             .name("punktfunk-pw-audio".into())
             .spawn(move || {
-                if let Err(e) = pw_thread(tx, quit_rx, channels) {
+                if let Err(e) = pw_thread(tx, quit_rx, channels, thread_sink_name, ready_tx) {
                     tracing::error!(error = %format!("{e:#}"), "pipewire audio thread failed");
                 }
             })
             .context("spawn pipewire audio thread")?;
+        match ready_rx.recv_timeout(Duration::from_secs(5)) {
+            Ok(Ok(())) => {}
+            Ok(Err(e)) => return Err(e),
+            Err(_) => return Err(anyhow!("pipewire audio init timed out")),
+        }
+        // The capturer opens at session start, so the routing claim begins here; the paired
+        // release is `idle()` (parked between sessions) or Drop.
+        let claimed = match &sink_name {
+            Some(name) => {
+                stream_sink::claim(name);
+                true
+            }
+            None => false,
+        };
         Ok(PwAudioCapturer {
             chunks: rx,
             channels,
             quit: quit_tx,
+            sink_name,
+            claimed,
         })
     }
 }
 
 impl Drop for PwAudioCapturer {
     fn drop(&mut self) {
+        if self.claimed {
+            self.claimed = false;
+            stream_sink::release();
+        }
         // Ask the loop thread to quit; the stream/core/loop unwind there (RAII). A failed
         // send means the thread already exited — nothing to tear down.
         let _ = self.quit.send(Terminate);
@@ -80,6 +150,19 @@ impl AudioCapturer for PwAudioCapturer {
 
     fn drain(&mut self) {
         while self.chunks.try_recv().is_ok() {}
+        // A parked capturer being reused = a new session starting: re-claim the default sink
+        // (released by `idle()` when the previous session parked us).
+        if let (Some(name), false) = (&self.sink_name, self.claimed) {
+            stream_sink::claim(name);
+            self.claimed = true;
+        }
+    }
+
+    fn idle(&mut self) {
+        if self.claimed {
+            self.claimed = false;
+            stream_sink::release();
+        }
     }
 }
 
@@ -484,137 +567,205 @@ fn pw_thread(
     tx: std::sync::mpsc::SyncSender<Vec<f32>>,
     quit_rx: pipewire::channel::Receiver<Terminate>,
     channels: u32,
+    sink_name: Option<String>,
+    ready: std::sync::mpsc::SyncSender<Result<()>>,
 ) -> Result<()> {
     use pipewire as pw;
     use pw::{properties::properties, spa};
     use spa::param::audio::{AudioFormat, AudioInfoRaw};
     use spa::pod::Pod;
 
-    crate::pwinit::ensure_init();
+    // Setup errors funnel through the ready handshake (mirrors mic_pw_thread's IIFE).
-    let mainloop = pw::main_loop::MainLoopRc::new(None).context("pw audio MainLoop")?;
+    let result = (|| -> Result<()> {
-    let context = pw::context::ContextRc::new(&mainloop, None).context("pw audio Context")?;
+        crate::pwinit::ensure_init();
-    let core = context
+        let mainloop = pw::main_loop::MainLoopRc::new(None).context("pw audio MainLoop")?;
-        .connect_rc(None)
+        let context = pw::context::ContextRc::new(&mainloop, None).context("pw audio Context")?;
-        .context("pw audio connect (is PipeWire running in this session?)")?;
+        let core = context
+            .connect_rc(None)
+            .context("pw audio connect (is PipeWire running in this session?)")?;
 
-    // Cross-thread teardown: the capturer's Drop sends Terminate; quit the loop here.
+        // Cross-thread teardown: the capturer's Drop sends Terminate; quit the loop here.
-    let _quit_guard = quit_rx.attach(mainloop.loop_(), {
+        let _quit_guard = quit_rx.attach(mainloop.loop_(), {
-        let mainloop = mainloop.clone();
+            let mainloop = mainloop.clone();
-        move |_| mainloop.quit()
+            move |_| mainloop.quit()
-    });
+        });
 
-    let stream = pw::stream::StreamBox::new(
+        // Death detection (same contract as the virtual mic below): a core error — the daemon
-        &core,
+        // restarted/went away — ends this thread, so the chunk channel disconnects and
-        "punktfunk-audio",
+        // `next_chunk` returns Err, engaging the sessions' reopen-with-backoff. Without this, a
-        properties! {
+        // PipeWire restart mid-session left a zombie capture thread whose `next_chunk` returned
-            *pw::keys::MEDIA_TYPE          => "Audio",
+        // quiet-sink empty chunks forever — audio silently dead for the rest of the session.
-            *pw::keys::MEDIA_CATEGORY      => "Capture",
+        let _core_listener = core
-            *pw::keys::MEDIA_ROLE          => "Music",
+            .add_listener_local()
-            // Capture the default sink's monitor (system output), not a microphone.
+            .error({
-            *pw::keys::STREAM_CAPTURE_SINK => "true",
+                let mainloop = mainloop.clone();
-            // Ask for a ~5ms quantum (= one Opus frame) so buffers arrive smoothly rather than
+                move |id, _seq, res, message| {
-            // in large bursts the client's low-latency jitter buffer would hear as glitching.
+                    tracing::warn!(id, res, message, "pipewire core error — audio capture ends");
-            *pw::keys::NODE_LATENCY        => "240/48000",
+                    mainloop.quit();
-        },
+                }
-    )
+            })
-    .context("pw audio Stream")?;
+            .register();
 
-    let _listener = stream
+        let props = match &sink_name {
-        .add_local_listener_with_user_data(tx)
+            // Stream-sink mode: this stream IS the sink (media.class + Direction::Input). Apps
-        .state_changed(|_s, _ud, old, new| {
+            // play into it, PipeWire mixes them, process() receives the mix. Mirrors the
-            tracing::info!(?old, ?new, "pipewire audio stream state");
+            // validated PwMicSource recipe (stream node + RT_PROCESS; see its property
-        })
+            // comments) — do NOT "modernize" either into a `support.null-audio-sink` adapter
-        .param_changed(|_stream, _tx, id, param| {
+            // without re-running that validation.
-            let Some(param) = param else { return };
+            Some(name) => {
-            if id != pw::spa::param::ParamType::Format.as_raw() {
+                let mut p = properties! {
-                return;
+                    *pw::keys::MEDIA_TYPE       => "Audio",
-            }
+                    *pw::keys::MEDIA_CLASS      => "Audio/Sink",
-            let mut info = AudioInfoRaw::default();
+                    *pw::keys::NODE_DESCRIPTION => "Punktfunk Stream Speaker",
-            if info.parse(param).is_ok() {
+                    *pw::keys::NODE_VIRTUAL     => "true",
-                tracing::info!(
+                    // Ask for a ~5ms quantum (= one Opus frame) so buffers arrive smoothly
-                    format = ?info.format(),
+                    // rather than in bursts the client's jitter buffer would hear as glitching.
-                    rate = info.rate(),
+                    *pw::keys::NODE_LATENCY     => "240/48000",
-                    channels = info.channels(),
+                    // LOW priority — the opposite of the mic's 3000: between sessions the sink
-                    "audio format negotiated"
+                    // node stays alive (parked capturer) but must never win WirePlumber's auto
-                );
+                    // default election against real hardware; session routing comes from the
-            }
+                    // stream_sink claim, not from priority.
-        })
+                    "priority.session"          => "50",
-        .process(|stream, tx| {
-            let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
-                let Some(mut buffer) = stream.dequeue_buffer() else {
-                    return;
                 };
-                let datas = buffer.datas_mut();
+                p.insert(*pw::keys::NODE_NAME, name.as_str());
-                if datas.is_empty() {
+                p
-                    return;
-                }
-                let d = &mut datas[0];
-                let (offset, size) = {
-                    let c = d.chunk();
-                    (c.offset() as usize, c.size() as usize)
-                };
-                let Some(buf) = d.data() else { return };
-                if offset > buf.len() {
-                    return;
-                }
-                let region = &buf[offset..(offset + size).min(buf.len())];
-                // Negotiated as F32LE; reinterpret the byte region as interleaved f32.
-                let n = region.len() / 4;
-                static FIRST: std::sync::atomic::AtomicBool =
-                    std::sync::atomic::AtomicBool::new(true);
-                if FIRST.swap(false, std::sync::atomic::Ordering::Relaxed) {
-                    tracing::info!(samples = n, "audio first capture buffer");
-                }
-                let mut samples = Vec::with_capacity(n);
-                for i in 0..n {
-                    let b = [
-                        region[i * 4],
-                        region[i * 4 + 1],
-                        region[i * 4 + 2],
-                        region[i * 4 + 3],
-                    ];
-                    samples.push(f32::from_le_bytes(b));
-                }
-                let _ = tx.try_send(samples); // drop if the encoder is behind
-            }));
-            if outcome.is_err() {
-                tracing::error!("panic in pipewire audio callback — chunk dropped");
             }
-        })
+            // Legacy: capture the default sink's monitor (system output), not a microphone.
-        .register()
+            None => properties! {
-        .context("register audio stream listener")?;
+                *pw::keys::MEDIA_TYPE          => "Audio",
+                *pw::keys::MEDIA_CATEGORY      => "Capture",
+                *pw::keys::MEDIA_ROLE          => "Music",
+                *pw::keys::STREAM_CAPTURE_SINK => "true",
+                *pw::keys::NODE_LATENCY        => "240/48000",
+            },
+        };
+        let stream = pw::stream::StreamBox::new(&core, "punktfunk-audio", props)
+            .context("pw audio Stream")?;
 
-    // Request F32LE, 48 kHz, at the session's channel count with explicit positions —
+        let _listener = stream
-    // PipeWire's channel-mixer up/downmixes the sink monitor to this layout.
+            .add_local_listener_with_user_data(tx)
-    let mut info = AudioInfoRaw::new();
+            .state_changed({
-    info.set_format(AudioFormat::F32LE);
+                let mainloop = mainloop.clone();
-    info.set_rate(SAMPLE_RATE);
+                move |_s, _ud, old, new| {
-    info.set_channels(channels);
+                    tracing::info!(?old, ?new, "pipewire audio stream state");
-    info.set_position(spa_positions(channels));
+                    // A stream error is unrecoverable for this instance — exit so the sessions'
-    let obj = pw::spa::pod::Object {
+                    // reopen path builds a fresh one (same contract as the core-error path above).
-        type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
+                    if matches!(new, pw::stream::StreamState::Error(_)) {
-        id: pw::spa::param::ParamType::EnumFormat.as_raw(),
+                        mainloop.quit();
-        properties: info.into(),
+                    }
-    };
+                }
-    let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
+            })
-        std::io::Cursor::new(Vec::new()),
+            .param_changed(|_stream, _tx, id, param| {
-        &pw::spa::pod::Value::Object(obj),
+                let Some(param) = param else { return };
-    )
+                if id != pw::spa::param::ParamType::Format.as_raw() {
-    .context("serialize audio format pod")?
+                    return;
-    .0
+                }
-    .into_inner();
+                let mut info = AudioInfoRaw::default();
-    let mut params = [Pod::from_bytes(&values).context("audio pod from bytes")?];
+                if info.parse(param).is_ok() {
+                    tracing::info!(
+                        format = ?info.format(),
+                        rate = info.rate(),
+                        channels = info.channels(),
+                        "audio format negotiated"
+                    );
+                }
+            })
+            .process(|stream, tx| {
+                let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                    let Some(mut buffer) = stream.dequeue_buffer() else {
+                        return;
+                    };
+                    let datas = buffer.datas_mut();
+                    if datas.is_empty() {
+                        return;
+                    }
+                    let d = &mut datas[0];
+                    let (offset, size) = {
+                        let c = d.chunk();
+                        (c.offset() as usize, c.size() as usize)
+                    };
+                    let Some(buf) = d.data() else { return };
+                    if offset > buf.len() {
+                        return;
+                    }
+                    let region = &buf[offset..(offset + size).min(buf.len())];
+                    // Negotiated as F32LE; reinterpret the byte region as interleaved f32.
+                    let n = region.len() / 4;
+                    static FIRST: std::sync::atomic::AtomicBool =
+                        std::sync::atomic::AtomicBool::new(true);
+                    if FIRST.swap(false, std::sync::atomic::Ordering::Relaxed) {
+                        tracing::info!(samples = n, "audio first capture buffer");
+                    }
+                    let mut samples = Vec::with_capacity(n);
+                    for i in 0..n {
+                        let b = [
+                            region[i * 4],
+                            region[i * 4 + 1],
+                            region[i * 4 + 2],
+                            region[i * 4 + 3],
+                        ];
+                        samples.push(f32::from_le_bytes(b));
+                    }
+                    let _ = tx.try_send(samples); // drop if the encoder is behind
+                }));
+                if outcome.is_err() {
+                    tracing::error!("panic in pipewire audio callback — chunk dropped");
+                }
+            })
+            .register()
+            .context("register audio stream listener")?;
 
-    stream
+        // Request F32LE, 48 kHz, at the session's channel count with explicit positions. In
-        .connect(
+        // legacy mode PipeWire's channel-mixer up/downmixes the sink monitor to this layout;
-            spa::utils::Direction::Input,
+        // in stream-sink mode this IS the sink's advertised layout (apps mix/route to it).
-            None, // PW_ID_ANY — autoconnect to the default sink monitor
+        let mut info = AudioInfoRaw::new();
-            pw::stream::StreamFlags::AUTOCONNECT | pw::stream::StreamFlags::MAP_BUFFERS,
+        info.set_format(AudioFormat::F32LE);
-            &mut params,
+        info.set_rate(SAMPLE_RATE);
+        info.set_channels(channels);
+        info.set_position(spa_positions(channels));
+        let obj = pw::spa::pod::Object {
+            type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
+            id: pw::spa::param::ParamType::EnumFormat.as_raw(),
+            properties: info.into(),
+        };
+        let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
+            std::io::Cursor::new(Vec::new()),
+            &pw::spa::pod::Value::Object(obj),
         )
-        .context("pw audio stream connect")?;
+        .context("serialize audio format pod")?
+        .0
+        .into_inner();
+        let mut params = [Pod::from_bytes(&values).context("audio pod from bytes")?];
 
-    mainloop.run();
+        // RT_PROCESS in stream-sink mode for the same reason as the mic: the sink must be a
-    tracing::debug!("pipewire audio loop exited (capturer dropped)");
+        // *synchronous* graph node that joins its producers' driver group and is actually
-    Ok(())
+        // driven (see the mic's connect comment — async device-class stream nodes on a busy
+        // graph never acquire a driver and their process() never fires).
+        let mut flags = pw::stream::StreamFlags::AUTOCONNECT | pw::stream::StreamFlags::MAP_BUFFERS;
+        if sink_name.is_some() {
+            flags |= pw::stream::StreamFlags::RT_PROCESS;
+        }
+        stream
+            .connect(
+                spa::utils::Direction::Input, // we CONSUME samples (a sink / a monitor tap)
+                None,                         // PW_ID_ANY — legacy mode: the default sink monitor
+                flags,
+                &mut params,
+            )
+            .context("pw audio stream connect")?;
+
+        // Setup complete: the daemon connection and stream connect succeeded — report ready,
+        // then block until quit/death. (The connect is async server-side; if the caller's
+        // default-sink claim lands a few ms before the node registers, WirePlumber simply
+        // keeps the configured value and elects it the moment the node appears — verified
+        // live: configured values persist unelected while their target is absent.)
+        let _ = ready.send(Ok(()));
+        mainloop.run();
+        tracing::debug!("pipewire audio loop exited (capturer dropped)");
+        Ok(())
+    })();
+    if let Err(e) = &result {
+        let _ = ready.send(Err(anyhow!("{e:#}")));
+    }
+    result
 }

crates/punktfunk-host/src/audio/linux/stream_sink.rs

@@ -0,0 +1,367 @@
+//! Session-scoped **default-sink claim** for the host-owned stream sink.
+//!
+//! In stream-sink mode (see the module docs in [`super`]) the capture stream registers itself
+//! as an `Audio/Sink` node; for host apps to actually play into it, it must be the *default*
+//! sink for the duration of a stream session. WirePlumber elects the default from the
+//! `default.configured.audio.sink` metadata key (what `wpctl set-default` writes), and with
+//! `linking.follow-default-target` (default true) moves already-running app streams when it
+//! changes. So a claim is: save the current configured value, point it at our sink; a release
+//! restores it. Live-diagnosed motivation (bazzite host, 2026-07-14): the *hardware* default
+//! (HDMI audio on a TV) vanishes on every gamescope modeset, making WirePlumber ping-pong the
+//! default HDMI↔auto_null ~8×/s — a capture stream that follows the default relinks on every
+//! flip and the client hears crackle. A claimed stream sink is immune: nothing about it
+//! depends on display hardware.
+//!
+//! **Refcounted, latest-wins.** Concurrent sessions (GameStream + punktfunk/1) each hold a
+//! claim on their own capturer's sink; the newest claim points routing at *its* sink, and only
+//! the release of the last claim restores the pre-claim value — a session ending must never
+//! yank the default from under one still running. The ledger lock is held **across** the
+//! metadata round-trip so a racing claim/release pair can't interleave their writes (a stale
+//! restore overwriting a fresh claim would silence the surviving session).
+//!
+//! **Crash self-healing.** If the host dies while claimed, the configured default is left
+//! pointing at a `punktfunk-speaker-*` node that no longer exists; WirePlumber then falls back
+//! to availability-based election (local audio keeps working) and the next claim overwrites
+//! the stale value. A stale punktfunk name is never saved as a restore target — restoring it
+//! would wedge routing on a ghost sink forever — the restore degrades to *deleting* the key,
+//! i.e. handing the choice back to WirePlumber's automatic election.
+
+use anyhow::{anyhow, Context, Result};
+use std::sync::Mutex;
+
+/// `node.name` prefix for every host-owned stream sink (full names are uniqued per capturer:
+/// `punktfunk-speaker-<pid>-<seq>`, so overlapping capturers — mid-session reopen, concurrent
+/// sessions — never alias in `target`/metadata lookups). The staleness rule below matches on
+/// this prefix.
+pub(super) const SINK_NAME_PREFIX: &str = "punktfunk-speaker";
+
+/// The metadata key WirePlumber reads the user's preferred sink from (subject 0 on the
+/// `default` metadata object; value is `{"name":"<node.name>"}` typed `Spa:String:JSON`).
+const CONFIGURED_SINK_KEY: &str = "default.configured.audio.sink";
+
+/// What the last release writes back.
+#[derive(Debug, PartialEq)]
+enum Restore {
+    /// Re-set the saved pre-claim value (the raw `{"name":"..."}` JSON).
+    Value(String),
+    /// Remove the key: no pre-claim preference existed, or the saved one was a stale
+    /// punktfunk claim from a crashed host (see the module docs).
+    Delete,
+}
+
+/// Pure claim bookkeeping — separated from the PipeWire I/O so the refcount/restore rules are
+/// unit-testable on every platform.
+struct Ledger {
+    holders: u32,
+    restore: Option<Restore>,
+}
+
+impl Ledger {
+    const fn new() -> Ledger {
+        Ledger {
+            holders: 0,
+            restore: None,
+        }
+    }
+
+    /// Count a new claim; `true` means this is the first holder and the caller must save the
+    /// pre-claim value via [`note_previous`](Self::note_previous).
+    fn on_claim(&mut self) -> bool {
+        self.holders += 1;
+        self.holders == 1
+    }
+
+    /// Record what the first claim found, applying the staleness rule.
+    fn note_previous(&mut self, prev: Option<String>) {
+        self.restore = Some(match prev {
+            Some(v) if !v.contains(SINK_NAME_PREFIX) => Restore::Value(v),
+            _ => Restore::Delete,
+        });
+    }
+
+    /// Count a release; the last holder gets the restore action to apply.
+    fn on_release(&mut self) -> Option<Restore> {
+        self.holders = self.holders.saturating_sub(1);
+        if self.holders == 0 {
+            self.restore.take()
+        } else {
+            None
+        }
+    }
+}
+
+static LEDGER: Mutex<Ledger> = Mutex::new(Ledger::new());
+
+/// Point the configured default sink at `sink_name` (refcounted; see the module docs). Never
+/// fails the caller: a box where the metadata write doesn't work (WirePlumber absent) still
+/// gets a working capture — apps just aren't rerouted, which is exactly the legacy behaviour.
+pub(super) fn claim(sink_name: &str) {
+    let mut ledger = LEDGER.lock().unwrap();
+    let first = ledger.on_claim();
+    // Latest claim wins: even with an existing holder, route to the newest session's sink.
+    match set_configured_sink(Some(&format!(r#"{{"name":"{sink_name}"}}"#))) {
+        Ok(prev) => {
+            if first {
+                ledger.note_previous(prev);
+            }
+            tracing::info!(
+                sink = sink_name,
+                "claimed default sink for the stream session"
+            );
+        }
+        Err(e) => {
+            if first {
+                // Nothing knowable to restore — the release will hand election back to
+                // WirePlumber (Delete), which is also correct if IT starts working by then.
+                ledger.note_previous(None);
+            }
+            tracing::warn!(error = %format!("{e:#}"),
+                "could not claim the default sink — host apps may keep playing to the previous output");
+        }
+    }
+}
+
+/// Release one claim; the last release restores the pre-claim configured default.
+pub(super) fn release() {
+    let mut ledger = LEDGER.lock().unwrap();
+    let Some(restore) = ledger.on_release() else {
+        return;
+    };
+    let value = match &restore {
+        Restore::Value(v) => Some(v.as_str()),
+        Restore::Delete => None,
+    };
+    match set_configured_sink(value) {
+        Ok(_) => tracing::info!(
+            restored = value.unwrap_or("<automatic>"),
+            "restored default sink after the stream session"
+        ),
+        Err(e) => tracing::warn!(error = %format!("{e:#}"),
+            "could not restore the default sink — set it manually (wpctl set-default)"),
+    }
+}
+
+/// One-shot metadata round-trip: connect, find the `default` metadata object, read the current
+/// [`CONFIGURED_SINK_KEY`] value, then set it to `value` (`None` deletes the key). Returns the
+/// **previous** value. Runs its own short-lived main loop on the calling thread — claims come
+/// from session threads at start/end, never from a PipeWire callback.
+fn set_configured_sink(value: Option<&str>) -> Result<Option<String>> {
+    use pipewire as pw;
+    use std::cell::RefCell;
+    use std::rc::Rc;
+
+    crate::pwinit::ensure_init();
+    let mainloop = pw::main_loop::MainLoopRc::new(None).context("claim MainLoop")?;
+    let context = pw::context::ContextRc::new(&mainloop, None).context("claim Context")?;
+    let core = context
+        .connect_rc(None)
+        .context("claim connect (is PipeWire running in this session?)")?;
+    let registry = core.get_registry_rc().context("claim registry")?;
+
+    /// Round-trip phases: 0 = globals replaying, 1 = metadata properties replaying,
+    /// 2 = mutation flushing.
+    struct Op {
+        metadata: Option<pw::metadata::Metadata>,
+        md_listener: Option<pw::metadata::MetadataListener>,
+        previous: Option<String>,
+        phase: u8,
+        expected: Option<pw::spa::utils::result::AsyncSeq>,
+        outcome: Option<Result<()>>,
+    }
+    let op = Rc::new(RefCell::new(Op {
+        metadata: None,
+        md_listener: None,
+        previous: None,
+        phase: 0,
+        expected: None,
+        outcome: None,
+    }));
+
+    let _registry_listener = registry
+        .add_listener_local()
+        .global({
+            let op = op.clone();
+            let registry = registry.clone();
+            move |global| {
+                if global.type_ != pw::types::ObjectType::Metadata
+                    || op.borrow().metadata.is_some()
+                    || global.props.as_ref().and_then(|p| p.get("metadata.name")) != Some("default")
+                {
+                    return;
+                }
+                match registry.bind::<pw::metadata::Metadata, _>(global) {
+                    Ok(md) => {
+                        // The server replays existing properties to a fresh bind; capture the
+                        // current configured sink before mutating it.
+                        let listener = md
+                            .add_listener_local()
+                            .property({
+                                let op = op.clone();
+                                move |subject, key, _type, value| {
+                                    if subject == 0 && key == Some(CONFIGURED_SINK_KEY) {
+                                        op.borrow_mut().previous = value.map(str::to_owned);
+                                    }
+                                    0
+                                }
+                            })
+                            .register();
+                        let mut o = op.borrow_mut();
+                        o.metadata = Some(md);
+                        o.md_listener = Some(listener);
+                    }
+                    Err(e) => {
+                        op.borrow_mut().outcome = Some(Err(anyhow!("bind default metadata: {e}")));
+                    }
+                }
+            }
+        })
+        .register();
+
+    let value_owned = value.map(str::to_owned);
+    let _core_listener = core
+        .add_listener_local()
+        .done({
+            let op = op.clone();
+            let core = core.clone();
+            let mainloop = mainloop.clone();
+            move |id, seq| {
+                if id != pw::core::PW_ID_CORE {
+                    return;
+                }
+                let mut o = op.borrow_mut();
+                if o.expected != Some(seq) || o.outcome.is_some() {
+                    return;
+                }
+                match o.phase {
+                    0 => {
+                        // All pre-existing globals replayed. No `default` metadata → no
+                        // session manager to negotiate with.
+                        if o.metadata.is_none() {
+                            o.outcome = Some(Err(anyhow!(
+                                "no 'default' metadata object (is WirePlumber running?)"
+                            )));
+                            mainloop.quit();
+                            return;
+                        }
+                        o.phase = 1;
+                        o.expected = core.sync(0).ok();
+                    }
+                    1 => {
+                        // Property replay complete — `previous` holds the pre-claim value.
+                        let md = o.metadata.as_ref().unwrap();
+                        md.set_property(
+                            0,
+                            CONFIGURED_SINK_KEY,
+                            value_owned.as_ref().map(|_| "Spa:String:JSON"),
+                            value_owned.as_deref(),
+                        );
+                        o.phase = 2;
+                        o.expected = core.sync(0).ok();
+                    }
+                    _ => {
+                        o.outcome = Some(Ok(()));
+                        mainloop.quit();
+                    }
+                }
+            }
+        })
+        .error({
+            let op = op.clone();
+            let mainloop = mainloop.clone();
+            move |id, _seq, res, message| {
+                op.borrow_mut().outcome.get_or_insert(Err(anyhow!(
+                    "pipewire core error id={id} res={res}: {message}"
+                )));
+                mainloop.quit();
+            }
+        })
+        .register();
+
+    // A sick-but-connected daemon must not wedge a session start/end (the ledger lock is held
+    // across this call) — bail out after a bounded wait.
+    let timer = mainloop.loop_().add_timer({
+        let op = op.clone();
+        let mainloop = mainloop.clone();
+        move |_| {
+            op.borrow_mut()
+                .outcome
+                .get_or_insert(Err(anyhow!("metadata round-trip timed out")));
+            mainloop.quit();
+        }
+    });
+    let _ = timer.update_timer(Some(std::time::Duration::from_secs(5)), None);
+
+    op.borrow_mut().expected = core.sync(0).ok();
+    mainloop.run();
+
+    let mut o = op.borrow_mut();
+    match o.outcome.take() {
+        Some(Ok(())) => Ok(o.previous.take()),
+        Some(Err(e)) => Err(e),
+        None => Err(anyhow!("metadata loop exited unexpectedly")),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// First claim saves the pre-claim value; the last release yields it for restore.
+    #[test]
+    fn claim_release_roundtrip() {
+        let mut l = Ledger::new();
+        assert!(l.on_claim(), "first claim must save the previous value");
+        l.note_previous(Some(r#"{"name":"alsa_output.hdmi"}"#.into()));
+        assert_eq!(
+            l.on_release(),
+            Some(Restore::Value(r#"{"name":"alsa_output.hdmi"}"#.into()))
+        );
+    }
+
+    /// Nested claims (concurrent sessions): only the FIRST saves, only the LAST restores.
+    #[test]
+    fn nested_claims_restore_once() {
+        let mut l = Ledger::new();
+        assert!(l.on_claim());
+        l.note_previous(Some(r#"{"name":"alsa_output.hdmi"}"#.into()));
+        assert!(
+            !l.on_claim(),
+            "second claim must not overwrite the saved value"
+        );
+        assert_eq!(l.on_release(), None, "inner release must not restore");
+        assert_eq!(
+            l.on_release(),
+            Some(Restore::Value(r#"{"name":"alsa_output.hdmi"}"#.into()))
+        );
+    }
+
+    /// A stale punktfunk claim left by a crashed host must NEVER become the restore target —
+    /// it degrades to deleting the key (automatic election).
+    #[test]
+    fn stale_own_claim_degrades_to_delete() {
+        let mut l = Ledger::new();
+        assert!(l.on_claim());
+        l.note_previous(Some(r#"{"name":"punktfunk-speaker-4242-0"}"#.into()));
+        assert_eq!(l.on_release(), Some(Restore::Delete));
+    }
+
+    /// No pre-claim preference → restore deletes the key.
+    #[test]
+    fn unset_previous_deletes() {
+        let mut l = Ledger::new();
+        assert!(l.on_claim());
+        l.note_previous(None);
+        assert_eq!(l.on_release(), Some(Restore::Delete));
+    }
+
+    /// Release without claim (defensive) must not underflow or restore.
+    #[test]
+    fn unbalanced_release_is_harmless() {
+        let mut l = Ledger::new();
+        assert_eq!(l.on_release(), None);
+        assert!(
+            l.on_claim(),
+            "ledger must stay usable after an unbalanced release"
+        );
+    }
+}

crates/punktfunk-host/src/gamestream/audio.rs

@@ -297,6 +297,7 @@ fn run(
         None => audio::open_audio_capture(want).context("open audio capture")?,
     };
     let result = audio_body(&mut *cap, &sock, gcm_key, rikeyid, params, running);
+    cap.idle(); // parked between sessions — release the routing claim (Linux stream sink)
     *audio_cap.lock().unwrap() = Some(cap);
     result
 }

crates/punktfunk-host/src/punktfunk1.rs

@@ -2501,9 +2501,9 @@ fn audio_thread(
     // Reuse the cached capturer ONLY when its channel count matches this session's; a stereo
     // capturer left by a prior session must not feed a 5.1/7.1 session (the encoder + the client's
     // decoder are sized for `want`, so a mismatched capturer would garble/desync the audio).
-    let capturer = match audio_cap.lock().unwrap().take() {
+    let mut capturer = match audio_cap.lock().unwrap().take() {
         Some(mut c) if c.channels() == want as u32 => {
-            c.drain(); // discard audio captured between sessions
+            c.drain(); // discard audio captured between sessions (also re-claims routing)
             c
         }
         prev => {
@@ -2521,6 +2521,7 @@ fn audio_thread(
         Ok(e) => e,
         Err(e) => {
             tracing::error!(error = %e, "opus encoder");
+            capturer.idle(); // parked, not streaming — release the routing claim
             *audio_cap.lock().unwrap() = Some(capturer);
             return;
         }
@@ -2590,8 +2591,10 @@ fn audio_thread(
             }
         }
     }
-    // Return the live capturer for the next session (None if it died and never reopened).
+    // Return the live capturer for the next session (None if it died and never reopened),
-    if let Some(c) = capturer {
+    // releasing its session-scoped routing claim (Linux: the default sink moves back).
+    if let Some(mut c) = capturer {
+        c.idle();
         *audio_cap.lock().unwrap() = Some(c);
     }
 }