host: emit relay-only ticket (drop direct IP candidates)

The host ticket embedded every direct IP candidate the endpoint
discovered — on this machine that was 10 addrs, 7 of them useless
Docker-bridge gateways (172.16.0.0/12) plus LAN/public v4/v6. That
bloated the ticket to ~320 chars and leaked local network topology to
whoever received it.

Keep only the endpoint id + relay URL (~140 chars). The relay
coordinates hole-punching to a direct path after connect, so peer
reachability is unchanged; the direct addrs in the ticket only ever
shaved a moment off the first connection attempt, and n0 DNS discovery
already publishes the full addr keyed by id as a backstop.

Await endpoint.online() (15s cap) before building the ticket so the
relay URL is reliably populated; a relay outage degrades to a
possibly-incomplete ticket rather than a hang.

Experimental — isolated on feat/short-ticket pending an end-to-end
cross-machine connect test before merging to main.

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

Cargo.lock

@@ -3074,6 +3074,7 @@ dependencies = [
  "anyhow",
  "arboard",
  "ashpd",
+ "chrono",
  "clap",
  "dialoguer",
  "directories",
@@ -3086,8 +3087,10 @@ dependencies = [
  "thiserror 2.0.18",
  "tokio",
  "tokio-util",
+ "toml",
  "tracing",
  "tracing-subscriber",
+ "ureq",
  "uuid",
  "x11rb",
 ]
@@ -4448,6 +4451,34 @@ version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
 
+[[package]]
+name = "ureq"
+version = "3.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dea7109cdcd5864d4eeb1b58a1648dc9bf520360d7af16ec26d0a9354bafcfc0"
+dependencies = [
+ "base64",
+ "log",
+ "percent-encoding",
+ "rustls",
+ "rustls-pki-types",
+ "ureq-proto",
+ "utf8-zero",
+ "webpki-roots",
+]
+
+[[package]]
+name = "ureq-proto"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e994ba84b0bd1b1b0cf92878b7ef898a5c1760108fe7b6010327e274917a808c"
+dependencies = [
+ "base64",
+ "http",
+ "httparse",
+ "log",
+]
+
 [[package]]
 name = "url"
 version = "2.5.8"
@@ -4461,6 +4492,12 @@ dependencies = [
  "serde_derive",
 ]
 
+[[package]]
+name = "utf8-zero"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e"
+
 [[package]]
 name = "utf8_iter"
 version = "1.0.4"

Cargo.toml

@@ -2,7 +2,7 @@
 name = "pixelpass"
 version = "0.1.0"
 edition = "2024"
-description = "P2P screen sharing CLI over iroh + ffmpeg"
+description = "P2P screen sharing CLI over iroh"
 license = "MIT OR Apache-2.0"
 publish = false
 
@@ -30,6 +30,9 @@ uuid = { version = "1", features = ["v4"] }
 iroh-tickets = "1.0.0-rc.0"
 dialoguer = { version = "0.12", default-features = false }
 arboard = { version = "3", default-features = false, features = ["wayland-data-control"] }
+ureq = { version = "3", default-features = false, features = ["rustls"] }
+toml = "1"
+chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] }
 
 [profile.release]
 lto = "thin"

README.md

@@ -0,0 +1,240 @@
+# pixelpass
+
+P2P screen sharing CLI for Linux. Single binary, hole-punched over
+[iroh](https://www.iroh.computer/) — no port forwarding, no signup, no
+server-side accounts. Hardware-encoded H.264 + AAC audio, viewed in
+mpv or VLC.
+
+Built for people who just want to show their screen to a friend
+without spinning up a Discord call or fighting with NAT.
+
+## Status
+
+**v0.1.0** — verified end-to-end on the public internet (LTE relay path,
+~2s latency, real carrier-grade NAT) as of 2026-05-20.
+
+Working:
+- Wayland capture via the screencast portal (KDE Plasma 6 confirmed; other
+  Wayland compositors with the portal should work but are untested)
+- VAAPI H.264 encode in GStreamer (RDNA3 confirmed; other VAAPI-capable
+  GPUs should work)
+- Audio capture of the default sink's monitor, with optional per-app
+  routing (`--app <name>`) and microphone mixing (`--mic`)
+- `--repair` cleanup of orphaned PipeWire state left by a crashed host
+- iroh QUIC bi-stream tunnel, direct-UDP and relay paths both verified
+- Interactive Host/View menu with clipboard auto-copy and mpv/VLC picker
+- Headless mode for scripts (`pixelpass <ticket>`)
+- Multi-viewer fanout (default 2, configurable via `--max-viewers`;
+  shared gst pipeline, one broadcast channel per host)
+- First-run upstream bandwidth pre-flight, persisted to
+  `~/.config/pixelpass/config.toml` and used to auto-size the default
+  viewer cap
+
+Not yet working:
+- X11 capture (stubbed, returns an error — Phase 2 follow-up)
+
+## Quick start
+
+### Interactive (recommended)
+
+```sh
+pixelpass
+```
+
+On the host machine: pick "Host", share a monitor via the portal dialog,
+the ticket lands on your clipboard. Send it to your viewer however you
+like (chat, email, paste in a note). The same ticket works for multiple
+viewers up to your `--max-viewers` cap.
+
+The very first host launch offers a one-time upstream bandwidth test
+(~5 s, ~5 MB to Cloudflare's open speed-test endpoint) so it can pick
+a sensible default for the viewer cap. You can skip it and a
+conservative default (2 viewers) is used; re-run it later with
+`pixelpass --reconfigure`.
+
+On the viewer machine: run `pixelpass`, pick "View", paste the ticket,
+pick mpv or VLC. The player launches detached and the stream starts.
+
+### Headless
+
+```sh
+# host: prints a ticket on stdout, waits for a peer
+pixelpass
+
+# viewer: skips the menu
+pixelpass <ticket>
+# then run the printed mpv command in another terminal
+```
+
+## Requirements
+
+- Linux (Wayland session for now; X11 stubbed)
+- A VAAPI-capable GPU and the right driver:
+  - AMD: `libva-mesa-driver`
+  - Intel: `intel-media-driver` (modern iGPUs) or `intel-vaapi-driver` (older)
+  - NVIDIA: `libva-nvidia-driver` (untested)
+- `vainfo` from `libva-utils` should list at least one H.264 entrypoint
+- GStreamer with these plugin packages installed:
+  - `gstreamer`, `gst-plugins-base`, `gst-plugins-good`, `gst-plugins-bad`,
+    `gst-plugins-ugly`, `gst-libav`, `gst-plugin-va`, `gst-plugin-pipewire`
+- A player: `mpv` (recommended) or `vlc`
+- If you use VLC, two split plugin packages are also needed on Arch-family
+  distros — the base `vlc` package does not pull them in:
+  - `vlc-plugin-dvb` — provides the MPEG-TS demuxer (`libts_plugin.so`).
+    Without it, VLC can't parse the container.
+  - `vlc-plugin-ffmpeg` — provides the H.264 decoder
+    (`libavcodec_plugin.so`). Without it, VLC parses the container,
+    identifies the codec as H.264, then errors with
+    `Codec h264 ... is not supported`.
+  mpv ships its own decoder stack and doesn't share either dependency.
+- PipeWire (for screencast portal + audio capture)
+
+On Arch / CachyOS / EndeavourOS:
+
+```sh
+sudo pacman -S gstreamer gst-plugins-base gst-plugins-good gst-plugins-bad \
+               gst-plugins-ugly gst-libav gst-plugin-va gst-plugin-pipewire \
+               libva-utils mpv
+# plus your GPU's VAAPI driver
+# plus, if you want to use VLC instead of mpv:
+sudo pacman -S vlc vlc-plugin-dvb vlc-plugin-ffmpeg
+```
+
+If the viewer is running on battery, set the CPU governor to performance
+or balanced — power-saver can choke even hardware-decoded 1080p H.264.
+
+## Build
+
+```sh
+cargo build --release
+./target/release/pixelpass --help
+```
+
+`rustc` 1.95+ / edition 2024.
+
+## How it works
+
+```
+  Host                                                          Viewer
+  ────                                                          ──────
+  Wayland portal (ashpd) ──> PipeWire fd
+                              │
+                              ▼
+              gst-launch: pipewiresrc -> videorate -> vah264enc ->
+                          h264parse -> mpegtsmux
+                          (audio:    pulsesrc <sink>.monitor ->
+                                     avenc_aac -> aacparse ─┘)
+                              │ stdout
+                              ▼
+              tokio HTTP server (in-process, ~30 lines)
+                              │
+                              ▼
+              iroh QUIC bi-stream (ALPN pixelpass/0) ◄══════════►
+                                                                 │
+                                                                 ▼
+                                                  tokio TcpListener
+                                                  on 127.0.0.1:rand
+                                                                 │
+                                                                 ▼
+                                                  mpv / VLC HTTP client
+```
+
+The viewer's player connects to a localhost HTTP server, which is
+just one end of the iroh tunnel. The host's HTTP server sits on the
+other end and streams GStreamer's stdout (an MPEG-TS containing
+hardware-encoded H.264 + AAC) through with no demux or remux.
+
+iroh handles NAT traversal: direct UDP if hole-punching succeeds,
+relay path otherwise. Both have been verified end-to-end.
+
+## Why these choices
+
+- **iroh over Holesail / dumbpipe / Tailscale**: single Rust dep, no Node
+  runtime, no signup, no daemon — fits the "one self-contained binary"
+  goal.
+- **GStreamer for capture/encode, not ffmpeg**: stride/format pitfalls
+  when bridging raw video between processes; one in-process pipeline
+  sidesteps them.
+- **In-process Rust HTTP server, not ffmpeg-as-server**: ffmpeg's
+  `-listen 1` is one-shot and probe-budget-sensitive; the Rust task is
+  pure passthrough with no codec assumptions.
+- **MPEG-TS over fragmented MP4**: every player on Linux handles it
+  out of the box. AV1-in-MPEG-TS was tried and is unworkable through
+  libavformat — if AV1 ever comes back, it has to ride a different
+  container.
+- **VAAPI H.264 over x264**: ~5% of one CPU core instead of ~50% on
+  the host's hardware.
+
+## Configuration
+
+`pixelpass` keeps a small TOML config at `~/.config/pixelpass/config.toml`
+(or the XDG equivalent). Right now it only stores the result of the
+bandwidth pre-flight:
+
+```toml
+[bandwidth]
+status = "measured"            # measured | skipped | failed | unmeasured
+upstream_mbps = 8.78           # safe estimate (raw * 0.8)
+measured_at = "2026-05-21T20:41:16Z"
+```
+
+- `pixelpass --reconfigure` re-runs the test (e.g. after an ISP change).
+- Deleting the file resets pixelpass to first-run state.
+- Skip is sticky — once you skip the test, pixelpass won't ask again
+  unless you reconfigure.
+
+## Audio
+
+By default pixelpass captures the default sink's monitor — the viewer
+hears whatever the host hears. Two flags adjust this:
+
+- `--app <name>` routes only a single application's audio. pixelpass
+  creates a per-PID null-sink and uses libpipewire to reroute matching
+  `Stream/Output/Audio` nodes (by `application.name`) into it, so the
+  viewer hears just that app instead of the whole desktop. In the
+  interactive menu you can pick the app from a list of what's currently
+  playing.
+- `--mic` mixes the default microphone source into the stream alongside
+  system audio.
+
+If a host crashes mid-session it can leave orphaned `pixelpass_capture_*`
+null-sinks and their paired loopbacks loaded in PipeWire. Run
+`pixelpass --repair` to unload them and exit.
+
+## Multi-viewer
+
+One gst capture pipeline fans out to N concurrent viewers via a
+`tokio::sync::broadcast` channel. The same ticket is reusable: as long
+as a viewer is connected, capture stays alive; when the last one
+leaves, the pipeline tears down and the portal stops streaming. A new
+viewer connecting after that re-triggers the portal dialog.
+
+Capacity is bounded by upstream bandwidth (each viewer is its own
+encrypted egress). The default cap comes from the bandwidth pre-flight
+result; `--max-viewers <N>` overrides it. When the cap is hit,
+additional connections are politely refused with a "host is full"
+message and the host keeps running.
+
+For more viewers, drop the per-viewer bitrate: e.g. `pixelpass
+--bitrate 2500 --max-viewers 4` fits four 2.5 Mbps streams in roughly
+12 Mbps of upstream.
+
+## Known limitations and gotchas
+
+- **VLC needs `vlc-plugin-dvb` and `vlc-plugin-ffmpeg`** on Arch-family
+  distros — the base `vlc` package doesn't pull these in, and missing
+  either one breaks playback (the first kills the demuxer, the second
+  kills the H.264 decoder). pixelpass warns at player-launch time if
+  either plugin isn't on disk. mpv doesn't share these dependencies.
+- **Audio echo** if the host plays the stream through speakers and
+  captures system audio — expected, the mic / monitor picks up the
+  playback. Headphones bypass it.
+- **Late joiners see ~2 s of garbage** before the next keyframe lets
+  their decoder lock. Expected behavior, not a bug.
+- **VAAPI driver must be package-tracked**, not an orphaned `.so` on
+  disk. mpv's `--hwdec=auto` silently falls back to software decode
+  otherwise, which then chokes on a low-power viewer.
+
+## License
+
+MIT OR Apache-2.0, your pick.

src/cli.rs

@@ -4,7 +4,7 @@ use clap::{Parser, ValueEnum};
 #[command(
     name = "pixelpass",
     version,
-    about = "P2P screen sharing over iroh + ffmpeg",
+    about = "P2P screen sharing over iroh",
     long_about = "Run with no arguments for an interactive Host/View menu. \
                   Pass a ticket positionally to skip the menu and view headlessly."
 )]
@@ -41,9 +41,12 @@ pub struct Cli {
     #[arg(long)]
     pub no_hwencode: bool,
 
-    /// Use low-latency SRT transport instead of HTTP MPEG-TS (Phase 2/3).
+    /// Maximum number of concurrent viewers. Additional connections are
+    /// politely refused with a "host full" message. Defaults to the
+    /// connection-aware recommendation from the bandwidth pre-flight if
+    /// available, otherwise 2.
     #[arg(long)]
-    pub low_latency: bool,
+    pub max_viewers: Option<u32>,
 
     // ── viewer options ────────────────────────────────────────────────
     /// Local TCP port for the viewer to expose (default: random).
@@ -58,6 +61,12 @@ pub struct Cli {
     /// Clean up orphaned PipeWire state from a crashed host run, then exit.
     #[arg(long)]
     pub repair: bool,
+
+    /// Re-run the bandwidth pre-flight test, save the result, then exit.
+    /// Use this if your connection has changed (new ISP, moved house, etc.)
+    /// or if the previously saved test result is stale.
+    #[arg(long)]
+    pub reconfigure: bool,
 }
 
 #[derive(ValueEnum, Clone, Copy, Debug)]
@@ -75,7 +84,7 @@ pub struct HostOpts {
     pub bitrate: u32,
     pub framerate: u32,
     pub no_hwencode: bool,
-    pub low_latency: bool,
+    pub max_viewers: Option<u32>,
     pub interactive: bool,
 }
 
@@ -95,7 +104,7 @@ impl Cli {
             bitrate: self.bitrate,
             framerate: self.framerate,
             no_hwencode: self.no_hwencode,
-            low_latency: self.low_latency,
+            max_viewers: self.max_viewers,
             interactive,
         }
     }

src/common/bandwidth.rs

@@ -0,0 +1,73 @@
+//! One-shot upstream bandwidth measurement against Cloudflare's open
+//! speed-test endpoint. POST a fixed payload, time it, derive Mbps.
+//!
+//! Run via `tokio::task::spawn_blocking` from async contexts — ureq is a
+//! blocking client and we don't want to wedge the tokio runtime during
+//! the test.
+
+use anyhow::{Context, Result};
+use std::time::{Duration, Instant};
+
+const ENDPOINT: &str = "https://speed.cloudflare.com/__up";
+const PAYLOAD_BYTES: usize = 5 * 1024 * 1024; // 5 MiB
+const HTTP_TIMEOUT: Duration = Duration::from_secs(30);
+/// Multiplier applied to the raw measurement. TCP slow-start, ramp-up, and
+/// real-world contention all mean a one-shot upstream test slightly
+/// overestimates sustainable throughput; clamp to 80% for headroom.
+const SAFETY_FACTOR: f64 = 0.80;
+
+/// Result of a successful measurement.
+#[derive(Debug, Clone)]
+pub struct Measurement {
+    /// Raw measured throughput in megabits per second.
+    pub raw_mbps: f64,
+    /// `raw_mbps * SAFETY_FACTOR` — the value to use when sizing things.
+    pub safe_mbps: f64,
+    /// How long the upload took.
+    pub elapsed: Duration,
+}
+
+/// Blocking upload-speed test. Call from a `spawn_blocking` task.
+pub fn measure_upstream_blocking() -> Result<Measurement> {
+    let payload = vec![0u8; PAYLOAD_BYTES];
+    let agent = ureq::Agent::config_builder()
+        .timeout_global(Some(HTTP_TIMEOUT))
+        .build()
+        .new_agent();
+
+    let start = Instant::now();
+    let response = agent
+        .post(ENDPOINT)
+        .content_type("application/octet-stream")
+        .send(&payload[..])
+        .context("upload request to Cloudflare failed")?;
+    let elapsed = start.elapsed();
+
+    let status = response.status();
+    if !status.is_success() {
+        anyhow::bail!("Cloudflare returned HTTP {status}");
+    }
+
+    let bits = (PAYLOAD_BYTES as f64) * 8.0;
+    let seconds = elapsed.as_secs_f64().max(0.001);
+    let raw_mbps = bits / seconds / 1_000_000.0;
+    let safe_mbps = raw_mbps * SAFETY_FACTOR;
+
+    Ok(Measurement {
+        raw_mbps,
+        safe_mbps,
+        elapsed,
+    })
+}
+
+/// Convert a safe-upstream Mbps figure plus the host's per-viewer bitrate
+/// (kbps for video, ignoring audio + protocol overhead which we account for
+/// via SAFETY_FACTOR) into a recommended viewer count. Floors to at least 1.
+pub fn recommended_max_viewers(safe_mbps: f64, bitrate_kbps: u32) -> u32 {
+    let per_viewer_mbps = (bitrate_kbps as f64) / 1000.0;
+    if per_viewer_mbps <= 0.0 {
+        return 1;
+    }
+    let n = (safe_mbps / per_viewer_mbps).floor();
+    if n < 1.0 { 1 } else { n as u32 }
+}

src/common/config.rs

@@ -0,0 +1,101 @@
+//! Persistent user-level config at `~/.config/pixelpass/config.toml`.
+//!
+//! Right now this only tracks the bandwidth pre-flight result. Future
+//! preferences (default player, default bitrate, etc.) can hang off the
+//! same file under their own `[section]`.
+
+use anyhow::{Context, Result};
+use chrono::{DateTime, Utc};
+use directories::ProjectDirs;
+use serde::{Deserialize, Serialize};
+use std::fs;
+use std::io::Write;
+use std::path::PathBuf;
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct Config {
+    #[serde(default)]
+    pub bandwidth: BandwidthEntry,
+}
+
+/// Result of the first-run upstream measurement.
+///
+/// `status = "unmeasured"` means we've never asked the user — show the
+/// first-run dialog. `"measured"` means we have a number. `"skipped"`
+/// means the user opted out (sticky — don't ask again). `"failed"`
+/// means the last attempt errored and we should ask the user on next
+/// interactive launch whether to retry.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct BandwidthEntry {
+    #[serde(default = "default_status")]
+    pub status: BandwidthStatus,
+    #[serde(default)]
+    pub upstream_mbps: Option<f64>,
+    #[serde(default)]
+    pub measured_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum BandwidthStatus {
+    Unmeasured,
+    Measured,
+    Skipped,
+    Failed,
+}
+
+impl Default for BandwidthStatus {
+    fn default() -> Self {
+        Self::Unmeasured
+    }
+}
+
+fn default_status() -> BandwidthStatus {
+    BandwidthStatus::Unmeasured
+}
+
+/// Returns `~/.config/pixelpass/config.toml` (or the XDG equivalent on other
+/// platforms). The parent directory is created lazily by [`save`].
+pub fn config_path() -> Result<PathBuf> {
+    let dirs = ProjectDirs::from("", "", "pixelpass")
+        .context("could not locate a config directory for pixelpass")?;
+    Ok(dirs.config_dir().join("config.toml"))
+}
+
+/// Returns the loaded config, or a `Default` instance if the file doesn't
+/// exist yet. Bubble up parse errors so we don't silently overwrite a
+/// hand-edited config the user is debugging.
+pub fn load() -> Result<Config> {
+    let path = config_path()?;
+    match fs::read_to_string(&path) {
+        Ok(s) => toml::from_str::<Config>(&s)
+            .with_context(|| format!("failed to parse {}", path.display())),
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Config::default()),
+        Err(e) => Err(e).with_context(|| format!("failed to read {}", path.display())),
+    }
+}
+
+/// Atomic write via tempfile-in-same-dir + rename.
+pub fn save(cfg: &Config) -> Result<()> {
+    let path = config_path()?;
+    let parent = path
+        .parent()
+        .context("config path has no parent directory")?;
+    fs::create_dir_all(parent)
+        .with_context(|| format!("failed to create {}", parent.display()))?;
+
+    let serialized =
+        toml::to_string_pretty(cfg).context("failed to serialize config to TOML")?;
+
+    let tmp = parent.join(format!(".config.toml.tmp.{}", std::process::id()));
+    {
+        let mut f = fs::File::create(&tmp)
+            .with_context(|| format!("failed to create {}", tmp.display()))?;
+        f.write_all(serialized.as_bytes())
+            .with_context(|| format!("failed to write {}", tmp.display()))?;
+        f.sync_all().ok();
+    }
+    fs::rename(&tmp, &path)
+        .with_context(|| format!("failed to rename {} -> {}", tmp.display(), path.display()))?;
+    Ok(())
+}

src/common/mod.rs

@@ -1,4 +1,6 @@
 pub mod alpn;
+pub mod bandwidth;
+pub mod config;
 pub mod deps;
 pub mod display;
 pub mod process;

src/host/audio.rs

@@ -0,0 +1,539 @@
+//! Per-app audio routing.
+//!
+//! Two cooperating layers:
+//!
+//! - **Null-sink + loopback** (pactl shell-out): a per-PID null-sink
+//!   `pixelpass_capture_<pid>` 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.
+//!
+//! - **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).
+//!
+//! 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.
+
+use anyhow::{Context, Result, bail};
+use std::cell::RefCell;
+use std::collections::BTreeMap;
+use std::process::Command;
+use std::rc::Rc;
+use std::sync::{Arc, Mutex};
+use std::thread::JoinHandle;
+
+use crate::cli::HostOpts;
+
+/// 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 so failures get logged.
+pub struct Routing {
+    sink_module: Option<u32>,
+    /// Shared with the event task so it can `take()` and unload on the
+    /// first successful route. `Routing::shutdown` unloads whatever
+    /// remains.
+    loopback_module: Arc<Mutex<Option<u32>>>,
+    sink_name: String,
+    stream_router: Option<StreamRouter>,
+    event_task: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl Routing {
+    /// Create the per-PID null-sink + loopback. If `opts.app` is set,
+    /// also spawn the libpipewire thread that reroutes matching streams.
+    pub async fn start(opts: &HostOpts) -> Result<Self> {
+        let pid = std::process::id();
+        let sink_name = format!("pixelpass_capture_{pid}");
+
+        let sink_module =
+            load_module(&["module-null-sink", &format!("sink_name={sink_name}")])
+                .context("failed to load module-null-sink")?;
+
+        // 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)")?;
+
+        tracing::info!(
+            sink_module,
+            loopback_module,
+            %sink_name,
+            "audio routing: null-sink + loopback ready"
+        );
+
+        let loopback_arc = Arc::new(Mutex::new(Some(loopback_module)));
+        let mut routing = Self {
+            sink_module: Some(sink_module),
+            loopback_module: Arc::clone(&loopback_arc),
+            sink_name: sink_name.clone(),
+            stream_router: None,
+            event_task: None,
+        };
+
+        if let Some(app) = &opts.app {
+            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 event_task = tokio::spawn(async move {
+                while let Some(ev) = event_rx.recv().await {
+                    match ev {
+                        Event::FirstRoutedStream => {
+                            let mid = loopback_for_task.lock().unwrap().take();
+                            if let Some(id) = mid {
+                                tracing::info!(
+                                    "audio routing: first stream routed → unloading default-sink loopback"
+                                );
+                                unload_module(id);
+                            }
+                        }
+                        Event::LastRoutedStreamGone => {
+                            // Routed app exited mid-session. Restore the
+                            // default-sink loopback so the viewer hears
+                            // system audio again instead of silence.
+                            if loopback_for_task.lock().unwrap().is_some() {
+                                continue;
+                            }
+                            tracing::info!(
+                                "audio routing: last routed stream gone → restoring default-sink loopback"
+                            );
+                            match load_module(&[
+                                "module-loopback",
+                                "source=@DEFAULT_SINK@.monitor",
+                                &format!("sink={sink_name_for_task}"),
+                                "latency_msec=20",
+                            ]) {
+                                Ok(id) => {
+                                    *loopback_for_task.lock().unwrap() = Some(id);
+                                }
+                                Err(e) => {
+                                    tracing::warn!(
+                                        "audio routing: failed to re-load loopback: {e:#}"
+                                    );
+                                }
+                            }
+                        }
+                    }
+                }
+            });
+            routing.stream_router = Some(router);
+            routing.event_task = Some(event_task);
+        }
+
+        Ok(routing)
+    }
+
+    pub fn sink_name(&self) -> &str {
+        &self.sink_name
+    }
+
+    /// Stop the stream router (if any), then unload loopback (if still
+    /// loaded), then unload the null-sink. Order matters: PipeWire can
+    /// leave zombie links if you destroy a sink with active inputs.
+    pub fn shutdown(mut self) {
+        if let Some(router) = self.stream_router.take() {
+            router.shutdown();
+        }
+        if let Some(task) = self.event_task.take() {
+            task.abort();
+        }
+        if let Some(id) = self.loopback_module.lock().unwrap().take() {
+            unload_module(id);
+        }
+        if let Some(id) = self.sink_module.take() {
+            unload_module(id);
+        }
+    }
+}
+
+impl Drop for Routing {
+    fn drop(&mut self) {
+        if let Some(router) = self.stream_router.take() {
+            router.shutdown();
+        }
+        if let Some(task) = self.event_task.take() {
+            task.abort();
+        }
+        if let Some(id) = self.loopback_module.lock().unwrap().take() {
+            unload_module(id);
+        }
+        if let Some(id) = self.sink_module.take() {
+            unload_module(id);
+        }
+    }
+}
+
+// ──────────────────────────────────────────────────────────────────────
+// App enumeration (interactive picker source)
+// ──────────────────────────────────────────────────────────────────────
+
+/// One deduplicated app currently producing audio. The picker in
+/// interactive mode shows these as the per-app capture choices.
+#[derive(Debug, Clone)]
+pub struct App {
+    pub name: String,
+    pub stream_count: u32,
+}
+
+/// Enumerate apps currently sending audio to any sink, deduplicated by
+/// `application.name`. Returns an empty Vec if nothing is playing.
+pub fn list_playing_apps() -> Result<Vec<App>> {
+    let output = Command::new("pactl")
+        .args(["-f", "json", "list", "sink-inputs"])
+        .output()
+        .context("failed to run `pactl -f json list sink-inputs`")?;
+    if !output.status.success() {
+        bail!(
+            "pactl list sink-inputs failed: {}",
+            String::from_utf8_lossy(&output.stderr).trim()
+        );
+    }
+    parse_sink_inputs(&output.stdout)
+}
+
+fn parse_sink_inputs(stdout: &[u8]) -> Result<Vec<App>> {
+    let entries: Vec<SinkInput> =
+        serde_json::from_slice(stdout).context("pactl returned unparseable JSON")?;
+    let mut counts: BTreeMap<String, u32> = BTreeMap::new();
+    for entry in entries {
+        let Some(name) = entry.properties.application_name else { continue };
+        let trimmed = name.trim();
+        if trimmed.is_empty() {
+            continue;
+        }
+        *counts.entry(trimmed.to_string()).or_insert(0) += 1;
+    }
+    Ok(counts
+        .into_iter()
+        .map(|(name, stream_count)| App { name, stream_count })
+        .collect())
+}
+
+#[derive(serde::Deserialize)]
+struct SinkInput {
+    properties: SinkInputProperties,
+}
+
+#[derive(serde::Deserialize)]
+struct SinkInputProperties {
+    #[serde(rename = "application.name")]
+    application_name: Option<String>,
+}
+
+// ──────────────────────────────────────────────────────────────────────
+// pactl module helpers
+// ──────────────────────────────────────────────────────────────────────
+
+fn load_module(args: &[&str]) -> Result<u32> {
+    let output = Command::new("pactl")
+        .arg("load-module")
+        .args(args)
+        .output()
+        .context("failed to run pactl load-module")?;
+    if !output.status.success() {
+        bail!(
+            "pactl load-module failed: {}",
+            String::from_utf8_lossy(&output.stderr).trim()
+        );
+    }
+    let id_str = String::from_utf8(output.stdout)
+        .context("pactl returned non-UTF-8")?
+        .trim()
+        .to_string();
+    id_str
+        .parse::<u32>()
+        .with_context(|| format!("pactl returned unexpected module ID: {id_str:?}"))
+}
+
+fn unload_module(id: u32) {
+    let result = Command::new("pactl")
+        .arg("unload-module")
+        .arg(id.to_string())
+        .output();
+    match result {
+        Ok(output) if output.status.success() => {
+            tracing::info!(module = id, "audio routing: unloaded pactl module");
+        }
+        Ok(output) => {
+            tracing::warn!(
+                module = id,
+                stderr = %String::from_utf8_lossy(&output.stderr).trim(),
+                "audio routing: pactl unload-module exited non-zero"
+            );
+        }
+        Err(e) => {
+            tracing::warn!(
+                module = id,
+                "audio routing: failed to run pactl unload-module: {e}"
+            );
+        }
+    }
+}
+
+// ──────────────────────────────────────────────────────────────────────
+// 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<Cmd>,
+    thread: Option<JoinHandle<()>>,
+}
+
+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,
+    ) -> Result<(Self, tokio::sync::mpsc::UnboundedReceiver<Event>)> {
+        let (cmd_tx, cmd_rx) = pipewire::channel::channel::<Cmd>();
+        let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel::<Event>();
+
+        let thread = std::thread::Builder::new()
+            .name("pixelpass-pw-router".to_string())
+            .spawn(move || {
+                if let Err(e) = run_router(filter_name, sink_name, cmd_rx, event_tx) {
+                    tracing::warn!("audio routing: libpipewire thread exited with error: {e:#}");
+                }
+            })
+            .context("failed to spawn libpipewire router thread")?;
+
+        Ok((
+            Self {
+                cmd_tx,
+                thread: Some(thread),
+            },
+            event_rx,
+        ))
+    }
+
+    fn shutdown(mut self) {
+        // Best-effort: if the send fails the thread is already gone.
+        let _ = self.cmd_tx.send(Cmd::Shutdown);
+        if let Some(t) = self.thread.take()
+            && let Err(e) = t.join()
+        {
+            tracing::warn!("audio routing: pw thread join failed: {e:?}");
+        }
+    }
+}
+
+/// 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<Cmd>,
+    event_tx: tokio::sync::mpsc::UnboundedSender<Event>,
+) -> 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 _cmd_recv = cmd_rx.attach(main_loop.loop_(), move |cmd| match cmd {
+        Cmd::Shutdown => {
+            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()) {
+                        if let Some(serial) = props
+                            .get("object.serial")
+                            .and_then(|s| s.parse::<u32>().ok())
+                        {
+                            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);
+                        }
+                        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");
+    main_loop.run();
+    tracing::info!("audio routing: pw thread exiting");
+    Ok(())
+}
+
+struct RouterState {
+    sink_serial: Option<u32>,
+    default_metadata: Option<pipewire::metadata::Metadata>,
+    routed_node_ids: Vec<u32>,
+    pending: Vec<u32>,
+}
+
+/// 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<RefCell<RouterState>>,
+    event_tx: &tokio::sync::mpsc::UnboundedSender<Event>,
+    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<RefCell<RouterState>>,
+    event_tx: &tokio::sync::mpsc::UnboundedSender<Event>,
+) {
+    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);
+    }
+}

src/host/mod.rs

@@ -1,14 +1,34 @@
+pub mod audio;
 mod capture;
+mod serve;
 mod wayland;
 
 use anyhow::{Result, bail};
-use iroh::Endpoint;
 use iroh::endpoint::{Connection, presets};
+use iroh::{Endpoint, EndpointAddr};
 use iroh_tickets::endpoint::EndpointTicket;
+use std::time::Duration;
+use tokio::sync::{mpsc, oneshot};
 use tokio_util::sync::CancellationToken;
 
 use crate::cli::HostOpts;
-use crate::common::{alpn::ALPN, deps, display::DisplayServer, signal};
+use crate::common::{
+    alpn::ALPN, bandwidth, config, config::BandwidthStatus, deps, display::DisplayServer, signal,
+    tunnel,
+};
+
+use self::capture::CaptureHandle;
+
+/// Messages from per-viewer tasks to the capture supervisor.
+enum SupervisorMsg {
+    /// A new viewer wants in. Supervisor replies with the local capture
+    /// HTTP port to connect to, or an error string if the host is full or
+    /// capture spawn failed.
+    AddViewer(oneshot::Sender<Result<u16, String>>),
+    /// A viewer's session ended. Supervisor decrements the count and tears
+    /// down capture if it just hit zero.
+    RemoveViewer,
+}
 
 pub async fn run(opts: HostOpts) -> Result<()> {
     let display = DisplayServer::resolve(opts.display_server);
@@ -21,6 +41,11 @@ pub async fn run(opts: HostOpts) -> Result<()> {
         );
     }
 
+    let resolution = resolve_max_viewers(&opts);
+    if resolution.value == 0 {
+        bail!("--max-viewers must be at least 1");
+    }
+
     let cancel = signal::install_ctrl_c();
 
     let endpoint = Endpoint::builder(presets::N0)
@@ -28,76 +53,195 @@ pub async fn run(opts: HostOpts) -> Result<()> {
         .bind()
         .await?;
 
+    // Relay-only ticket: wait for the home relay to connect, then keep only
+    // the endpoint id + relay URL and drop the direct IP candidates. The relay
+    // coordinates hole-punching to a direct path right after connect, so this
+    // doesn't change whether peers can reach each other — it just keeps the
+    // ticket short (~140 vs ~320 chars) and stops it from leaking LAN /
+    // Docker-bridge addresses to whoever receives the ticket. Awaiting online()
+    // first guarantees the relay URL is actually present (addr() right after
+    // bind can return before the relay handshake completes); the 15s cap means
+    // a relay outage degrades to a possibly-incomplete ticket rather than a hang
+    // (n0 DNS discovery still resolves the id in that case).
+    if tokio::time::timeout(Duration::from_secs(15), endpoint.online())
+        .await
+        .is_err()
+    {
+        tracing::warn!("home relay not connected within 15s; ticket may be incomplete");
+    }
     let addr = endpoint.addr();
-    let ticket = EndpointTicket::new(addr);
+    let relay_only =
+        EndpointAddr::new(addr.id).with_addrs(addr.addrs.iter().filter(|a| a.is_relay()).cloned());
+    let ticket = EndpointTicket::new(relay_only);
     let clipboard_ok = opts.interactive && copy_to_clipboard(&ticket.to_string());
-    print_host_banner(&ticket, display, &opts, clipboard_ok);
+    print_host_banner(&ticket, display, &opts, &resolution, clipboard_ok);
 
-    let result = accept_loop(&endpoint, display, &opts, cancel.clone()).await;
+    let (sup_tx, sup_rx) = mpsc::channel::<SupervisorMsg>(16);
+    let supervisor = tokio::spawn(supervise(opts.clone(), display, resolution.value, sup_rx));
+
+    accept_loop(&endpoint, sup_tx.clone(), cancel.clone()).await;
+
+    drop(sup_tx);
+    let _ = supervisor.await;
 
     endpoint.close().await;
-    result
+    Ok(())
 }
 
 async fn accept_loop(
     endpoint: &Endpoint,
-    display: DisplayServer,
+    sup_tx: mpsc::Sender<SupervisorMsg>,
-    opts: &HostOpts,
     cancel: CancellationToken,
-) -> Result<()> {
+) {
-    tokio::select! {
+    loop {
-        _ = cancel.cancelled() => {
+        tokio::select! {
-            tracing::info!("cancellation requested before any peer connected");
+            _ = cancel.cancelled() => {
-            Ok(())
+                tracing::info!("cancellation requested — closing accept loop");
-        }
+                return;
-        accepted = endpoint.accept() => {
+            }
-            let Some(incoming) = accepted else {
+            accepted = endpoint.accept() => {
-                bail!("endpoint stopped accepting connections");
+                let Some(incoming) = accepted else {
-            };
+                    tracing::info!("endpoint stopped accepting connections");
-            let conn = incoming.await?;
+                    return;
-            let remote = conn.remote_id();
+                };
-            tracing::info!(%remote, "peer connected");
+                let conn = match incoming.await {
-            eprintln!("\n[pixelpass] peer connected: {remote}\n");
+                    Ok(c) => c,
-            handle_peer(conn, display, opts, cancel).await
+                    Err(e) => {
+                        tracing::warn!("incoming connection failed: {e:#}");
+                        continue;
+                    }
+                };
+                let sup_tx = sup_tx.clone();
+                let cancel = cancel.clone();
+                tokio::spawn(handle_peer(conn, sup_tx, cancel));
+            }
         }
     }
 }
 
 async fn handle_peer(
     conn: Connection,
-    display: DisplayServer,
+    sup_tx: mpsc::Sender<SupervisorMsg>,
-    opts: &HostOpts,
     cancel: CancellationToken,
-) -> Result<()> {
+) {
-    let (quic_send, quic_recv) = conn.accept_bi().await?;
+    let remote = conn.remote_id();
 
-    let capture_handle = capture::spawn(display, opts).await?;
+    let (reply_tx, reply_rx) = oneshot::channel();
-    let port = capture_handle.local_port();
+    if sup_tx.send(SupervisorMsg::AddViewer(reply_tx)).await.is_err() {
-    let tcp = wayland::connect_to_capture(port, std::time::Duration::from_secs(5)).await?;
+        tracing::warn!(%remote, "supervisor channel closed; dropping peer");
-
+        return;
-    let bridge = crate::common::tunnel::bridge(quic_send, quic_recv, tcp);
+    }
-
+    let port = match reply_rx.await {
-    tokio::select! {
+        Ok(Ok(p)) => p,
-        res = bridge => {
+        Ok(Err(reason)) => {
-            if let Err(e) = res {
+            tracing::warn!(%remote, %reason, "refusing viewer");
-                tracing::warn!("bridge ended with error: {e:#}");
+            eprintln!("[pixelpass] refusing viewer {remote}: {reason}");
-            } else {
+            return;
-                tracing::info!("bridge closed cleanly");
-            }
         }
+        Err(_) => {
+            tracing::warn!(%remote, "supervisor reply dropped; dropping peer");
+            return;
+        }
+    };
+
+    let (quic_send, quic_recv) = match conn.accept_bi().await {
+        Ok(s) => s,
+        Err(e) => {
+            tracing::warn!(%remote, "accept_bi failed: {e:#}");
+            let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
+            return;
+        }
+    };
+
+    eprintln!("[pixelpass] viewer connected: {remote}");
+
+    let tcp = match serve::connect_to_capture(port, Duration::from_secs(5)).await {
+        Ok(t) => t,
+        Err(e) => {
+            tracing::warn!(%remote, "connect_to_capture failed: {e:#}");
+            let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
+            return;
+        }
+    };
+
+    let bridge = tunnel::bridge(quic_send, quic_recv, tcp);
+    tokio::select! {
+        res = bridge => match res {
+            Ok(()) => tracing::info!(%remote, "bridge closed cleanly"),
+            Err(e) => tracing::info!(%remote, "bridge ended: {e:#}"),
+        },
         _ = cancel.cancelled() => {
-            tracing::info!("cancellation requested during stream");
+            tracing::info!(%remote, "cancellation during stream");
         }
     }
 
-    capture_handle.shutdown().await;
+    eprintln!("[pixelpass] viewer disconnected: {remote}");
-    Ok(())
+    let _ = sup_tx.send(SupervisorMsg::RemoveViewer).await;
+}
+
+/// Owns the single shared CaptureHandle and the active viewer count. Spawns
+/// capture lazily on the first AddViewer; tears it down when the count drops
+/// back to zero. Enforces the max-viewers cap by refusing AddViewer when
+/// the count is already at the cap.
+async fn supervise(
+    opts: HostOpts,
+    display: DisplayServer,
+    max_viewers: u32,
+    mut rx: mpsc::Receiver<SupervisorMsg>,
+) {
+    let mut handle: Option<CaptureHandle> = None;
+    let mut count: u32 = 0;
+
+    while let Some(msg) = rx.recv().await {
+        match msg {
+            SupervisorMsg::AddViewer(reply) => {
+                if count >= max_viewers {
+                    let _ = reply.send(Err(format!(
+                        "host is full ({count} of {max_viewers} viewers connected)"
+                    )));
+                    continue;
+                }
+
+                if handle.is_none() {
+                    tracing::info!("first viewer arriving — spawning capture");
+                    match capture::spawn(display, &opts).await {
+                        Ok(h) => handle = Some(h),
+                        Err(e) => {
+                            let _ = reply.send(Err(format!("capture spawn failed: {e:#}")));
+                            continue;
+                        }
+                    }
+                }
+
+                let port = handle.as_ref().expect("handle was just set").local_port();
+                count += 1;
+                let _ = reply.send(Ok(port));
+                tracing::info!(active = count, cap = max_viewers, "viewer joined");
+            }
+            SupervisorMsg::RemoveViewer => {
+                count = count.saturating_sub(1);
+                tracing::info!(active = count, cap = max_viewers, "viewer left");
+                if count == 0
+                    && let Some(h) = handle.take()
+                {
+                    tracing::info!("last viewer left — tearing down capture");
+                    h.shutdown().await;
+                }
+            }
+        }
+    }
+
+    if let Some(h) = handle.take() {
+        tracing::info!("host shutdown — tearing down capture");
+        h.shutdown().await;
+    }
 }
 
 fn print_host_banner(
     ticket: &EndpointTicket,
     display: DisplayServer,
     opts: &HostOpts,
+    resolution: &MaxViewersResolution,
     clipboard_ok: bool,
 ) {
     eprintln!();
@@ -106,23 +250,79 @@ fn print_host_banner(
     eprintln!("│  capture        : {}", capture_summary(opts));
     eprintln!("│  bitrate / fps  : {} kbps @ {} fps", opts.bitrate, opts.framerate);
     eprintln!("│  hw encode      : {}", if opts.no_hwencode { "off" } else { "auto (VAAPI if available)" });
+    eprintln!("│  max viewers    : {} ({})", resolution.value, resolution.source.label());
     eprintln!("│");
     if clipboard_ok {
         eprintln!("│  Your share code has been copied to your clipboard.");
-        eprintln!("│  Send it to your viewer. (If clipboard didn't work, the");
+        eprintln!("│  Send it to your viewer(s). (If clipboard didn't work, the");
         eprintln!("│  code is also shown below for manual copy.)");
     } else {
-        eprintln!("│  Share this ticket with your viewer:");
+        eprintln!("│  Share this ticket with your viewer(s):");
     }
     eprintln!("│");
     eprintln!("│      pixelpass {ticket}");
     eprintln!("│");
-    eprintln!("│  Capture will not start until the viewer connects.");
+    eprintln!("│  Capture starts when the first viewer connects, runs while");
-    eprintln!("│  Press Ctrl+C to stop.");
+    eprintln!("│  any viewer is connected, and tears down when the last one");
+    eprintln!("│  leaves. Press Ctrl+C to stop the host entirely.");
     eprintln!("└────────────────────────────────────────────────────────────");
     eprintln!();
 }
 
+/// How we arrived at the final viewer cap. Surfaced in the banner so the
+/// user can tell at a glance whether the number is what they specified,
+/// what their measured upstream supports, or just the fallback default.
+struct MaxViewersResolution {
+    value: u32,
+    source: MaxViewersSource,
+}
+
+enum MaxViewersSource {
+    /// User passed --max-viewers explicitly.
+    UserFlag,
+    /// Derived from the saved bandwidth measurement.
+    BandwidthMeasurement { safe_mbps: f64 },
+    /// No flag, no measurement — falling back.
+    DefaultFallback,
+}
+
+impl MaxViewersSource {
+    fn label(&self) -> String {
+        match self {
+            MaxViewersSource::UserFlag => "user-specified".to_string(),
+            MaxViewersSource::BandwidthMeasurement { safe_mbps } => {
+                format!("auto: {safe_mbps:.1} Mbps measured upstream")
+            }
+            MaxViewersSource::DefaultFallback => {
+                "default — run `pixelpass --reconfigure` for a connection-aware value".to_string()
+            }
+        }
+    }
+}
+
+fn resolve_max_viewers(opts: &HostOpts) -> MaxViewersResolution {
+    if let Some(n) = opts.max_viewers {
+        return MaxViewersResolution {
+            value: n,
+            source: MaxViewersSource::UserFlag,
+        };
+    }
+    if let Ok(cfg) = config::load()
+        && cfg.bandwidth.status == BandwidthStatus::Measured
+        && let Some(upstream) = cfg.bandwidth.upstream_mbps
+    {
+        let n = bandwidth::recommended_max_viewers(upstream, opts.bitrate);
+        return MaxViewersResolution {
+            value: n,
+            source: MaxViewersSource::BandwidthMeasurement { safe_mbps: upstream },
+        };
+    }
+    MaxViewersResolution {
+        value: 2,
+        source: MaxViewersSource::DefaultFallback,
+    }
+}
+
 fn copy_to_clipboard(text: &str) -> bool {
     match arboard::Clipboard::new().and_then(|mut cb| cb.set_text(text.to_owned())) {
         Ok(()) => true,

src/host/serve.rs

@@ -0,0 +1,191 @@
+//! Display-server-agnostic serving layer: takes a capture child's stdout
+//! producing MPEG-TS bytes and fans them out to N concurrent HTTP viewers
+//! on a localhost port. One reader task pumps stdout chunks into a
+//! tokio::sync::broadcast channel; the accept loop spawns one drain task
+//! per accepted TCP connection. Slow consumers see Lagged and skip ahead;
+//! MPEG-TS resyncs at the next keyframe.
+//!
+//! Backends (host/wayland.rs, future host/x11.rs) build their own gst
+//! pipeline and hand the resulting ChildStdout to [`Serve::bind`].
+
+use anyhow::{Context, Result, bail};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
+use tokio::process::ChildStdout;
+use tokio::sync::broadcast;
+use tokio::task::JoinHandle;
+use tokio::time::{Instant, sleep};
+
+/// Broadcast-channel capacity in chunks. Each chunk is up to 64 KiB from
+/// the capture child's stdout, so 16 chunks ≈ 1 MiB ≈ ~2 s of buffered
+/// jitter at typical bitrates. A viewer that falls behind by more than
+/// this gets Lagged and skips ahead — MPEG-TS recovers at the next
+/// keyframe.
+const FANOUT_CAPACITY: usize = 16;
+
+/// Size of each chunk read from the capture child's stdout.
+const READ_CHUNK: usize = 64 * 1024;
+
+/// Owns the localhost HTTP listener and the two long-running tasks that
+/// pump bytes from a capture child to all connected viewers.
+pub struct Serve {
+    port: u16,
+    reader: Option<JoinHandle<()>>,
+    server: Option<JoinHandle<()>>,
+}
+
+impl Serve {
+    /// Bind a localhost listener on a random port, set up the broadcast
+    /// fanout, and spawn the reader + accept-loop tasks. The provided
+    /// `stdout` is assumed to produce MPEG-TS bytes.
+    pub async fn bind(stdout: ChildStdout) -> Result<Self> {
+        let listener = TcpListener::bind("127.0.0.1:0")
+            .await
+            .context("could not bind local capture HTTP listener")?;
+        let port = listener.local_addr()?.port();
+
+        let (tx, _) = broadcast::channel::<Arc<Vec<u8>>>(FANOUT_CAPACITY);
+        let reader = tokio::spawn(pump_to_broadcast(stdout, tx.clone()));
+        let server = tokio::spawn(run_accept_loop(listener, tx));
+
+        Ok(Self {
+            port,
+            reader: Some(reader),
+            server: Some(server),
+        })
+    }
+
+    pub fn local_port(&self) -> u16 {
+        self.port
+    }
+
+    /// Abort the reader and accept-loop tasks. Backends typically call this
+    /// after killing their capture child so the reader sees stdout EOF and
+    /// exits on its own; the abort is a backstop.
+    pub async fn shutdown(mut self) {
+        if let Some(task) = self.reader.take() {
+            task.abort();
+        }
+        if let Some(task) = self.server.take() {
+            task.abort();
+        }
+    }
+}
+
+impl Drop for Serve {
+    fn drop(&mut self) {
+        if let Some(task) = self.reader.as_ref() {
+            task.abort();
+        }
+        if let Some(task) = self.server.as_ref() {
+            task.abort();
+        }
+    }
+}
+
+/// Connect to the local capture HTTP listener, retrying until it's up or
+/// we time out. Returns the connected socket — the bridge layer pipes
+/// QUIC↔this socket once it's open.
+pub async fn connect_to_capture(port: u16, max_wait: Duration) -> Result<TcpStream> {
+    let deadline = Instant::now() + max_wait;
+    loop {
+        match TcpStream::connect(("127.0.0.1", port)).await {
+            Ok(stream) => return Ok(stream),
+            Err(_) if Instant::now() < deadline => {
+                sleep(Duration::from_millis(50)).await;
+            }
+            Err(e) => bail!("capture HTTP listener never came up on 127.0.0.1:{port}: {e}"),
+        }
+    }
+}
+
+/// Read the capture child's stdout in chunks and broadcast each to all
+/// current subscribers. `broadcast::send` returns Err when there are no
+/// receivers; we ignore it so the capture child isn't backpressured
+/// waiting for a viewer.
+async fn pump_to_broadcast(mut stdout: ChildStdout, tx: broadcast::Sender<Arc<Vec<u8>>>) {
+    let mut buf = vec![0u8; READ_CHUNK];
+    loop {
+        match stdout.read(&mut buf).await {
+            Ok(0) => {
+                tracing::info!("capture stdout EOF — fanout reader exiting");
+                return;
+            }
+            Ok(n) => {
+                let chunk = Arc::new(buf[..n].to_vec());
+                let _ = tx.send(chunk);
+            }
+            Err(e) => {
+                tracing::warn!("capture stdout read error: {e}");
+                return;
+            }
+        }
+    }
+}
+
+async fn run_accept_loop(listener: TcpListener, tx: broadcast::Sender<Arc<Vec<u8>>>) {
+    loop {
+        let sock = match listener.accept().await {
+            Ok((s, _)) => s,
+            Err(e) => {
+                tracing::warn!("capture HTTP accept failed: {e}");
+                return;
+            }
+        };
+        let rx = tx.subscribe();
+        tokio::spawn(serve_one_viewer(sock, rx));
+    }
+}
+
+async fn serve_one_viewer(mut sock: TcpStream, mut rx: broadcast::Receiver<Arc<Vec<u8>>>) {
+    if !drain_http_request(&mut sock).await {
+        return;
+    }
+
+    const RESPONSE: &[u8] = b"HTTP/1.1 200 OK\r\n\
+        Content-Type: video/mp2t\r\n\
+        Cache-Control: no-cache, no-store\r\n\
+        Connection: close\r\n\
+        \r\n";
+    if sock.write_all(RESPONSE).await.is_err() {
+        return;
+    }
+
+    loop {
+        match rx.recv().await {
+            Ok(chunk) => {
+                if sock.write_all(&chunk).await.is_err() {
+                    return;
+                }
+            }
+            Err(broadcast::error::RecvError::Lagged(skipped)) => {
+                tracing::warn!(
+                    skipped,
+                    "viewer fanout lagged — MPEG-TS will resync at next keyframe"
+                );
+                continue;
+            }
+            Err(broadcast::error::RecvError::Closed) => return,
+        }
+    }
+}
+
+async fn drain_http_request(sock: &mut TcpStream) -> bool {
+    let mut buf = [0u8; 1024];
+    let mut total = Vec::with_capacity(512);
+    loop {
+        match sock.read(&mut buf).await {
+            Ok(0) => return false,
+            Ok(n) => total.extend_from_slice(&buf[..n]),
+            Err(_) => return false,
+        }
+        if total.windows(4).any(|w| w == b"\r\n\r\n") {
+            return true;
+        }
+        if total.len() > 16 * 1024 {
+            return false;
+        }
+    }
+}

src/host/wayland.rs

@@ -1,7 +1,6 @@
-//! Wayland capture: ashpd ScreenCast portal → PipeWire fd → gst-launch
+//! Wayland capture: ashpd ScreenCast portal → PipeWire fd → gst-launch.
-//! pipewiresrc → MPEG-TS on gst stdout → in-process HTTP server bound on a
+//! Builds the gst pipeline that produces MPEG-TS on stdout, then hands
-//! random localhost port. The host bridge TCP-connects to that server and
+//! that stdout to [`super::serve::Serve`] which handles the HTTP fanout.
-//! pumps bytes to QUIC.
 
 use anyhow::{Context, Result, bail};
 use ashpd::{
@@ -17,28 +16,31 @@ use nix::unistd::{Pid, close};
 use std::os::fd::{AsFd, IntoRawFd, OwnedFd, RawFd};
 use std::process::Stdio;
 use std::time::Duration;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::process::{Child, Command};
-use tokio::net::{TcpListener, TcpStream};
+use tokio::time::timeout;
-use tokio::process::{Child, ChildStdout, Command};
-use tokio::task::JoinHandle;
-use tokio::time::{Instant, sleep, timeout};
 
+use super::audio::Routing;
+use super::serve::Serve;
 use crate::cli::HostOpts;
 
 pub struct CaptureHandle {
-    port: u16,
     gst: Option<Child>,
-    server: Option<JoinHandle<()>>,
+    audio: Option<Routing>,
+    serve: Option<Serve>,
 }
 
 impl CaptureHandle {
     pub fn local_port(&self) -> u16 {
-        self.port
+        self.serve
+            .as_ref()
+            .expect("serve is always Some until shutdown")
+            .local_port()
     }
 
-    /// Graceful teardown: SIGTERM gst, give it ~1s to exit, then SIGKILL, then
+    /// Graceful teardown: SIGTERM gst, give it ~1s to exit, then SIGKILL,
-    /// abort the HTTP server task. Call this before dropping; Drop only fires
+    /// unload audio routing (if any), then tear down the serve layer.
-    /// the kill backstop.
+    /// The serve reader will see EOF on gst stdout and exit on its own;
+    /// serve.shutdown() is the backstop.
     pub async fn shutdown(mut self) {
         if let Some(child) = self.gst.as_mut()
             && let Some(pid) = child.id()
@@ -49,8 +51,11 @@ impl CaptureHandle {
             let _ = timeout(Duration::from_millis(1000), child.wait()).await;
             let _ = child.start_kill();
         }
-        if let Some(task) = self.server.take() {
+        if let Some(audio) = self.audio.take() {
-            task.abort();
+            audio.shutdown();
+        }
+        if let Some(serve) = self.serve.take() {
+            serve.shutdown().await;
         }
     }
 }
@@ -60,9 +65,7 @@ impl Drop for CaptureHandle {
         if let Some(child) = self.gst.as_mut() {
             let _ = child.start_kill();
         }
-        if let Some(task) = self.server.as_ref() {
+        // Routing's and Serve's own Drop impls handle the rest.
-            task.abort();
-        }
     }
 }
 
@@ -109,20 +112,35 @@ pub async fn start(opts: &HostOpts) -> Result<CaptureHandle> {
     clear_cloexec(&pw_fd)?;
     let raw_fd: RawFd = pw_fd.into_raw_fd();
 
-    // 2. Bind the in-process HTTP listener on a random localhost port.
+    // 2. Spawn gst-launch with the full pipeline: video AND audio captured,
-    let listener = TcpListener::bind("127.0.0.1:0")
-        .await
-        .context("could not bind local capture HTTP listener")?;
-    let port = listener.local_addr()?.port();
-
-    // 3. Spawn gst-launch with the full pipeline: video AND audio captured,
     //    encoded, and muxed into MPEG-TS inside gst. Output goes to stdout,
-    //    which we pipe straight to our HTTP server task — no demux/remux,
+    //    which the serve layer pipes to its HTTP fanout — no demux/remux,
     //    no codec assumptions.
     let key_interval = (opts.framerate * 2).to_string();
     let bitrate = opts.bitrate.to_string();
-    let audio_monitor = default_audio_monitor().await?;
+
-    let audio_device = format!("device={audio_monitor}");
+    // Audio routing activates when either:
+    // - `opts.app` is set (per-stream rerouting to a per-PID null-sink),
+    // - or `PIXELPASS_AUDIO_VIA_NULL_SINK=1` is set (no app filter, just
+    //   captures everything via the null-sink → useful for development
+    //   and dogfooding the loopback path before app filtering is picked).
+    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)
+                .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}")
+    };
     let mut gst_cmd = Command::new("gst-launch-1.0");
     gst_cmd
         .args([
@@ -199,62 +217,17 @@ pub async fn start(opts: &HostOpts) -> Result<CaptureHandle> {
         .take()
         .context("gst-launch-1.0 stdout pipe unavailable")?;
 
-    // 4. Spawn the HTTP server task. It owns the listener + gst stdout: it
+    // 3. Hand stdout to the serve layer, which binds the localhost HTTP
-    //    accepts one client (the host's bridge socket via connect_to_capture),
+    //    listener and runs the broadcast fanout.
-    //    drains the HTTP request, writes a fixed MPEG-TS response, then
+    let serve = Serve::bind(gst_stdout).await?;
-    //    copies gst stdout to the socket forever.
-    let server = tokio::spawn(serve_capture(listener, gst_stdout));
 
     Ok(CaptureHandle {
-        port,
         gst: Some(gst),
-        server: Some(server),
+        audio: audio_routing,
+        serve: Some(serve),
     })
 }
 
-async fn serve_capture(listener: TcpListener, mut gst_stdout: ChildStdout) {
-    let mut sock = match listener.accept().await {
-        Ok((s, _)) => s,
-        Err(e) => {
-            tracing::warn!("capture HTTP accept failed: {e}");
-            return;
-        }
-    };
-
-    if !drain_http_request(&mut sock).await {
-        return;
-    }
-
-    const RESPONSE: &[u8] = b"HTTP/1.1 200 OK\r\n\
-        Content-Type: video/mp2t\r\n\
-        Cache-Control: no-cache, no-store\r\n\
-        Connection: close\r\n\
-        \r\n";
-    if sock.write_all(RESPONSE).await.is_err() {
-        return;
-    }
-
-    let _ = tokio::io::copy(&mut gst_stdout, &mut sock).await;
-}
-
-async fn drain_http_request(sock: &mut TcpStream) -> bool {
-    let mut buf = [0u8; 1024];
-    let mut total = Vec::with_capacity(512);
-    loop {
-        match sock.read(&mut buf).await {
-            Ok(0) => return false,
-            Ok(n) => total.extend_from_slice(&buf[..n]),
-            Err(_) => return false,
-        }
-        if total.windows(4).any(|w| w == b"\r\n\r\n") {
-            return true;
-        }
-        if total.len() > 16 * 1024 {
-            return false;
-        }
-    }
-}
-
 fn clear_cloexec(fd: &impl AsFd) -> Result<()> {
     let flags_int = fcntl(fd.as_fd(), FcntlArg::F_GETFD).context("F_GETFD on pipewire fd")?;
     let mut flags = FdFlag::from_bits_truncate(flags_int);
@@ -263,22 +236,6 @@ fn clear_cloexec(fd: &impl AsFd) -> Result<()> {
     Ok(())
 }
 
-/// Connect to the in-process capture HTTP listener, retrying until it's up or
-/// we time out. Returns the connected socket — the listener accepts exactly
-/// one connection (the bridge socket), so this stream IS the bridge socket.
-pub async fn connect_to_capture(port: u16, max_wait: Duration) -> Result<TcpStream> {
-    let deadline = Instant::now() + max_wait;
-    loop {
-        match TcpStream::connect(("127.0.0.1", port)).await {
-            Ok(stream) => return Ok(stream),
-            Err(_) if Instant::now() < deadline => {
-                sleep(Duration::from_millis(50)).await;
-            }
-            Err(e) => bail!("capture HTTP listener never came up on 127.0.0.1:{port}: {e}"),
-        }
-    }
-}
-
 async fn default_audio_monitor() -> Result<String> {
     let output = Command::new("pactl")
         .arg("get-default-sink")

src/interactive.rs

@@ -4,6 +4,7 @@ use iroh_tickets::endpoint::EndpointTicket;
 use std::str::FromStr;
 
 use crate::cli::Cli;
+use crate::common::{bandwidth, config};
 use crate::{host, viewer};
 
 pub async fn run(cli: Cli) -> Result<()> {
@@ -20,7 +21,14 @@ pub async fn run(cli: Cli) -> Result<()> {
         .interact()?;
 
     match choice {
-        0 => host::run(cli.into_host_opts(true)).await,
+        0 => {
+            preflight_if_needed(&theme).await;
+            let mut cli = cli;
+            if cli.app.is_none() {
+                cli.app = pick_app(&theme)?;
+            }
+            host::run(cli.into_host_opts(true)).await
+        }
         _ => {
             let ticket = prompt_ticket(&theme)?;
             viewer::run(ticket, cli.into_viewer_opts(true)).await
@@ -28,6 +36,182 @@ pub async fn run(cli: Cli) -> Result<()> {
     }
 }
 
+/// Picker for the per-app audio capture choice. Lists apps currently
+/// producing audio (deduped by `application.name`); user picks one or
+/// the "all system audio" default. Bypassed when `--app NAME` was given
+/// on the CLI.
+fn pick_app(theme: &ColorfulTheme) -> Result<Option<String>> {
+    let apps = match host::audio::list_playing_apps() {
+        Ok(a) => a,
+        Err(e) => {
+            tracing::warn!("could not enumerate playing apps: {e:#}");
+            return Ok(None);
+        }
+    };
+
+    eprintln!();
+    eprintln!("Audio capture");
+    eprintln!("─────────────");
+    if apps.is_empty() {
+        eprintln!("No other apps are currently producing audio.");
+        eprintln!("Start your game / music / call first if you want to pick it specifically.");
+    }
+    eprintln!();
+
+    let mut items = vec!["Capture all system audio (default)".to_string()];
+    for app in &apps {
+        items.push(if app.stream_count == 1 {
+            app.name.clone()
+        } else {
+            format!("{} ({} streams)", app.name, app.stream_count)
+        });
+    }
+
+    let choice = Select::with_theme(theme)
+        .with_prompt("What audio should the viewer hear?")
+        .items(&items)
+        .default(0)
+        .interact()?;
+
+    if choice == 0 {
+        Ok(None)
+    } else {
+        Ok(Some(apps[choice - 1].name.clone()))
+    }
+}
+
+/// `pixelpass --reconfigure` entry point: unconditionally re-run the
+/// bandwidth pre-flight test, save the result, and return. Used to
+/// refresh a stale measurement (e.g. user moved house, changed ISP).
+pub async fn run_reconfigure() -> Result<()> {
+    eprintln!();
+    eprintln!("Re-running bandwidth pre-flight test…");
+    let mut cfg = config::load().unwrap_or_default();
+    run_bandwidth_test(&mut cfg).await;
+    Ok(())
+}
+
+/// First-run pre-flight gate. Called once, when the user picks "Host" in
+/// the interactive menu. Behavior by saved status:
+/// - Unmeasured (first ever launch): explain + offer Run / Skip
+/// - Failed (previous attempt errored): offer Retry / give-up-and-skip
+/// - Measured or Skipped: silent — never re-prompts
+async fn preflight_if_needed(theme: &ColorfulTheme) {
+    let mut cfg = config::load().unwrap_or_default();
+    match cfg.bandwidth.status {
+        config::BandwidthStatus::Measured | config::BandwidthStatus::Skipped => return,
+        config::BandwidthStatus::Unmeasured => {
+            eprintln!();
+            eprintln!("First-time setup");
+            eprintln!("────────────────");
+            eprintln!("PixelPass can measure your upload speed to recommend a safe");
+            eprintln!("default for how many viewers your connection can handle.");
+            eprintln!("The test takes about 5 seconds and uploads ~5 MB to");
+            eprintln!("Cloudflare's open speed-test endpoint.");
+            eprintln!();
+            eprintln!("If you skip, a conservative default (2 viewers) is used.");
+            eprintln!("You can run the test later with `pixelpass --reconfigure`.");
+            eprintln!();
+
+            let Ok(choice) = Select::with_theme(theme)
+                .with_prompt("What would you like to do?")
+                .items(&[
+                    "Run the bandwidth test (recommended)",
+                    "Skip — use the conservative default",
+                ])
+                .default(0)
+                .interact()
+            else {
+                return;
+            };
+
+            if choice == 1 {
+                cfg.bandwidth = config::BandwidthEntry {
+                    status: config::BandwidthStatus::Skipped,
+                    upstream_mbps: None,
+                    measured_at: None,
+                };
+                let _ = config::save(&cfg);
+                eprintln!("Pre-flight skipped.");
+                return;
+            }
+            run_bandwidth_test(&mut cfg).await;
+        }
+        config::BandwidthStatus::Failed => {
+            eprintln!();
+            let Ok(choice) = Select::with_theme(theme)
+                .with_prompt("Last bandwidth test failed. Try again?")
+                .items(&[
+                    "Yes — retry now",
+                    "No — use the conservative default",
+                ])
+                .default(0)
+                .interact()
+            else {
+                return;
+            };
+
+            if choice == 1 {
+                cfg.bandwidth = config::BandwidthEntry {
+                    status: config::BandwidthStatus::Skipped,
+                    upstream_mbps: None,
+                    measured_at: None,
+                };
+                let _ = config::save(&cfg);
+                eprintln!("OK — using the conservative default.");
+                return;
+            }
+            run_bandwidth_test(&mut cfg).await;
+        }
+    }
+}
+
+async fn run_bandwidth_test(cfg: &mut config::Config) {
+    eprintln!();
+    eprintln!("Measuring upstream…");
+
+    let result = tokio::task::spawn_blocking(bandwidth::measure_upstream_blocking).await;
+    let measurement = match result {
+        Ok(Ok(m)) => m,
+        Ok(Err(e)) => {
+            eprintln!("Test failed: {e:#}");
+            eprintln!("Marking as failed — you'll be asked again on next launch.");
+            cfg.bandwidth = config::BandwidthEntry {
+                status: config::BandwidthStatus::Failed,
+                upstream_mbps: None,
+                measured_at: None,
+            };
+            let _ = config::save(cfg);
+            return;
+        }
+        Err(join_err) => {
+            eprintln!("Test task panicked: {join_err}");
+            cfg.bandwidth = config::BandwidthEntry {
+                status: config::BandwidthStatus::Failed,
+                upstream_mbps: None,
+                measured_at: None,
+            };
+            let _ = config::save(cfg);
+            return;
+        }
+    };
+
+    eprintln!(
+        "Measured {:.2} Mbps up (safe estimate {:.2} Mbps, took {:.1}s).",
+        measurement.raw_mbps,
+        measurement.safe_mbps,
+        measurement.elapsed.as_secs_f64()
+    );
+    cfg.bandwidth = config::BandwidthEntry {
+        status: config::BandwidthStatus::Measured,
+        upstream_mbps: Some(measurement.safe_mbps),
+        measured_at: Some(chrono::Utc::now()),
+    };
+    if let Err(e) = config::save(cfg) {
+        eprintln!("Warning: failed to save result: {e:#}");
+    }
+}
+
 fn print_welcome() {
     eprintln!();
     eprintln!("Welcome to PixelPass.");
@@ -68,14 +252,52 @@ impl Player {
                     url,
                 ],
             ),
-            Player::Vlc => crate::common::process::spawn_detached(
+            Player::Vlc => {
-                "vlc",
+                warn_if_vlc_plugins_missing();
-                &["--network-caching=200", "--live-caching=200", url],
+                crate::common::process::spawn_detached(
-            ),
+                    "vlc",
+                    &["--network-caching=200", "--live-caching=200", url],
+                )
+            }
         }
     }
 }
 
+// On Arch-family distros, the base `vlc` package omits two plugins
+// pixelpass needs: the MPEG-TS demuxer (`vlc-plugin-dvb`) and the
+// libavcodec-based H.264 decoder (`vlc-plugin-ffmpeg`). Missing either
+// produces a confusing error chain — warn at launch.
+fn warn_if_vlc_plugins_missing() {
+    const REQUIRED: &[(&str, &str)] = &[
+        (
+            "/usr/lib/vlc/plugins/demux/libts_plugin.so",
+            "vlc-plugin-dvb",
+        ),
+        (
+            "/usr/lib/vlc/plugins/codec/libavcodec_plugin.so",
+            "vlc-plugin-ffmpeg",
+        ),
+    ];
+    let missing: Vec<&(&str, &str)> = REQUIRED
+        .iter()
+        .filter(|(p, _)| !std::path::Path::new(p).exists())
+        .collect();
+    if missing.is_empty() {
+        return;
+    }
+    eprintln!();
+    eprintln!("Warning: VLC is missing plugins pixelpass needs:");
+    for (path, pkg) in &missing {
+        eprintln!("  - {path} (install `{pkg}`)");
+    }
+    eprintln!("On Arch / CachyOS / EndeavourOS: `sudo pacman -S {}`.", {
+        let names: Vec<&str> = missing.iter().map(|(_, p)| *p).collect();
+        names.join(" ")
+    });
+    eprintln!("mpv is unaffected.");
+    eprintln!();
+}
+
 pub fn prompt_player() -> Result<Player> {
     let theme = ColorfulTheme::default();
     let choice = Select::with_theme(&theme)

src/main.rs

@@ -16,10 +16,18 @@ async fn main() -> Result<()> {
     let cli = Cli::parse();
     init_tracing(cli.verbose);
 
+    // libpipewire requires global init before any pw_* call. Idempotent;
+    // safe to call even when the per-app audio thread never spawns.
+    pipewire::init();
+
     if cli.repair {
         return repair::run().await;
     }
 
+    if cli.reconfigure {
+        return interactive::run_reconfigure().await;
+    }
+
     match cli.ticket.as_deref() {
         Some(s) => {
             let ticket: EndpointTicket = s.parse().map_err(|e| {

src/repair.rs

@@ -1,12 +1,197 @@
-//! `--repair`: clean up any null sinks / loopbacks that a crashed pixelpass
+//! `--repair`: clean up null-sinks and loopbacks left behind by a crashed
-//! host left behind. Phase 2 will scan PipeWire for nodes tagged with the
+//! pixelpass host. Identifies orphans by the `pixelpass_capture_<pid>`
-//! `pixelpass.session = <uuid>` property and destroy them.
+//! name pattern + dead-PID check, then unloads paired loopbacks first
+//! (mirrors `Routing::shutdown`'s order so PipeWire doesn't leave zombie
+//! links). Live PIDs — including this process and any other running
+//! pixelpass — are left alone.
 
-use anyhow::Result;
+use anyhow::{Context, Result, bail};
+use std::collections::HashSet;
+use std::path::Path;
+use std::process::Command;
+
+const SINK_NAME_PREFIX: &str = "pixelpass_capture_";
 
 pub async fn run() -> Result<()> {
-    eprintln!("[pixelpass] --repair: PipeWire scan not yet implemented (Phase 2).");
+    let modules = list_modules().context("failed to list pactl modules")?;
-    eprintln!("            Run `pactl list short sinks | grep pixelpass` to spot orphans,");
+
-    eprintln!("            and `pactl unload-module <id>` to remove them manually.");
+    let mut dead_sinks: Vec<OrphanSink> = Vec::new();
+    let mut dead_pids: HashSet<u32> = HashSet::new();
+    let mut live_skipped: u32 = 0;
+
+    for m in &modules {
+        if m.name != "module-null-sink" {
+            continue;
+        }
+        let Some(sink_name) = extract_kv(&m.args, "sink_name") else {
+            continue;
+        };
+        let Some(pid_str) = sink_name.strip_prefix(SINK_NAME_PREFIX) else {
+            continue;
+        };
+        let Ok(pid) = pid_str.parse::<u32>() else {
+            continue;
+        };
+
+        if is_pid_alive(pid) {
+            live_skipped += 1;
+            continue;
+        }
+        dead_pids.insert(pid);
+        dead_sinks.push(OrphanSink {
+            id: m.id,
+            sink_name: sink_name.to_string(),
+            pid,
+        });
+    }
+
+    let mut dead_loopbacks: Vec<u32> = Vec::new();
+    for m in &modules {
+        if m.name != "module-loopback" {
+            continue;
+        }
+        let Some(sink) = extract_kv(&m.args, "sink") else {
+            continue;
+        };
+        let Some(pid_str) = sink.strip_prefix(SINK_NAME_PREFIX) else {
+            continue;
+        };
+        let Ok(pid) = pid_str.parse::<u32>() else {
+            continue;
+        };
+        if dead_pids.contains(&pid) {
+            dead_loopbacks.push(m.id);
+        }
+    }
+
+    if dead_sinks.is_empty() && dead_loopbacks.is_empty() {
+        if live_skipped > 0 {
+            println!(
+                "[pixelpass] --repair: nothing to clean up ({live_skipped} live pixelpass host(s) left alone)."
+            );
+        } else {
+            println!("[pixelpass] --repair: nothing to clean up.");
+        }
+        return Ok(());
+    }
+
+    let mut unloaded = 0u32;
+    let mut failed = 0u32;
+
+    for id in &dead_loopbacks {
+        match unload_module(*id) {
+            Ok(()) => {
+                println!("[pixelpass] --repair: unloaded loopback module #{id}");
+                unloaded += 1;
+            }
+            Err(e) => {
+                eprintln!("[pixelpass] --repair: failed to unload loopback #{id}: {e:#}");
+                failed += 1;
+            }
+        }
+    }
+    for orphan in &dead_sinks {
+        match unload_module(orphan.id) {
+            Ok(()) => {
+                println!(
+                    "[pixelpass] --repair: unloaded {} (orphaned from pid {})",
+                    orphan.sink_name, orphan.pid
+                );
+                unloaded += 1;
+            }
+            Err(e) => {
+                eprintln!(
+                    "[pixelpass] --repair: failed to unload {} (#{}): {e:#}",
+                    orphan.sink_name, orphan.id
+                );
+                failed += 1;
+            }
+        }
+    }
+
+    if live_skipped > 0 {
+        println!(
+            "[pixelpass] --repair: left {live_skipped} live pixelpass host(s) alone."
+        );
+    }
+
+    if failed > 0 {
+        bail!("--repair: {failed} module(s) failed to unload (see errors above)");
+    }
+    println!("[pixelpass] --repair: cleaned up {unloaded} module(s).");
+    Ok(())
+}
+
+struct Module {
+    id: u32,
+    name: String,
+    args: String,
+}
+
+struct OrphanSink {
+    id: u32,
+    sink_name: String,
+    pid: u32,
+}
+
+fn list_modules() -> Result<Vec<Module>> {
+    let output = Command::new("pactl")
+        .args(["list", "short", "modules"])
+        .output()
+        .context("failed to run `pactl list short modules`")?;
+    if !output.status.success() {
+        bail!(
+            "pactl list short modules failed: {}",
+            String::from_utf8_lossy(&output.stderr).trim()
+        );
+    }
+    let text = String::from_utf8(output.stdout).context("pactl returned non-UTF-8")?;
+    let mut modules = Vec::new();
+    // `pactl list short modules` is tab-separated, but some modules have
+    // multi-line `{ ... }` argument blocks that wrap onto continuation
+    // lines starting with whitespace. The wrap lines never parse as a
+    // u32 ID, so the simple per-line + parse-id filter is robust.
+    for line in text.lines() {
+        let mut parts = line.splitn(4, '\t');
+        let Some(id_str) = parts.next() else { continue };
+        let Ok(id) = id_str.parse::<u32>() else { continue };
+        let Some(name) = parts.next() else { continue };
+        let args = parts.next().unwrap_or("").to_string();
+        modules.push(Module {
+            id,
+            name: name.to_string(),
+            args,
+        });
+    }
+    Ok(modules)
+}
+
+fn extract_kv<'a>(args: &'a str, key: &str) -> Option<&'a str> {
+    for token in args.split_whitespace() {
+        if let Some(rest) = token.strip_prefix(key)
+            && let Some(value) = rest.strip_prefix('=')
+        {
+            return Some(value);
+        }
+    }
+    None
+}
+
+fn is_pid_alive(pid: u32) -> bool {
+    Path::new(&format!("/proc/{pid}")).exists()
+}
+
+fn unload_module(id: u32) -> Result<()> {
+    let output = Command::new("pactl")
+        .arg("unload-module")
+        .arg(id.to_string())
+        .output()
+        .context("failed to run pactl unload-module")?;
+    if !output.status.success() {
+        bail!(
+            "pactl unload-module #{id}: {}",
+            String::from_utf8_lossy(&output.stderr).trim()
+        );
+    }
     Ok(())
 }