host/taint: correct the buffered-echo scoping (in-threat-model); pin ambiguous-client (Codex round 6)

Codex refuted my round-5 disposition and was right: the buffered-echo gap
is NOT limited to keyless/unbounded readers. A normal PID-bearing app —
recorder, DAW, GStreamer — can read the call, buffer it in application
memory, fully tear down its PipeWire Node *and* Client, then (still the
same live process) open a fresh Client + output and replay. `seed_sticky`
drops the PID fingerprint once every old serial is gone, so the replayed
leg is Eligible. That is in-threat-model, so my "outside the threat model"
claim was false.

- Rewrote the module-doc gap note honestly: in-threat-model, reachable by
  non-adversarial software, sitting on the design's §6.1.3 "full teardown
  ⇒ starts clean" boundary. Framed the two options — (A) accept as a
  documented v1 limitation, (B) process-generation lifetime (PID + /proc
  start-time, phase 3 supplies liveness, §6.1.3 revised). This is a
  designer's decision (it revises the security surface); NOT resolved in
  code. `a_fingerprint_does_not_outlive_its_owner` currently encodes
  Option A and flips under B.
- P2 (fixed): pinned the ambiguous-client-id branch. A mutation
  remembering only the first of two clients claiming one global id
  survived the suite; added a test scoped to the ambiguous owner (the
  global count was masked by the peerspeak owner's client). Verified the
  `.next()` mutation now fails it.

57 tests. Phase 2 is NOT converged — the buffered-echo design decision is
owed to the user before merge.

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

Cargo.toml

@@ -32,7 +32,7 @@ name = "pixelpass"
 path = "src/main.rs"
 
 [dependencies]
-iroh = "1.0.0-rc.0"
+iroh = "1.0.2"
 tokio = { version = "1", features = ["macros", "rt-multi-thread", "io-util", "net", "signal", "process", "sync", "time"] }
 tokio-util = { version = "0.7", features = ["io"] }
 clap = { version = "4", features = ["derive"] }
@@ -48,7 +48,7 @@ ashpd = { version = "0.9", default-features = false, features = ["tokio"] }
 pipewire = "0.9"
 x11rb = { version = "0.13", default-features = false, features = ["allow-unsafe-code"] }
 uuid = { version = "1", features = ["v4"] }
-iroh-tickets = "1.0.0-rc.0"
+iroh-tickets = "1.0.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"] }

README.md

@@ -23,6 +23,8 @@ Working:
 - Audio capture of the default sink's monitor, with optional per-app
   routing (`--app <name>`)
 - `--repair` cleanup of orphaned PipeWire state left by a crashed host
+- `--doctor` environment diagnostic (capture/encode deps, VA-API H.264,
+  viewer player, relay reachability) — see [Diagnostics](#diagnostics)
 - 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>`)
@@ -135,6 +137,35 @@ 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.
 
+## Diagnostics
+
+`pixelpass --doctor` prints a one-shot report of everything the above
+requirements cover and exits — run it on any machine before a real session:
+
+```sh
+pixelpass --doctor
+```
+
+It checks, and prints a `✓ / ! / ✗` line for each:
+
+- **display server** — Wayland vs. X11 (autodetected), the raw session env
+  vars, and the X server's vendor/version (so an xlibre server is visible)
+- **capture** — the GStreamer tools plus the source element for your backend
+  (`pipewiresrc` on Wayland, `ximagesrc` on X11)
+- **encode** — whether hardware H.264 works (the `vah264enc` plugin, a DRM
+  render node, and a VA-API H.264 *encode* entrypoint via `vainfo`), and
+  whether the software `x264enc` fallback is available. This is the usual
+  culprit when a viewer "can't connect": a GPU with no H.264 encode entrypoint
+  produces no video under the default encoder — the report tells you to host
+  with `--no-hwencode`
+- **mux / audio** — the TS mux + AAC + PulseAudio tail, and `pactl`
+- **viewer** — whether `mpv` or `vlc` is installed
+- **network** — binds a real endpoint and checks a relay is reachable
+
+Each failing line includes a distro-aware install hint, and the closing summary
+says whether the machine can host and how. The exit code is non-zero if any
+hard requirement is missing, so it can gate a script or CI.
+
 ## Build
 
 ```sh

src/cli.rs

@@ -105,6 +105,14 @@ pub struct Cli {
     #[arg(long)]
     pub repair: bool,
 
+    /// Print an environment diagnostic report (display server, capture/encode
+    /// dependencies, VA-API H.264 support, viewer player, relay reachability),
+    /// then exit. Use this to check a machine can host or view before a real
+    /// session — especially to confirm hardware H.264 encode works, since a GPU
+    /// without it silently produces no video under the default encoder.
+    #[arg(long)]
+    pub doctor: 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.

src/common/deps.rs

@@ -56,12 +56,7 @@ fn require(bin: &str) -> Result<PathBuf> {
 }
 
 fn require_gst_element(name: &str) -> Result<()> {
-    let ok = Command::new("gst-inspect-1.0")
+    if !gst_element_exists(name) {
-        .args(["--exists", name])
-        .status()
-        .map(|s| s.success())
-        .unwrap_or(false);
-    if !ok {
         bail!(
             "GStreamer element `{name}` not available.\n{}",
             install_hint_for_gst_element(name)
@@ -70,7 +65,17 @@ fn require_gst_element(name: &str) -> Result<()> {
     Ok(())
 }
 
-fn which(bin: &str) -> Option<PathBuf> {
+/// Whether a GStreamer element is registered, via `gst-inspect-1.0 --exists`.
+/// Non-bailing counterpart to [`require_gst_element`] for the `doctor` report.
+pub(crate) fn gst_element_exists(name: &str) -> bool {
+    Command::new("gst-inspect-1.0")
+        .args(["--exists", name])
+        .status()
+        .map(|s| s.success())
+        .unwrap_or(false)
+}
+
+pub(crate) fn which(bin: &str) -> Option<PathBuf> {
     let path = std::env::var_os("PATH")?;
     for dir in std::env::split_paths(&path) {
         let candidate = dir.join(bin);
@@ -81,7 +86,7 @@ fn which(bin: &str) -> Option<PathBuf> {
     None
 }
 
-fn install_hint_for_bin(bin: &str) -> String {
+pub(crate) fn install_hint_for_bin(bin: &str) -> String {
     let distro = detect_distro();
     let pkg = match bin {
         "gst-launch-1.0" | "gst-inspect-1.0" => match distro.as_deref() {
@@ -113,7 +118,7 @@ fn install_hint_for_bin(bin: &str) -> String {
     install_command(&distro, pkg)
 }
 
-fn install_hint_for_gst_element(name: &str) -> String {
+pub(crate) fn install_hint_for_gst_element(name: &str) -> String {
     let distro = detect_distro();
     let pkg = match name {
         "pipewiresrc" => match distro.as_deref() {
@@ -210,7 +215,7 @@ fn install_command(distro: &Option<String>, pkg: &str) -> String {
     format!("Install hint: {cmd}")
 }
 
-fn detect_distro() -> Option<String> {
+pub(crate) fn detect_distro() -> Option<String> {
     let contents = std::fs::read_to_string("/etc/os-release").ok()?;
     for line in contents.lines() {
         if let Some(rest) = line.strip_prefix("ID=") {

src/doctor.rs

@@ -0,0 +1,648 @@
+//! `pixelpass doctor` — environment diagnostics.
+//!
+//! Screen-share failures are usually not pixelpass bugs but environment gaps:
+//! a missing GStreamer plugin, an X vs. Wayland mismatch, or — the common one —
+//! a GPU/driver with no working VA-API H.264 encoder, so the default
+//! `vah264enc` pipeline never produces a byte and the viewer "can't connect."
+//! `doctor` probes all of that up front and prints one actionable report, so a
+//! remote tester can read it over a call instead of us guessing from logs. It
+//! also validates any X11/Wayland test environment we stand up.
+//!
+//! Unlike [`crate::common::deps::check_host_binaries`], which bails on the first
+//! missing dependency, doctor runs *every* check and reports them together — a
+//! diagnostic wants the whole picture, not the first failure.
+
+use anyhow::Result;
+use std::time::Duration;
+
+use crate::common::deps;
+use crate::common::display::DisplayServer;
+use crate::common::endpoint;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Status {
+    /// Working as needed.
+    Ok,
+    /// Degraded but not fatal (e.g. a fallback path is available).
+    Warn,
+    /// Screen-sharing will not work until this is fixed.
+    Fail,
+    /// Neutral fact, no judgement.
+    Info,
+}
+
+impl Status {
+    fn icon(self) -> char {
+        match self {
+            Self::Ok => '✓',
+            Self::Warn => '!',
+            Self::Fail => '✗',
+            Self::Info => '·',
+        }
+    }
+}
+
+/// One line in the report: a status, a short label, a detail, and an optional
+/// remediation hint printed on its own indented line.
+pub struct Check {
+    pub status: Status,
+    pub label: String,
+    pub detail: String,
+    pub hint: Option<String>,
+}
+
+impl Check {
+    fn new(status: Status, label: impl Into<String>, detail: impl Into<String>) -> Self {
+        Self {
+            status,
+            label: label.into(),
+            detail: detail.into(),
+            hint: None,
+        }
+    }
+
+    fn ok(label: impl Into<String>, detail: impl Into<String>) -> Self {
+        Self::new(Status::Ok, label, detail)
+    }
+    fn warn(label: impl Into<String>, detail: impl Into<String>) -> Self {
+        Self::new(Status::Warn, label, detail)
+    }
+    fn fail(label: impl Into<String>, detail: impl Into<String>) -> Self {
+        Self::new(Status::Fail, label, detail)
+    }
+    fn info(label: impl Into<String>, detail: impl Into<String>) -> Self {
+        Self::new(Status::Info, label, detail)
+    }
+
+    fn with_hint(mut self, hint: impl Into<String>) -> Self {
+        self.hint = Some(hint.into());
+        self
+    }
+}
+
+/// Tally of the non-trivial statuses across every section.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub struct Summary {
+    pub fails: usize,
+    pub warns: usize,
+}
+
+/// A named group of checks, printed under a header.
+struct Section {
+    name: &'static str,
+    checks: Vec<Check>,
+}
+
+/// Run all diagnostics and print the report. Always prints; the process exit
+/// code is non-zero only when a hard failure (a `Fail`) was found, so scripts
+/// and CI can gate on it while a human still sees everything.
+pub async fn run(relay: Option<String>) -> Result<()> {
+    let display = DisplayServer::detect();
+
+    let sections = vec![
+        system_section(display),
+        capture_section(display),
+        encode_section(),
+        mux_audio_section(),
+        viewer_section(),
+        network_section(relay.as_deref()).await,
+    ];
+
+    print_report(&sections);
+
+    let summary = summarize(sections.iter().flat_map(|s| s.checks.iter()));
+    print_summary(summary, &sections);
+
+    if summary.fails > 0 {
+        std::process::exit(1);
+    }
+    Ok(())
+}
+
+// ── sections ──────────────────────────────────────────────────────────────
+
+fn system_section(display: DisplayServer) -> Section {
+    let mut checks = vec![
+        Check::info(
+            "pixelpass",
+            format!("{} (gui: {})", env!("CARGO_PKG_VERSION"), gui_built()),
+        ),
+        Check::info("distro", distro_detail()),
+        display_check(display),
+    ];
+
+    // Probe the actual X server when one is reachable — this is where an xlibre
+    // vs. Xorg difference (the thing we most want to see on a tester's box)
+    // shows up. Skip it on a pure Wayland session with no X at all.
+    if display == DisplayServer::X11 || std::env::var_os("DISPLAY").is_some() {
+        checks.push(x_server_check());
+    }
+
+    Section {
+        name: "System",
+        checks,
+    }
+}
+
+fn capture_section(display: DisplayServer) -> Section {
+    let mut checks = vec![
+        bin_check("gst-launch-1.0", "gstreamer tools"),
+        bin_check("gst-inspect-1.0", "gstreamer tools"),
+    ];
+
+    match display {
+        DisplayServer::Wayland => {
+            checks.push(gst_check("pipewiresrc", "Wayland capture"));
+        }
+        DisplayServer::X11 => {
+            checks.push(gst_check("ximagesrc", "X11 capture"));
+            checks.push(match deps::which("xwininfo") {
+                Some(p) => Check::ok("window picker", p.display().to_string())
+                    .with_hint("needed only for `--window` (share a single window)"),
+                None => Check::info("window picker", "xwininfo not found")
+                    .with_hint("optional — only `--window` needs it"),
+            });
+        }
+        DisplayServer::Unknown => {
+            checks.push(
+                Check::info("capture backend", "unknown — cannot probe a source element")
+                    .with_hint("force one with `--display-server x11|wayland` when hosting"),
+            );
+        }
+    }
+
+    Section {
+        name: "Capture (host)",
+        checks,
+    }
+}
+
+fn encode_section() -> Section {
+    Section {
+        name: "Encode",
+        checks: vec![hardware_encode_check(), software_encode_check()],
+    }
+}
+
+/// The load-bearing check for the common "viewer can't connect" report: the
+/// default host pipeline uses `vah264enc`, which needs both the GStreamer VA
+/// plugin *and* a GPU/driver that actually exposes an H.264 encode entrypoint.
+/// A box with the plugin but no encode entrypoint (or no render node) produces
+/// no video — the exact silent failure `--no-hwencode` works around.
+fn hardware_encode_check() -> Check {
+    if !deps::gst_element_exists("vah264enc") {
+        return Check::warn("hardware H.264", "vah264enc plugin not installed").with_hint(format!(
+            "{} — or just host with `--no-hwencode` (software x264)",
+            deps::install_hint_for_gst_element("vah264enc")
+        ));
+    }
+
+    if !has_render_node() {
+        return Check::warn(
+            "hardware H.264",
+            "vah264enc present, but no DRM render node (/dev/dri/renderD*)",
+        )
+        .with_hint("GPU encode is unavailable here — host with `--no-hwencode`");
+    }
+
+    match vainfo_output() {
+        Some(out) if vainfo_has_h264_encode(&out) => Check::ok(
+            "hardware H.264",
+            "VA-API H.264 encode available (vah264enc)",
+        ),
+        Some(_) => Check::warn(
+            "hardware H.264",
+            "vah264enc present, but VA-API reports no H.264 encode entrypoint",
+        )
+        .with_hint("this GPU/driver can't hardware-encode H.264 — host with `--no-hwencode`"),
+        None => Check::info(
+            "hardware H.264",
+            "vah264enc + render node present; couldn't confirm the VA-API encode entrypoint",
+        )
+        .with_hint("install `vainfo` (libva-utils) to verify, or just test a real host session"),
+    }
+}
+
+fn software_encode_check() -> Check {
+    if deps::gst_element_exists("x264enc") {
+        Check::ok("software H.264", "x264enc available (`--no-hwencode`)")
+    } else {
+        Check::warn("software H.264", "x264enc not installed").with_hint(format!(
+            "{} — the fallback for GPUs without VA-API H.264 encode",
+            deps::install_hint_for_gst_element("x264enc")
+        ))
+    }
+}
+
+fn mux_audio_section() -> Section {
+    // These live in plugins-bad/-good/-libav and plugins-base; all are required
+    // for either backend, so a miss here is a hard Fail.
+    let tail = [
+        "h264parse",
+        "mpegtsmux",
+        "aacparse",
+        "avenc_aac",
+        "pulsesrc",
+        "videoscale",
+    ];
+    let missing: Vec<&str> = tail
+        .iter()
+        .copied()
+        .filter(|e| !deps::gst_element_exists(e))
+        .collect();
+
+    let tail_check = if missing.is_empty() {
+        Check::ok("mux + audio tail", tail.join(", "))
+    } else {
+        Check::fail(
+            "mux + audio tail",
+            format!("missing: {}", missing.join(", ")),
+        )
+        .with_hint(deps::install_hint_for_gst_element(missing[0]))
+    };
+
+    Section {
+        name: "Mux / audio",
+        checks: vec![tail_check, bin_check("pactl", "pactl")],
+    }
+}
+
+fn viewer_section() -> Section {
+    let mpv = deps::which("mpv");
+    let vlc = deps::which("vlc");
+
+    let check = match (mpv, vlc) {
+        (Some(p), _) => Check::ok("player", format!("mpv ({})", p.display())),
+        (None, Some(p)) => Check::ok("player", format!("vlc ({})", p.display()))
+            .with_hint("mpv is the recommended player; vlc needs the dvb + ffmpeg plugins"),
+        (None, None) => Check::warn("player", "neither mpv nor vlc found")
+            .with_hint("a viewer needs one of them; the GUI launches mpv by default"),
+    };
+
+    Section {
+        name: "Viewer",
+        checks: vec![check],
+    }
+}
+
+/// Bind a real video-plane endpoint and wait briefly for a relay, mirroring
+/// what a host does. Directly relevant to "couldn't connect": if this machine
+/// can't reach a relay, hole-punching to a peer is unlikely to work either.
+async fn network_section(relay: Option<&str>) -> Section {
+    let check = match endpoint::bind(relay).await {
+        Ok(ep) => {
+            let online = tokio::time::timeout(Duration::from_secs(8), ep.online())
+                .await
+                .is_ok();
+            let relay_count = ep.addr().addrs.iter().filter(|a| a.is_relay()).count();
+            let where_ = relay.map(|r| format!(" ({r})")).unwrap_or_default();
+            // Close gracefully so iroh doesn't log a scary "Endpoint dropped
+            // without calling close" error into the middle of the report.
+            ep.close().await;
+
+            if online && relay_count > 0 {
+                Check::ok("relay", format!("home relay reachable{where_}"))
+            } else if online {
+                Check::warn(
+                    "relay",
+                    format!("endpoint online but no relay address{where_}"),
+                )
+                .with_hint(
+                    "n0 DNS discovery may still connect peers, but relay fallback is degraded",
+                )
+            } else {
+                Check::warn("relay", format!("no relay connected within 8s{where_}")).with_hint(
+                    "check connectivity/firewall; peers behind NAT rely on the relay to rendezvous",
+                )
+            }
+        }
+        Err(e) => Check::fail("relay", format!("could not bind endpoint: {e}")),
+    };
+
+    Section {
+        name: "Network",
+        checks: vec![check],
+    }
+}
+
+// ── small check builders ────────────────────────────────────────────────────
+
+fn bin_check(bin: &str, label: &str) -> Check {
+    match deps::which(bin) {
+        Some(p) => Check::ok(label, format!("{bin} ({})", p.display())),
+        None => Check::fail(label, format!("{bin} not found on PATH"))
+            .with_hint(deps::install_hint_for_bin(bin)),
+    }
+}
+
+fn gst_check(element: &str, label: &str) -> Check {
+    if deps::gst_element_exists(element) {
+        Check::ok(label, element.to_string())
+    } else {
+        Check::fail(
+            label,
+            format!("GStreamer element `{element}` not available"),
+        )
+        .with_hint(deps::install_hint_for_gst_element(element))
+    }
+}
+
+fn display_check(display: DisplayServer) -> Check {
+    let env = display_env_summary();
+    match display {
+        DisplayServer::Wayland => Check::ok("display server", format!("Wayland ({env})")),
+        DisplayServer::X11 => Check::ok("display server", format!("X11 ({env})")),
+        DisplayServer::Unknown => Check::fail("display server", format!("undetected ({env})"))
+            .with_hint(
+                "no WAYLAND_DISPLAY/DISPLAY/XDG_SESSION_TYPE — capture can't start; \
+                 run inside a graphical session or pass `--display-server`",
+            ),
+    }
+}
+
+/// Connect to the X server and report its vendor + version. This is how an
+/// xlibre server distinguishes itself from stock Xorg (vendor string / release
+/// number), which is exactly what we want to see on a tester's machine.
+fn x_server_check() -> Check {
+    use x11rb::connection::Connection;
+    match x11rb::connect(None) {
+        Ok((conn, _screen)) => {
+            let setup = conn.setup();
+            let vendor = String::from_utf8_lossy(&setup.vendor);
+            let detail = format!(
+                "vendor \"{}\", protocol {}.{}, release {}",
+                vendor.trim(),
+                setup.protocol_major_version,
+                setup.protocol_minor_version,
+                setup.release_number,
+            );
+            let label = "X server";
+            if vendor.to_lowercase().contains("xlibre") {
+                Check::info(label, format!("XLibre — {detail}"))
+            } else {
+                Check::info(label, detail)
+            }
+        }
+        Err(_) => Check::info("X server", "DISPLAY set but the X server is unreachable"),
+    }
+}
+
+// ── environment helpers ─────────────────────────────────────────────────────
+
+fn gui_built() -> &'static str {
+    if cfg!(feature = "gui") { "yes" } else { "no" }
+}
+
+fn distro_detail() -> String {
+    let id = deps::detect_distro();
+    let pretty = os_release_field("PRETTY_NAME");
+    match (id, pretty) {
+        (Some(id), Some(p)) => format!("{id} ({p})"),
+        (Some(id), None) => id,
+        (None, Some(p)) => p,
+        (None, None) => "unknown".to_string(),
+    }
+}
+
+fn os_release_field(key: &str) -> Option<String> {
+    let contents = std::fs::read_to_string("/etc/os-release").ok()?;
+    for line in contents.lines() {
+        if let Some(rest) = line.strip_prefix(&format!("{key}=")) {
+            return Some(rest.trim_matches('"').to_string());
+        }
+    }
+    None
+}
+
+fn display_env_summary() -> String {
+    let mut parts = Vec::new();
+    for var in [
+        "WAYLAND_DISPLAY",
+        "DISPLAY",
+        "XDG_SESSION_TYPE",
+        "XDG_CURRENT_DESKTOP",
+    ] {
+        if let Some(v) = std::env::var_os(var) {
+            parts.push(format!("{var}={}", v.to_string_lossy()));
+        }
+    }
+    if parts.is_empty() {
+        "no display env vars set".to_string()
+    } else {
+        parts.join(", ")
+    }
+}
+
+fn has_render_node() -> bool {
+    let Ok(entries) = std::fs::read_dir("/dev/dri") else {
+        return false;
+    };
+    entries
+        .flatten()
+        .any(|e| e.file_name().to_string_lossy().starts_with("renderD"))
+}
+
+fn vainfo_output() -> Option<String> {
+    deps::which("vainfo")?;
+    let out = std::process::Command::new("vainfo").output().ok()?;
+    // vainfo prints its profile/entrypoint table to stdout; some builds also
+    // spill driver banners to stderr. Concatenate both so parsing is robust.
+    let mut s = String::from_utf8_lossy(&out.stdout).into_owned();
+    s.push_str(&String::from_utf8_lossy(&out.stderr));
+    Some(s)
+}
+
+/// Pure: does a `vainfo` dump advertise an H.264 *encode* entrypoint? vainfo
+/// lists one `VAProfile… : VAEntrypoint…` pair per line; hardware H.264 encode
+/// is any `VAProfileH264*` profile paired with an `EncSlice`/`EncSliceLP`
+/// entrypoint. VLD-only H.264 (decode) does not count.
+fn vainfo_has_h264_encode(output: &str) -> bool {
+    output.lines().any(|line| {
+        line.contains("VAProfileH264")
+            && (line.contains("VAEntrypointEncSlice") || line.contains("VAEntrypointEncSliceLP"))
+    })
+}
+
+// ── reporting ───────────────────────────────────────────────────────────────
+
+fn print_report(sections: &[Section]) {
+    println!("pixelpass doctor\n");
+    for section in sections {
+        println!("{}", section.name);
+        for check in &section.checks {
+            println!(
+                "  {} {:<16}  {}",
+                check.status.icon(),
+                check.label,
+                check.detail
+            );
+            if let Some(hint) = &check.hint {
+                println!("        → {hint}");
+            }
+        }
+        println!();
+    }
+}
+
+fn summarize<'a>(checks: impl Iterator<Item = &'a Check>) -> Summary {
+    let mut summary = Summary::default();
+    for check in checks {
+        match check.status {
+            Status::Fail => summary.fails += 1,
+            Status::Warn => summary.warns += 1,
+            Status::Ok | Status::Info => {}
+        }
+    }
+    summary
+}
+
+fn print_summary(summary: Summary, sections: &[Section]) {
+    let hosting = hosting_verdict(sections);
+    let counts = match (summary.fails, summary.warns) {
+        (0, 0) => "all checks passed".to_string(),
+        (0, w) => format!("{w} warning{}", plural(w)),
+        (f, 0) => format!("{f} failure{}", plural(f)),
+        (f, w) => format!("{f} failure{}, {w} warning{}", plural(f), plural(w)),
+    };
+    println!("Summary: {counts}. {hosting}");
+}
+
+fn plural(n: usize) -> &'static str {
+    if n == 1 { "" } else { "s" }
+}
+
+/// A one-line verdict on whether this box can host, and how. Reads the actual
+/// encode + capture checks rather than the raw tally so the advice is specific.
+fn hosting_verdict(sections: &[Section]) -> String {
+    let find = |section: &str, label: &str| -> Option<Status> {
+        sections
+            .iter()
+            .find(|s| s.name == section)?
+            .checks
+            .iter()
+            .find(|c| c.label == label)
+            .map(|c| c.status)
+    };
+
+    let hw = find("Encode", "hardware H.264");
+    let sw_ok = find("Encode", "software H.264") == Some(Status::Ok);
+    let capture_broken = sections
+        .iter()
+        .find(|s| s.name == "Capture (host)")
+        .map(|s| s.checks.iter().any(|c| c.status == Status::Fail))
+        .unwrap_or(false);
+
+    if capture_broken {
+        "Hosting will fail: the capture backend is incomplete (see Capture above).".to_string()
+    } else if hw == Some(Status::Ok) {
+        "Hosting will work (hardware H.264 encode).".to_string()
+    } else if hw == Some(Status::Info) && sw_ok {
+        // Plugin + render node present but VA-API unverified (no vainfo): the
+        // default encoder is likely fine; `--no-hwencode` is the safe fallback.
+        "Hosting should work (hardware H.264 likely; `--no-hwencode` is the fallback).".to_string()
+    } else if sw_ok {
+        "Hosting should work with `--no-hwencode` (software H.264 encode).".to_string()
+    } else {
+        "Hosting may fail: no working H.264 encoder found (see Encode above).".to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn vainfo_detects_h264_encode_entrypoint() {
+        // Realistic AMD/RADV-style dump: H.264 has both decode (VLD) and encode.
+        let dump = "\
+      VAProfileH264Main               : VAEntrypointVLD
+      VAProfileH264Main               : VAEntrypointEncSlice
+      VAProfileH264High               : VAEntrypointVLD
+      VAProfileHEVCMain               : VAEntrypointEncSlice";
+        assert!(vainfo_has_h264_encode(dump));
+    }
+
+    #[test]
+    fn vainfo_low_power_encode_counts() {
+        let dump = "VAProfileH264ConstrainedBaseline: VAEntrypointEncSliceLP";
+        assert!(vainfo_has_h264_encode(dump));
+    }
+
+    #[test]
+    fn vainfo_decode_only_h264_is_not_encode() {
+        // Decode-only H.264 (VLD) plus HEVC encode must NOT be read as H.264
+        // encode — this is exactly the "default encoder fails" case.
+        let dump = "\
+      VAProfileH264Main               : VAEntrypointVLD
+      VAProfileH264High               : VAEntrypointVLD
+      VAProfileHEVCMain               : VAEntrypointEncSlice";
+        assert!(!vainfo_has_h264_encode(dump));
+    }
+
+    #[test]
+    fn vainfo_empty_is_not_encode() {
+        assert!(!vainfo_has_h264_encode(""));
+    }
+
+    #[test]
+    fn summarize_counts_fails_and_warns_only() {
+        let checks = [
+            Check::ok("a", "x"),
+            Check::info("b", "x"),
+            Check::warn("c", "x"),
+            Check::warn("d", "x"),
+            Check::fail("e", "x"),
+        ];
+        let summary = summarize(checks.iter());
+        assert_eq!(summary, Summary { fails: 1, warns: 2 });
+    }
+
+    #[test]
+    fn hosting_verdict_prefers_hardware_then_software() {
+        let hw = vec![Section {
+            name: "Encode",
+            checks: vec![
+                Check::ok("hardware H.264", "ok"),
+                Check::ok("software H.264", "ok"),
+            ],
+        }];
+        assert!(hosting_verdict(&hw).contains("hardware"));
+
+        let sw = vec![Section {
+            name: "Encode",
+            checks: vec![
+                Check::warn("hardware H.264", "no"),
+                Check::ok("software H.264", "ok"),
+            ],
+        }];
+        assert!(sw_verdict_uses_no_hwencode(&hosting_verdict(&sw)));
+
+        let none = vec![Section {
+            name: "Encode",
+            checks: vec![
+                Check::warn("hardware H.264", "no"),
+                Check::warn("software H.264", "no"),
+            ],
+        }];
+        assert!(hosting_verdict(&none).contains("may fail"));
+    }
+
+    fn sw_verdict_uses_no_hwencode(v: &str) -> bool {
+        v.contains("--no-hwencode")
+    }
+
+    #[test]
+    fn capture_failure_dominates_verdict() {
+        let sections = vec![
+            Section {
+                name: "Capture (host)",
+                checks: vec![Check::fail("X11 capture", "missing")],
+            },
+            Section {
+                name: "Encode",
+                checks: vec![Check::ok("hardware H.264", "ok")],
+            },
+        ];
+        assert!(hosting_verdict(&sections).contains("capture"));
+    }
+}

src/host/audio.rs

@@ -365,6 +365,9 @@ fn load_module(args: &[&str]) -> Result<u32> {
         .context("pactl returned non-UTF-8")?
         .trim()
         .to_string();
+    // Genuinely 32-bit, unlike `object.serial`: this is a PulseAudio module
+    // index (`pa_module.index`, `uint32_t`), which `pactl unload-module` takes
+    // back verbatim. Do not widen it.
     id_str
         .parse::<u32>()
         .with_context(|| format!("pactl returned unexpected module ID: {id_str:?}"))
@@ -532,14 +535,22 @@ fn run_router(
                         return;
                     };
                     if props.get("node.name") == Some(sink_name_owned.as_str()) {
-                        if let Some(serial) = props
+                        match props.get("object.serial").and_then(parse_object_serial) {
-                            .get("object.serial")
+                            Some(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);
                             }
+                            // Never silently: without a serial `try_flush` can
+                            // never route anything, so the whole app-filter mode
+                            // is dead and the only symptom is missing audio.
+                            None => tracing::warn!(
+                                node_id = obj.id,
+                                serial = props.get("object.serial").unwrap_or("<absent>"),
+                                "audio routing: pixelpass sink has no usable object.serial; \
+                                 stream rerouting disabled"
+                            ),
+                        }
                         return;
                     }
                     if props.get("media.class") != Some("Stream/Output/Audio") {
@@ -591,8 +602,30 @@ fn run_router(
     Ok(())
 }
 
+/// Parse a PipeWire `object.serial` property value.
+///
+/// `object.serial` is a **64-bit** monotonically-increasing counter
+/// (`pw_global`'s serial is `uint64_t`); it is *not* a `pw` object id
+/// (those are `u32` and get recycled — the serial exists precisely so
+/// that recycled ids can be disambiguated). Parsing it as `u32` silently
+/// yields `None` past `u32::MAX`, which on a long-lived daemon means the
+/// sink is never registered and no stream is ever routed.
+///
+/// Strict on purpose: PipeWire emits a bare decimal, so anything else
+/// (empty, signed, whitespace-padded, non-numeric, overflowing) is a
+/// property we do not understand and must not guess at. Leading zeroes
+/// are accepted — they are unambiguous and parse to the same value.
+fn parse_object_serial(raw: &str) -> Option<u64> {
+    if raw.is_empty() || !raw.bytes().all(|b| b.is_ascii_digit()) {
+        return None;
+    }
+    raw.parse::<u64>().ok()
+}
+
 struct RouterState {
-    sink_serial: Option<u32>,
+    /// See [`parse_object_serial`] — 64-bit, and not interchangeable with
+    /// the `u32` node ids in `routed_node_ids` / `pending`.
+    sink_serial: Option<u64>,
     default_metadata: Option<pipewire::metadata::Metadata>,
     routed_node_ids: Vec<u32>,
     pending: Vec<u32>,
@@ -654,3 +687,79 @@ fn try_flush(
         let _ = event_tx.send(Event::FirstRoutedStream);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn object_serial_parses_past_u32() {
+        // The regression this fix exists for: a serial one past `u32::MAX`
+        // used to parse as `None` and silently disable rerouting.
+        let beyond = u64::from(u32::MAX) + 1;
+        assert_eq!(parse_object_serial(&beyond.to_string()), Some(beyond));
+        assert_eq!(
+            parse_object_serial(&u64::MAX.to_string()),
+            Some(u64::MAX),
+            "the full 64-bit range must round-trip"
+        );
+    }
+
+    #[test]
+    fn object_serial_accepts_ordinary_serials() {
+        // Without this the valid cases are only 1, 10 and 20 digits long, and
+        // a length-gated mutant (`if (2..10).contains(&raw.len()) { None }`)
+        // survives the whole suite while rejecting every serial a freshly
+        // started daemon actually hands out. (Codex, round 1.)
+        for serial in 0_u64..=1024 {
+            assert_eq!(parse_object_serial(&serial.to_string()), Some(serial));
+        }
+        assert_eq!(parse_object_serial("123456789"), Some(123_456_789));
+        assert_eq!(
+            parse_object_serial("007"),
+            Some(7),
+            "leading zeroes are fine"
+        );
+    }
+
+    #[test]
+    fn object_serial_boundary_values() {
+        assert_eq!(parse_object_serial("0"), Some(0));
+        assert_eq!(parse_object_serial("1"), Some(1));
+        let max32 = u64::from(u32::MAX);
+        assert_eq!(parse_object_serial(&max32.to_string()), Some(max32));
+        assert_eq!(
+            parse_object_serial(&(max32 - 1).to_string()),
+            Some(max32 - 1)
+        );
+    }
+
+    #[test]
+    fn object_serial_round_trips_through_the_metadata_string() {
+        // `try_flush` writes the serial back out as a decimal string for
+        // `target.object`; widening must not introduce a formatting change.
+        for raw in ["0", "4294967296", "18446744073709551615"] {
+            let parsed = parse_object_serial(raw).expect("valid serial");
+            assert_eq!(parsed.to_string(), raw);
+        }
+    }
+
+    #[test]
+    fn object_serial_rejects_malformed() {
+        for raw in [
+            "",
+            " 12",
+            "12 ",
+            "+12",
+            "-1",
+            "1.0",
+            "0x10",
+            "12a",
+            "abc",
+            // u64::MAX + 1 — overflow must be rejected, not wrapped.
+            "18446744073709551616",
+        ] {
+            assert_eq!(parse_object_serial(raw), None, "should reject {raw:?}");
+        }
+    }
+}

src/host/mod.rs

@@ -3,6 +3,7 @@ mod capture;
 mod pipeline;
 mod quality;
 mod serve;
+pub mod taint;
 mod wayland;
 mod x11;
 

src/host/pipeline.rs

@@ -189,9 +189,32 @@ fn build_args(
         "!".into(),
         "queue".into(),
         "!".into(),
+    ];
+    // Debug A/V-drift tap: when PIXELPASS_TS_DUMP=<path> is set, tee the exact
+    // muxed TS both to fd=1 (normal serve path, unchanged) and to a file, so the
+    // host-side stream can be ffprobe'd for capture-side audio/video PTS drift.
+    // Each tee branch has its own queue so the disk sink can't backpressure the
+    // live serve branch. No effect when unset. (Mirrors PIXELPASS_GST_DEBUG.)
+    if let Some(dump) = std::env::var_os("PIXELPASS_TS_DUMP") {
+        let path = dump.to_string_lossy().into_owned();
+        args.extend([
+            "tee".into(),
+            "name=dbgtee".into(),
+            "!".into(),
+            "queue".into(),
+            "!".into(),
             "fdsink".into(),
             "fd=1".into(),
-    ];
+            "dbgtee.".into(),
+            "!".into(),
+            "queue".into(),
+            "!".into(),
+            "filesink".into(),
+            format!("location={path}"),
+        ]);
+    } else {
+        args.extend(["fdsink".into(), "fd=1".into()]);
+    }
 
     // Downscale step for the quality presets. `None` = encode at native size
     // (the "Source" preset, or a source already at/below the target height — we

src/host/taint/fixture.rs

@@ -0,0 +1,338 @@
+//! Synthetic graph builders for the taint-engine tests.
+//!
+//! Serials are handed out monotonically and never reused, exactly as
+//! PipeWire does; global ids are handed out separately and **may be reused
+//! on purpose**, which is what the recycling tests need.
+
+use std::collections::BTreeMap;
+
+use super::snapshot::{
+    ClientSnapshot, GlobalId, GraphSnapshot, LinkSnapshot, MediaRole, NodeProps, NodeSnapshot,
+    PortDirection, PortSnapshot, Serial,
+};
+
+/// pipewire-pulse's PID, as measured on the target machine.
+pub const PULSE_PID: u32 = 2541;
+/// WirePlumber's PID — one process owning every device node on the box.
+pub const SESSION_PID: u32 = 900;
+
+/// A node's identity in a fixture: what tests pass around.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub struct NodeRef {
+    pub serial: Serial,
+    pub id: GlobalId,
+}
+
+#[derive(Default)]
+pub struct Graph {
+    next_serial: u64,
+    next_id: u32,
+    nodes: Vec<NodeSnapshot>,
+    ports: Vec<PortSnapshot>,
+    links: Vec<LinkSnapshot>,
+    clients: Vec<ClientSnapshot>,
+    /// One client connection per process / per module, which is what the
+    /// live graph looks like. Tests that need the *split*-client shape
+    /// (GStreamer opens one per stream) pass clients explicitly instead.
+    client_by_app: BTreeMap<u32, GlobalId>,
+    client_by_module: BTreeMap<u64, GlobalId>,
+    session_client: Option<GlobalId>,
+}
+
+impl Graph {
+    pub fn new() -> Self {
+        Self {
+            // Start past u32::MAX so every fixture also exercises the phase
+            // 0a widening: a serial that a u32 model would have truncated.
+            next_serial: u64::from(u32::MAX) + 1,
+            next_id: 1,
+            ..Self::default()
+        }
+    }
+
+    fn serial(&mut self) -> Serial {
+        self.next_serial += 1;
+        Serial(self.next_serial)
+    }
+
+    fn id(&mut self) -> GlobalId {
+        self.next_id += 1;
+        GlobalId(self.next_id)
+    }
+
+    /// A client object. `sec_pid` is `pipewire.sec.pid` — pipewire-pulse's
+    /// PID for Pulse-emulated clients.
+    pub fn client(&mut self, sec_pid: Option<u32>) -> GlobalId {
+        let serial = self.serial();
+        let id = self.id();
+        self.clients.push(ClientSnapshot {
+            serial,
+            id,
+            sec_pid,
+        });
+        id
+    }
+
+    /// The client connection an ordinary process holds — one per PID,
+    /// created on demand.
+    pub fn client_of_app(&mut self, pid: u32) -> GlobalId {
+        if let Some(id) = self.client_by_app.get(&pid) {
+            return *id;
+        }
+        let id = self.client(Some(PULSE_PID));
+        self.client_by_app.insert(pid, id);
+        id
+    }
+
+    /// An ordinary application stream: its own client, its own PID.
+    pub fn app_node(&mut self, name: &str, role: MediaRole, pid: u32) -> NodeRef {
+        let client = self.client_of_app(pid);
+        self.node(name, role, app(client, pid))
+    }
+
+    /// The client a pactl module holds. Measured: each module gets its own
+    /// (`sink-sunshine-*` were clients 83/86/92), which is why one tainted
+    /// module does not fuse with the next.
+    pub fn client_of_module(&mut self, module: u64) -> GlobalId {
+        match self.client_by_module.get(&module) {
+            Some(id) => *id,
+            None => {
+                let id = self.client(Some(PULSE_PID));
+                self.client_by_module.insert(module, id);
+                id
+            }
+        }
+    }
+
+    /// A leg of a pactl-loaded module: one client per module, and the
+    /// node's `application.process.id` is **pipewire-pulse's own**, because
+    /// pipewire-pulse genuinely is the client.
+    pub fn module_node(&mut self, name: &str, role: MediaRole, module: u64) -> NodeRef {
+        let client = self.client_of_module(module);
+        self.node(name, role, pulse_module(client, module, PULSE_PID))
+    }
+
+    /// A leg joined to its siblings by `node.link-group` — loopback,
+    /// filter-chain, echo-cancel.
+    pub fn group_node(&mut self, name: &str, role: MediaRole, group: &str, pid: u32) -> NodeRef {
+        let client = self.client_of_app(pid);
+        self.node(name, role, link_group(group, client, pid))
+    }
+
+    /// A device node as the session manager creates it: no strong key,
+    /// WirePlumber's client and PID — shared with every other device — and
+    /// a `device.id`, which is what marks it as session-manager-exported.
+    pub fn device_node(&mut self, name: &str, role: MediaRole) -> NodeRef {
+        let session = match self.session_client {
+            Some(id) => id,
+            None => {
+                let id = self.client(None);
+                self.session_client = Some(id);
+                id
+            }
+        };
+        self.node(name, role, device(session, SESSION_PID))
+    }
+
+    /// A node that *belongs to* a Device but is not a passive device node —
+    /// a filter associated with a card. Phase 3 must not classify this as a
+    /// session device, or it loses both its coarse owner keys and its
+    /// ability to trip the fail-closed backstop.
+    pub fn device_associated_filter(&mut self, name: &str, role: MediaRole, pid: u32) -> NodeRef {
+        let client = self.client_of_app(pid);
+        self.node(name, role, app(client, pid))
+    }
+
+    /// A **virtual** sink an application created natively: an `Audio/Sink`
+    /// with no `device.id` and no strong key, sharing one client with the
+    /// stream that re-emits what it receives. Coarse keys must still bridge
+    /// these two, or the whole call leaks through the re-emitting leg.
+    pub fn native_virtual_node(&mut self, name: &str, role: MediaRole, pid: u32) -> NodeRef {
+        let client = self.client_of_app(pid);
+        self.node(name, role, app(client, pid))
+    }
+
+    pub fn peerspeak_node(&mut self, name: &str, pid: u32) -> NodeRef {
+        let client = self.client_of_app(pid);
+        self.node(name, MediaRole::StreamOutput, peerspeak_owned(client, pid))
+    }
+
+    pub fn node(&mut self, name: &str, role: MediaRole, props: NodeProps) -> NodeRef {
+        let id = self.id();
+        self.node_with_id(name, role, id, props)
+    }
+
+    /// Force a global id — for reproducing id recycling after teardown.
+    pub fn node_with_id(
+        &mut self,
+        name: &str,
+        role: MediaRole,
+        id: GlobalId,
+        props: NodeProps,
+    ) -> NodeRef {
+        let serial = self.serial();
+        self.nodes.push(NodeSnapshot {
+            serial,
+            id,
+            name: Some(name.to_string()),
+            role,
+            props,
+        });
+        NodeRef { serial, id }
+    }
+
+    pub fn port(&mut self, node: NodeRef, direction: PortDirection, exclusive: bool) {
+        let serial = self.serial();
+        let id = self.id();
+        self.ports.push(PortSnapshot {
+            serial,
+            id,
+            node: node.id,
+            direction,
+            exclusive,
+            monitor: false,
+        });
+    }
+
+    /// A signal edge: audio flows `from → to`.
+    pub fn link(&mut self, from: NodeRef, to: NodeRef) {
+        self.link_ids(from.id, to.id);
+    }
+
+    /// A link naming raw ids, so a test can dangle an endpoint.
+    pub fn link_ids(&mut self, from: GlobalId, to: GlobalId) {
+        let serial = self.serial();
+        let id = self.id();
+        self.links.push(LinkSnapshot {
+            serial,
+            id,
+            output_node: from,
+            input_node: to,
+            output_port: None,
+            input_port: None,
+        });
+    }
+
+    /// An id that belongs to nothing — for unresolved-endpoint tests.
+    pub fn dangling_id(&mut self) -> GlobalId {
+        self.id()
+    }
+
+    pub fn build(&self) -> GraphSnapshot {
+        self.build_without(&[])
+    }
+
+    /// A later snapshot in which some nodes have gone away, along with
+    /// their ports and every link touching them. Surviving objects keep
+    /// their serials, which is what makes sticky-taint sequences testable.
+    pub fn build_without(&self, dropped: &[NodeRef]) -> GraphSnapshot {
+        let gone_serials: Vec<Serial> = dropped.iter().map(|n| n.serial).collect();
+        let nodes: Vec<NodeSnapshot> = self
+            .nodes
+            .iter()
+            .filter(|n| !gone_serials.contains(&n.serial))
+            .cloned()
+            .collect();
+        // Filter by what was *dropped*, not by what is live: a link to an id
+        // that never had a node is a dangling endpoint, and dropping those
+        // here would quietly disarm every unresolved-ancestry test.
+        let gone_ids: Vec<GlobalId> = dropped.iter().map(|n| n.id).collect();
+        GraphSnapshot::new(
+            nodes,
+            self.ports
+                .iter()
+                .filter(|p| !gone_ids.contains(&p.node))
+                .cloned()
+                .collect(),
+            self.links
+                .iter()
+                .filter(|l| !gone_ids.contains(&l.output_node) && !gone_ids.contains(&l.input_node))
+                .cloned()
+                .collect(),
+            self.clients.clone(),
+        )
+    }
+
+    /// Drop clients too — full owner teardown.
+    ///
+    /// Invalidates the per-app/per-module caches as well: leaving them
+    /// stale made a later `client_of_app` hand back the *removed* client's
+    /// id, so a test that meant "a brand-new client after teardown" was
+    /// really building a node pointing at a client object that no longer
+    /// existed (Codex round 1, finding 8).
+    pub fn drop_clients(&mut self, ids: &[GlobalId]) {
+        self.clients.retain(|c| !ids.contains(&c.id));
+        self.client_by_app.retain(|_, id| !ids.contains(id));
+        self.client_by_module.retain(|_, id| !ids.contains(id));
+        if self.session_client.is_some_and(|id| ids.contains(&id)) {
+            self.session_client = None;
+        }
+    }
+
+    /// A client that reuses a global id a dead client had — the recycling
+    /// case, with a fresh serial.
+    pub fn client_with_id(&mut self, id: GlobalId, sec_pid: Option<u32>) -> GlobalId {
+        let serial = self.serial();
+        self.clients.push(ClientSnapshot {
+            serial,
+            id,
+            sec_pid,
+        });
+        id
+    }
+}
+
+/// An ordinary application stream: real PID, one client connection.
+pub fn app(client: GlobalId, pid: u32) -> NodeProps {
+    NodeProps {
+        client_id: Some(client),
+        process_id: Some(pid),
+        ..NodeProps::default()
+    }
+}
+
+/// A pactl-module-created stream: the daemon is the client, so the node's
+/// `application.process.id` is pipewire-pulse's own.
+pub fn pulse_module(client: GlobalId, module: u64, pulse_pid: u32) -> NodeProps {
+    NodeProps {
+        pulse_module_id: Some(module),
+        client_id: Some(client),
+        process_id: Some(pulse_pid),
+        ..NodeProps::default()
+    }
+}
+
+/// A PipeWire-module leg joined to its siblings by `node.link-group`
+/// (loopback, filter-chain, echo-cancel).
+pub fn link_group(group: &str, client: GlobalId, pid: u32) -> NodeProps {
+    NodeProps {
+        link_group: Some(group.to_string()),
+        client_id: Some(client),
+        process_id: Some(pid),
+        ..NodeProps::default()
+    }
+}
+
+/// A device node as the session manager creates it: no strong key, and the
+/// session manager's own client and PID — shared with every other device.
+///
+/// Measured 2026-07-21: real ALSA device nodes carry the shared
+/// `client.id` but **no** `application.process.id` at all. Giving them one
+/// here is deliberately *more* pessimistic than reality — it hands the
+/// engine a second coarse key it could fuse devices on, so a test that
+/// passes here also passes against the real props.
+pub fn device(session_client: GlobalId, session_pid: u32) -> NodeProps {
+    NodeProps {
+        client_id: Some(session_client),
+        process_id: Some(session_pid),
+        session_device: true,
+        ..NodeProps::default()
+    }
+}
+
+pub fn peerspeak_owned(client: GlobalId, pid: u32) -> NodeProps {
+    NodeProps {
+        peerspeak_owned: true,
+        ..app(client, pid)
+    }
+}

src/host/taint/mod.rs

@@ -0,0 +1,956 @@
+//! The taint engine — decides which `Stream/Output/Audio` nodes may be
+//! fanned out into the screen-share capture without echoing peerspeak's own
+//! audio back at the viewer.
+//!
+//! Implements design v3.4 §6.1–§6.1.3 (`peerspeak/docs/
+//! screenshare-audio-exclusion-plan.md`), phase 2 of the implementation
+//! plan. **Pure**: no PipeWire types appear in any signature, nothing here
+//! touches the daemon, and every test builds its own graph.
+//!
+//! ## The one-sentence predicate
+//!
+//! > A node is eligible only if **no** signal path reaches it from a
+//! > peerspeak-owned node, the live AEC identity, or any pixelpass-owned
+//! > object. **Unresolvable ancestry is not eligible.**
+//!
+//! That last sentence is the invariant the whole design rests on: every
+//! other failure mode in here degrades into over-exclusion (one app's audio
+//! silently missing from the share) rather than into echo.
+//!
+//! ## Why a graph walk and not a property check
+//!
+//! Exclusion does not propagate downstream by itself. Any node that
+//! re-emits audio it received is a fresh, *untagged* `Stream/Output/Audio`
+//! carrying the mix — including the one peerspeak playback stream that was
+//! correctly excluded one hop earlier. EasyEffects, `module-loopback`,
+//! combine-sinks, tunnel/RTP sinks and virtual-sink forwarders all have this
+//! shape, and at least one such topology has been observed live on the
+//! target machine.
+//!
+//! Taint therefore flows over **three** edge types:
+//!
+//! 1. **Link edges** — `link.output.node → link.input.node`.
+//! 2. **Sink → monitor** — free at node granularity: the monitor connection
+//!    *is* a real Link whose output node is the sink node itself (measured).
+//!    A port-granular walk would need a synthetic edge; a node-granular one
+//!    does not.
+//! 3. **Owner bridges** — the intra-process hop the graph cannot see. See
+//!    [`owner`]; this is the hard one.
+//!
+//! ## Stickiness
+//!
+//! Taint is **sticky per owner** for the duration of the share, because a
+//! topological recompute forgets *buffered* audio: an app can read a tainted
+//! monitor into a 5-second ring buffer, then have its input leg vanish, and
+//! a purely topological engine would relink its output while it is still
+//! emitting peerspeak's audio out of that buffer. No graph event marks the
+//! moment a buffer drains.
+//!
+//! Stickiness is keyed on [`Serial`] — never on a node id, `client.id`,
+//! module index or `link-group` string, **all of which recycle on this
+//! stack**. An entry is cleared only once every member object has
+//! disappeared; a key that reappears after full teardown is a new owner and
+//! starts clean.
+//!
+//! ## ⚠️ KNOWN OPEN GAP — buffered audio across a full PipeWire teardown of
+//! ## a still-live process (Codex phase-2 rounds 5–6) — DESIGN DECISION OWED
+//!
+//! **This is an in-threat-model echo gap, not an outside-the-model one — an
+//! earlier version of this note wrongly scoped it to keyless streams.**
+//!
+//! The scenario, entirely with a real PID-bearing app (a recorder, a DAW,
+//! a GStreamer pipeline): it reads the call into an application buffer,
+//! **fully** tears down its PipeWire Node *and* Client while keeping that
+//! buffer, then — still the same live process — opens a fresh Client and a
+//! `Stream/Output/Audio` and replays. Every old serial is gone, so
+//! [`seed_sticky`] refuses to apply the remembered PID fingerprint (the
+//! fingerprint is lifetime-scoped to a live serial member, because bare keys
+//! recycle); no reader is live in the new epoch, so the backstop does not
+//! fire; the replayed leg is eligible.
+//!
+//! It is real and reachable by non-adversarial software. It also sits
+//! exactly on the design's stated boundary (v3.4 §6.1.3: "a key that
+//! reappears after full teardown is a new owner and starts clean"), so
+//! closing it is a **design change**, not a local bug fix:
+//!
+//! - **Option A — accept as a documented v1 limitation.** Contrived in
+//!   practice (most apps hold their PipeWire connection open for their
+//!   lifetime; the round-2 fix already covers the common
+//!   idle-a-client-and-open-another case), never a *silent* correctness
+//!   regression since it is written down, and phase 5's dry run would show
+//!   it. But it is a known echo path, which sits badly against the feature's
+//!   fail-closed ethos.
+//! - **Option B — process-generation lifetime.** Key the fingerprint's
+//!   lifetime on the owning **process** being alive — PID + `/proc` start
+//!   time (or a pidfd) to defeat PID reuse — instead of on a live PipeWire
+//!   object. Phase 3 supplies process liveness; §6.1.3's node/client-only
+//!   lifetime definition is revised. Closes the PID-bearing case; the truly
+//!   keyless sub-case (no PID at all) genuinely *is* outside the threat
+//!   model and stays a documented limit.
+//!
+//! The choice is the designer's (it revises the security surface). Until it
+//! is made, `a_fingerprint_does_not_outlive_its_owner` encodes Option A's
+//! behaviour — flip it if B is chosen. Owed to the design doc as round 8.
+
+// Phase 2 lands the engine behind its own test surface and nothing else:
+// the registry observer that will feed it is phase 3, so in a non-test
+// build every item here is legitimately unreachable for now.
+#![allow(dead_code)]
+
+pub mod owner;
+pub mod snapshot;
+
+#[cfg(test)]
+mod fixture;
+#[cfg(test)]
+mod tests;
+
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
+
+use owner::{OwnerComponents, OwnerKey};
+use snapshot::{GraphSnapshot, IdLookup, MediaRole, NodeSnapshot, Serial};
+
+/// The `node.name` prefix of a pixelpass capture sink. Any host's sink
+/// counts, not just ours — fanning out a stream that is downstream of
+/// *another* pixelpass host's capture sink builds a cycle (v3.4 §6.2).
+pub const CAPTURE_SINK_PREFIX: &str = "pixelpass_capture_";
+
+/// `node.link-group` prefix that marks *some* echo canceller. Hazard
+/// detection only — it does **not** identify peerspeak's instance, which is
+/// what `pulse.module.id` is for (v3.4 §5.2 correction 4).
+pub const ECHO_CANCEL_GROUP_PREFIX: &str = "echo-cancel-";
+
+/// Why a node is tainted or excluded. Stable machine-readable codes: this
+/// value is the phase 5 audit output, the phase 6 status event, and the
+/// eventual answer to "why isn't this app being shared?".
+#[derive(Clone, Copy, PartialEq, Eq, Debug, Hash)]
+pub enum Reason {
+    /// Carries the `peerspeak.owned` tag (v3.4 §5.1).
+    PeerspeakOwned,
+    /// `pulse.module.id` equals the live AEC module index — exact equality
+    /// only. "Has any `pulse.module.id`" is explicitly rejected as a rule:
+    /// tunnel/RTP/loopback modules may be the only carrier of audio the
+    /// user legitimately wants shared (v3.4 §5.2 correction 2).
+    AecIdentity,
+    /// A pixelpass-owned object, ours or another host's capture sink.
+    PixelpassOwned,
+    /// An `echo-cancel-*` group that is **not** our validated identity.
+    /// Decision D3: warn and exclude rather than fan out.
+    ForeignEchoCancel,
+    /// Reached by a signal path from a tainted node (link or monitor edge).
+    TaintedUpstream,
+    /// Reached across an owner bridge; the key that did it, when the
+    /// tainted member shares one directly rather than transitively.
+    TaintedOwnerBridge { key: Option<OwnerKey> },
+    /// A link endpoint, or a node's own id, could not be resolved in this
+    /// snapshot. Fail closed (v3.4 §6.1.4).
+    UnresolvedAncestry,
+    /// A tainted capture stream whose owner cannot be bounded by any usable
+    /// key, so its sibling output legs cannot be identified. Fail closed
+    /// (v3.4 §6.1.1, final paragraph).
+    UnresolvedOwner,
+    /// The observer has not reached a complete, coherent view of the graph
+    /// yet. No decision made from a partial graph is a decision.
+    GraphNotReady,
+    /// A `port.exclusive` port — fan-out will be refused (v3.4 §6.2). Local
+    /// to the node; does not propagate.
+    PortExclusive,
+    /// An encoded/passthrough stream — a second link would corrupt it.
+    /// Local to the node; does not propagate.
+    Passthrough,
+}
+
+impl Reason {
+    pub fn code(self) -> &'static str {
+        match self {
+            Self::PeerspeakOwned => "peerspeak-owned",
+            Self::AecIdentity => "aec-identity",
+            Self::PixelpassOwned => "pixelpass-owned",
+            Self::ForeignEchoCancel => "foreign-echo-cancel",
+            Self::TaintedUpstream => "tainted-upstream",
+            Self::TaintedOwnerBridge { .. } => "tainted-owner-bridge",
+            Self::UnresolvedAncestry => "unresolved-ancestry",
+            Self::UnresolvedOwner => "unresolved-owner",
+            Self::GraphNotReady => "graph-not-ready",
+            Self::PortExclusive => "port-exclusive",
+            Self::Passthrough => "passthrough",
+        }
+    }
+
+    /// Lower wins. A node can acquire taint several ways in one recompute
+    /// and the reported reason must not depend on traversal order, or the
+    /// audit output is unstable and the fixture tests are flaky. Explicit
+    /// priority, not BFS arrival order.
+    fn priority(self) -> u8 {
+        match self {
+            Self::PeerspeakOwned => 0,
+            Self::AecIdentity => 1,
+            Self::PixelpassOwned => 2,
+            Self::ForeignEchoCancel => 3,
+            Self::TaintedUpstream => 4,
+            Self::TaintedOwnerBridge { .. } => 5,
+            Self::UnresolvedAncestry => 6,
+            Self::UnresolvedOwner => 7,
+            // Non-propagating; never competes with the taint reasons above
+            // because it is only consulted for untainted candidates.
+            Self::GraphNotReady => 8,
+            Self::PortExclusive => 9,
+            Self::Passthrough => 10,
+        }
+    }
+
+    /// Does this reason spread to downstream nodes and owner siblings?
+    fn propagates(self) -> bool {
+        self.priority() <= Self::UnresolvedOwner.priority()
+    }
+}
+
+/// Everything the engine needs that is not in the graph itself.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct ExclusionCtx {
+    /// The **validated** live AEC module index, or `None` for `--aec=off`.
+    /// The validation state machine (phase 4) owns the transitions; if it
+    /// is still `Validating` or has `Failed`, its caller must not fan out at
+    /// all rather than passing `None` here, which would merely mean "there
+    /// is no AEC".
+    pub aec_module_id: Option<u64>,
+    /// pipewire-pulse's own PID, derived by the observer (phase 3) from a
+    /// consistent `pipewire.sec.pid` across Pulse clients validated against
+    /// `/proc/<pid>/comm`. `None` is safe but coarse — see [`owner`].
+    pub pipewire_pulse_pid: Option<u32>,
+    /// Serials of objects pixelpass itself created this run.
+    pub pixelpass_owned: BTreeSet<Serial>,
+    /// False until the readiness epoch has been reached (phase 3). Every
+    /// candidate is then ineligible: a decision from a partial graph is not
+    /// a decision.
+    pub graph_ready: bool,
+}
+
+/// Object identity for sticky bookkeeping. Always a [`Serial`] — never a
+/// recyclable id (v3.4 §6.1.3).
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug)]
+pub enum ObjectRef {
+    Node(Serial),
+    Client(Serial),
+}
+
+/// One owner that has been tainted, and every object observed to constitute
+/// it. Cleared only when **all** of them are gone.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct StickyOwner {
+    /// Every object seen to be part of this owner, ever. Membership
+    /// accumulates: that is what makes "clear only once all member objects
+    /// have disappeared" true across churn.
+    pub members: BTreeSet<ObjectRef>,
+    /// Owner keys remembered across connections — strong keys and a usable
+    /// process id, never `client.id`. Applied only while some serial member
+    /// above is still live, which is what keeps a recyclable key from
+    /// resurrecting a dead owner.
+    ///
+    /// Needed because a live Client is not the same thing as a live owner:
+    /// a process can leave one connection idle and open a second, and
+    /// GStreamer opens one connection per stream as a matter of course, so
+    /// following connections alone lets the next leg escape (Codex round 2,
+    /// finding 2).
+    pub fingerprints: BTreeSet<owner::Fingerprint>,
+    /// The reason recorded for each node that was tainted in its own right.
+    /// Kept per node rather than collapsed to one owner-wide reason, or a
+    /// forwarder's output leg inherits its *input* leg's `tainted-upstream`
+    /// and the audit output stops naming the mechanism that actually
+    /// excluded it.
+    pub node_reasons: BTreeMap<Serial, Reason>,
+}
+
+impl StickyOwner {
+    /// The reason to apply to a member: its own recorded one, or — for a
+    /// leg that appeared later — the fact that it belongs to a tainted
+    /// owner, which is a bridge by definition.
+    fn reason_for(&self, serial: Serial) -> Reason {
+        self.node_reasons
+            .get(&serial)
+            .copied()
+            .unwrap_or(Reason::TaintedOwnerBridge { key: None })
+    }
+}
+
+/// Threaded explicitly through [`evaluate`] so stickiness is testable as a
+/// sequence of snapshots rather than as hidden mutable state.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct StickyState {
+    pub owners: Vec<StickyOwner>,
+}
+
+impl StickyState {
+    pub fn is_empty(&self) -> bool {
+        self.owners.is_empty()
+    }
+}
+
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub enum Eligibility {
+    Eligible,
+    NotEligible {
+        reason: Reason,
+        /// The taint was carried over from a previous snapshot rather than
+        /// derived from the current topology.
+        sticky: bool,
+    },
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct NodeDecision {
+    pub serial: Serial,
+    pub name: Option<String>,
+    pub eligibility: Eligibility,
+}
+
+impl NodeDecision {
+    pub fn is_eligible(&self) -> bool {
+        matches!(self.eligibility, Eligibility::Eligible)
+    }
+
+    pub fn reason(&self) -> Option<Reason> {
+        match self.eligibility {
+            Eligibility::Eligible => None,
+            Eligibility::NotEligible { reason, .. } => Some(reason),
+        }
+    }
+}
+
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub struct TaintEntry {
+    pub reason: Reason,
+    pub sticky: bool,
+}
+
+/// The result of one recompute.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct Decisions {
+    /// Every `Stream/Output/Audio` node in the snapshot — the complete
+    /// candidate universe, so callers can assert an exact partition rather
+    /// than spot-checking named nodes.
+    pub candidates: BTreeMap<Serial, NodeDecision>,
+    /// Taint over *all* node roles, for diagnostics and for the phase 5
+    /// audit output.
+    pub taint: BTreeMap<Serial, TaintEntry>,
+}
+
+impl Decisions {
+    /// Serials of eligible candidates, ascending.
+    pub fn eligible(&self) -> Vec<Serial> {
+        self.candidates
+            .values()
+            .filter(|d| d.is_eligible())
+            .map(|d| d.serial)
+            .collect()
+    }
+
+    /// `(serial, reason code)` for excluded candidates, ascending.
+    pub fn excluded(&self) -> Vec<(Serial, &'static str)> {
+        self.candidates
+            .values()
+            .filter_map(|d| d.reason().map(|r| (d.serial, r.code())))
+            .collect()
+    }
+}
+
+/// Recompute eligibility for the whole graph.
+///
+/// Full recompute per graph event is the v1 design; there is deliberately
+/// no incremental dirty-set.
+///
+/// ⚠️ **Cost is not O(V+E), despite what v3.4 §6.4 says.** Each fixpoint
+/// pass re-runs a full link BFS *and* a full owner scan, and the bridge
+/// scans every tainted source in a component for each target, so the bound
+/// is `O(D · (V + E + Σ_C |sources_C|·|targets_C|))` — worst case
+/// `O(D · (V² + E))` — for an owner-bridge depth D. D is 1 for every
+/// topology observed so far and 2 for a forwarder feeding a forwarder, and
+/// components on a real desktop are two or three nodes; the quadratic term
+/// needs one owner with many legs. A 60-layer chain test guards the depth
+/// dimension only. Phase 5 records the real recompute-duration
+/// distribution and maximum, which is what "full recompute is fine for v1"
+/// should rest on — measured headroom, not a node count.
+pub fn evaluate(
+    snapshot: &GraphSnapshot,
+    ctx: &ExclusionCtx,
+    prior: &StickyState,
+) -> (Decisions, StickyState) {
+    let components = OwnerComponents::build(snapshot, ctx.pipewire_pulse_pid);
+    let keys = owner::OwnerKeyIndex::build(snapshot, ctx.pipewire_pulse_pid);
+
+    let mut taint: BTreeMap<Serial, Reason> = BTreeMap::new();
+    let mut sticky_serials: BTreeSet<Serial> = BTreeSet::new();
+
+    seed_local_roots(snapshot, ctx, &mut taint);
+    seed_sticky(
+        snapshot,
+        &keys,
+        prior,
+        &components,
+        &mut taint,
+        &mut sticky_serials,
+    );
+
+    // Monotone fixpoint: every step only adds taint, or lowers a node's
+    // reason priority, both of which are bounded. Link propagation and the
+    // owner bridge feed each other — a bridged output leg has downstream
+    // links, and a downstream monitor reader bridges to its own siblings —
+    // so neither can be run once.
+    let edges = downstream_edges(snapshot, &mut taint);
+    loop {
+        let mut changed = false;
+        changed |= propagate_links(&edges.edges, &mut taint);
+        changed |= propagate_owner_bridge(&keys, &components, &edges, &mut taint);
+        changed |= propagate_unresolved_owner(snapshot, &keys, &edges, &mut taint);
+        if !changed {
+            break;
+        }
+    }
+
+    let decisions = build_decisions(snapshot, ctx, &taint, &sticky_serials);
+    // ⚠️ Readiness gates **retirement only**, never addition (Codex rounds
+    // 1 and 2, which caught the two halves of this in turn). An object
+    // missing from an untrustworthy snapshot has not been observed to
+    // disappear, so retiring on that basis erases history and the next
+    // ready recompute hands back a clean bill of health. But taint
+    // *observed* during a not-ready epoch is real — a reader can consume
+    // and buffer the call and then vanish before readiness — so discarding
+    // additions was the same defect pointing the other way.
+    let next_sticky = build_sticky(snapshot, &keys, &components, &taint, prior, ctx.graph_ready);
+    (decisions, next_sticky)
+}
+
+/// Roots that are visible on the node itself.
+fn seed_local_roots(
+    snapshot: &GraphSnapshot,
+    ctx: &ExclusionCtx,
+    taint: &mut BTreeMap<Serial, Reason>,
+) {
+    for node in snapshot.nodes() {
+        if let Some(reason) = local_root_reason(node, ctx) {
+            raise(taint, node.serial, reason);
+        }
+        // A node whose own global id is ambiguous cannot be the reliable
+        // endpoint of any link, so its ancestry is unresolvable.
+        if snapshot.node_by_id(node.id) == Some(IdLookup::Ambiguous) {
+            raise(taint, node.serial, Reason::UnresolvedAncestry);
+        }
+    }
+}
+
+fn local_root_reason(node: &NodeSnapshot, ctx: &ExclusionCtx) -> Option<Reason> {
+    if node.props.peerspeak_owned {
+        return Some(Reason::PeerspeakOwned);
+    }
+    if let (Some(module), Some(aec)) = (node.props.pulse_module_id, ctx.aec_module_id)
+        && module == aec
+    {
+        return Some(Reason::AecIdentity);
+    }
+    if ctx.pixelpass_owned.contains(&node.serial)
+        || node
+            .name
+            .as_deref()
+            .is_some_and(|name| name.starts_with(CAPTURE_SINK_PREFIX))
+    {
+        return Some(Reason::PixelpassOwned);
+    }
+    if node
+        .props
+        .link_group
+        .as_deref()
+        .is_some_and(|group| group.starts_with(ECHO_CANCEL_GROUP_PREFIX))
+    {
+        return Some(Reason::ForeignEchoCancel);
+    }
+    None
+}
+
+/// Carry taint forward from previous snapshots (v3.4 §6.1.3).
+///
+/// An owner is re-seeded from three kinds of evidence, all lifetime-scoped
+/// to a still-live member: its own surviving nodes, nodes on a surviving
+/// **Client**, and nodes presenting a remembered owner **fingerprint**.
+fn seed_sticky(
+    snapshot: &GraphSnapshot,
+    keys: &owner::OwnerKeyIndex,
+    prior: &StickyState,
+    components: &OwnerComponents,
+    taint: &mut BTreeMap<Serial, Reason>,
+    sticky_serials: &mut BTreeSet<Serial>,
+) {
+    for entry in &prior.owners {
+        let mut live_nodes: Vec<Serial> = Vec::new();
+        for member in &entry.members {
+            match member {
+                ObjectRef::Node(serial) => {
+                    if snapshot.node(*serial).is_some() {
+                        live_nodes.push(*serial);
+                    }
+                }
+                // A surviving **Client** re-seeds too. An app can close
+                // every stream it had while keeping its PipeWire connection
+                // open, then open a fresh one — Firefox does exactly this.
+                ObjectRef::Client(serial) => {
+                    live_nodes.extend(nodes_of_client(snapshot, keys, *serial));
+                }
+            }
+        }
+        if live_nodes.is_empty() && !entry.members.iter().any(|m| is_live(snapshot, *m)) {
+            // Nothing of this owner remains; its fingerprints are just
+            // recyclable strings now and must not be applied to anyone.
+            continue;
+        }
+        // Fingerprints reach a *new connection* of the same still-live
+        // process, which neither of the two paths above can see.
+        for fingerprint in &entry.fingerprints {
+            live_nodes.extend(
+                snapshot
+                    .nodes()
+                    .filter(|node| keys.has_fingerprint(node.serial, fingerprint))
+                    .map(|node| node.serial),
+            );
+        }
+        // The owner is sticky, not the individual node: a leg that appears
+        // later in the same still-live owner inherits the taint.
+        for serial in live_nodes {
+            for member in components.members_with(serial) {
+                let reason = entry.reason_for(*member);
+                if raise(taint, *member, reason) || taint.get(member) == Some(&reason) {
+                    sticky_serials.insert(*member);
+                }
+            }
+        }
+    }
+}
+
+/// Nodes currently attached to a client, by the client's **serial**. The
+/// client's snapshot-local id is resolved fresh each time, so a recycled id
+/// can never resurrect a dead owner.
+///
+/// Nodes for which `client.id` is not a usable owner key — session-manager
+/// device nodes — are excluded, or the shared `WirePlumber [export]` Client
+/// would drag every sound card on the box into one sticky owner.
+///
+/// The same gate is applied when *recording* clients into a sticky entry
+/// (`owner::client_serials_of`). Either one alone closes the leak; both are
+/// kept because they answer different questions ("may this client be
+/// remembered?" and "may this client speak for that node?"), and the
+/// regression test kills the removal of the pair.
+fn nodes_of_client(
+    snapshot: &GraphSnapshot,
+    keys: &owner::OwnerKeyIndex,
+    client: Serial,
+) -> Vec<Serial> {
+    let Some(id) = snapshot
+        .clients()
+        .find(|c| c.serial == client)
+        .map(|c| c.id)
+    else {
+        return Vec::new();
+    };
+    snapshot
+        .nodes()
+        .filter(|node| node.props.client_id == Some(id))
+        .filter(|node| keys.uses_client_key(node.serial))
+        .map(|node| node.serial)
+        .collect()
+}
+
+/// `output node → input nodes`, resolving snapshot-local ids. An endpoint
+/// that does not resolve taints the *other* end as unresolved ancestry when
+/// that other end is the input side — we cannot know what is feeding it.
+fn downstream_edges(snapshot: &GraphSnapshot, taint: &mut BTreeMap<Serial, Reason>) -> Edges {
+    let mut edges: BTreeMap<Serial, Vec<Serial>> = BTreeMap::new();
+    let mut receivers: BTreeSet<Serial> = BTreeSet::new();
+    for link in snapshot.links() {
+        let from = snapshot.node_by_id(link.output_node);
+        let to = snapshot.node_by_id(link.input_node);
+        match (from, to) {
+            (Some(IdLookup::Unique(from)), Some(IdLookup::Unique(to))) => {
+                edges.entry(from).or_default().push(to);
+                receivers.insert(to);
+            }
+            (_, Some(IdLookup::Unique(to))) => {
+                // Something feeds this node and we cannot say what.
+                raise(taint, to, Reason::UnresolvedAncestry);
+                receivers.insert(to);
+            }
+            (_, Some(IdLookup::Ambiguous)) => {
+                // Several nodes claim the input id and we cannot say which
+                // one this link feeds, so every claimant is a receiver.
+                // They are already tainted as unresolved by their own
+                // ambiguous id — but taint without receiver status cannot
+                // start an owner bridge, so their sibling output legs stayed
+                // Eligible (Codex round 2, finding 3).
+                receivers.extend(
+                    snapshot
+                        .nodes_with_id(link.input_node)
+                        .map(|node| node.serial),
+                );
+            }
+            _ => {}
+        }
+    }
+    for targets in edges.values_mut() {
+        targets.sort_unstable();
+        targets.dedup();
+    }
+    // A node that receives audio by *role* counts even with no inbound link
+    // yet: a pixelpass capture sink is a taint root the moment it exists,
+    // and its owner's re-emitting leg must be bridged from it immediately.
+    receivers.extend(
+        snapshot
+            .nodes()
+            .filter(|node| node.role.receives_audio())
+            .map(|node| node.serial),
+    );
+    Edges { edges, receivers }
+}
+
+/// Resolved signal edges plus the set of nodes that can receive audio.
+struct Edges {
+    edges: BTreeMap<Serial, Vec<Serial>>,
+    /// ⚠️ Membership is "appears as a resolved `link.input.node`" **or**
+    /// "has a receiving role" — deliberately not role alone. Codex round 1:
+    /// a node whose `media.class` is absent or unexpected (`Other`), or an
+    /// `Audio/Source` that is really a filter output, can sit on an inbound
+    /// link carrying tainted audio; inferring "receives audio" from the role
+    /// alone left such a node unable to start an owner bridge, and its
+    /// sibling output leg stayed Eligible while re-emitting the call.
+    receivers: BTreeSet<Serial>,
+}
+
+fn propagate_links(
+    downstream: &BTreeMap<Serial, Vec<Serial>>,
+    taint: &mut BTreeMap<Serial, Reason>,
+) -> bool {
+    let mut changed = false;
+    let mut queue: VecDeque<Serial> = taint
+        .iter()
+        .filter(|(_, reason)| reason.propagates())
+        .map(|(serial, _)| *serial)
+        .collect();
+    while let Some(serial) = queue.pop_front() {
+        let Some(targets) = downstream.get(&serial) else {
+            continue;
+        };
+        for target in targets {
+            if raise(taint, *target, Reason::TaintedUpstream) {
+                changed = true;
+                queue.push_back(*target);
+            }
+        }
+    }
+    changed
+}
+
+/// The conditional owner bridge (v3.4 §6.1.1): taint crosses to an owner's
+/// other legs **only** when the tainted member is one that actually
+/// receives audio. The naive "this owner has both an input and an output
+/// leg ⇒ exclude the output" rule would exclude every app using a
+/// microphone, Firefox in a video call included.
+fn propagate_owner_bridge(
+    keys: &owner::OwnerKeyIndex,
+    components: &OwnerComponents,
+    edges: &Edges,
+    taint: &mut BTreeMap<Serial, Reason>,
+) -> bool {
+    let mut changed = false;
+    for members in components.components() {
+        let sources: BTreeSet<Serial> = members
+            .iter()
+            .copied()
+            .filter(|serial| {
+                taint.get(serial).is_some_and(|r| r.propagates())
+                    && edges.receivers.contains(serial)
+            })
+            .collect();
+        if sources.is_empty() {
+            continue;
+        }
+        for target in members {
+            if sources.contains(target) {
+                continue;
+            }
+            // Name the strongest key shared directly with any tainted
+            // member; `None` means the two are only transitively related.
+            let key = sources
+                .iter()
+                .filter_map(|source| keys.strongest_shared(*source, *target))
+                .min();
+            changed |= raise(taint, *target, Reason::TaintedOwnerBridge { key });
+        }
+    }
+    changed
+}
+
+/// Fail-closed backstop for an owner we cannot bound (v3.4 §6.1.1, final
+/// paragraph): something read tainted audio and nothing about the output
+/// legs on this box lets us enumerate which of them are its siblings, so we
+/// cannot know which one is re-emitting what it read. Exclude the output
+/// legs that are equally unbounded.
+///
+/// The trigger and the sweep, precisely (both edges hard-won across four
+/// Codex rounds):
+///
+/// - **Trigger — any tainted receiver that is not a real device node.** A
+///   tainted hardware sink is the normal case, not an anomaly (peerspeak's
+///   playback taints the default sink every recompute), so device nodes do
+///   not trip it. The source does **not** have to be unbounded: a reader
+///   with a `node.link-group` whose re-emitting leg carries none is bounded
+///   while its sibling is unfindable (round 1).
+/// - **Sweep — depends on whether any tainted reader is itself unbounded.**
+///   A *bounded* reader's siblings are exactly the outputs sharing its key,
+///   so only the unbounded outputs (which could share its unknowable-only-
+///   in-part identity) are swept; a differently-keyed output is provably a
+///   different owner. An *unbounded* reader could be **any** owner — a real
+///   process may present no PID on its reading leg (round 4) — so every
+///   output candidate is swept, real apps included.
+///
+/// **Two tiers, because a tainted reader we cannot bound is a bigger
+/// unknown than one we can** (Codex round 3 — the mirror image of the
+/// round-1 case):
+///
+/// - A *bounded* tainted reader has a strong key or a usable PID, so its
+///   siblings are exactly the output legs sharing that key. Any output leg
+///   that is *itself* bounded by a **different** key is provably a different
+///   owner and stays eligible; only unbounded output legs are its possible
+///   siblings. → exclude unbounded outputs.
+/// - An *unbounded* tainted reader has nothing that identifies its owner, so
+///   its re-emitting leg could be **any** output on the box, and no property
+///   on an output leg can prove it is unrelated. → exclude every output
+///   candidate.
+///
+///   ⚠️ I tried to narrow this to "daemon-owned outputs only", on the
+///   theory that an unbounded reader must be daemon-owned (a real app has a
+///   PID, which would bound it) so a real-PID output is provably a different
+///   owner. **Codex refuted it (round 4):** `application.process.id` is
+///   optional and client-controlled, so a real process can present *no* PID
+///   on its reading leg (unbounded) and a real PID on its output leg — one
+///   owner, spared by the narrowing, leaking the call. Only `pipewire.*`
+///   properties have protected identity; app properties cannot carry a
+///   soundness argument. So: exclude everything. The trigger is genuinely
+///   anomalous — a keyless reader actively consuming the call; EasyEffects
+///   and loopbacks carry a `node.link-group` and are *bounded*, so they do
+///   not trip this tier — and phase 5's dry run surfaces it before it can
+///   gate anything real.
+fn propagate_unresolved_owner(
+    snapshot: &GraphSnapshot,
+    keys: &owner::OwnerKeyIndex,
+    edges: &Edges,
+    taint: &mut BTreeMap<Serial, Reason>,
+) -> bool {
+    let mut has_tainted_reader = false;
+    let mut has_unbounded_tainted_reader = false;
+    for node in snapshot.nodes() {
+        let is_tainted_reader = !node.props.session_device
+            && edges.receivers.contains(&node.serial)
+            && taint.get(&node.serial).is_some_and(|r| r.propagates());
+        if is_tainted_reader {
+            has_tainted_reader = true;
+            has_unbounded_tainted_reader |= !keys.is_bounded(node.serial);
+        }
+    }
+    if !has_tainted_reader {
+        return false;
+    }
+    let mut changed = false;
+    for node in snapshot.nodes() {
+        if node.role == MediaRole::StreamOutput
+            && (has_unbounded_tainted_reader || !keys.is_bounded(node.serial))
+        {
+            changed |= raise(taint, node.serial, Reason::UnresolvedOwner);
+        }
+    }
+    changed
+}
+
+fn build_decisions(
+    snapshot: &GraphSnapshot,
+    ctx: &ExclusionCtx,
+    taint: &BTreeMap<Serial, Reason>,
+    sticky_serials: &BTreeSet<Serial>,
+) -> Decisions {
+    let mut candidates = BTreeMap::new();
+    for node in snapshot.nodes().filter(|n| n.role.is_candidate()) {
+        let sticky = sticky_serials.contains(&node.serial);
+        let eligibility = if !ctx.graph_ready {
+            Eligibility::NotEligible {
+                reason: Reason::GraphNotReady,
+                sticky: false,
+            }
+        } else if let Some(reason) = taint.get(&node.serial) {
+            Eligibility::NotEligible {
+                reason: *reason,
+                sticky,
+            }
+        } else if let Some(reason) = local_exclusion(snapshot, node) {
+            Eligibility::NotEligible {
+                reason,
+                sticky: false,
+            }
+        } else {
+            Eligibility::Eligible
+        };
+        candidates.insert(
+            node.serial,
+            NodeDecision {
+                serial: node.serial,
+                name: node.name.clone(),
+                eligibility,
+            },
+        );
+    }
+    Decisions {
+        candidates,
+        taint: taint
+            .iter()
+            .map(|(serial, reason)| {
+                (
+                    *serial,
+                    TaintEntry {
+                        reason: *reason,
+                        sticky: sticky_serials.contains(serial),
+                    },
+                )
+            })
+            .collect(),
+    }
+}
+
+/// Node-local reasons a link cannot be created even though the node is
+/// clean. These do not propagate — an exclusive-port stream is unlinkable,
+/// not hazardous.
+fn local_exclusion(snapshot: &GraphSnapshot, node: &NodeSnapshot) -> Option<Reason> {
+    if node.props.passthrough {
+        return Some(Reason::Passthrough);
+    }
+    if snapshot.ports_of(node.id).any(|port| port.exclusive) {
+        return Some(Reason::PortExclusive);
+    }
+    None
+}
+
+/// Sticky bookkeeping for the next recompute: every tainted owner, with
+/// every object observed to constitute it, merged with any prior entry that
+/// still overlaps. Members accumulate — that is what makes "clear only once
+/// all member objects have disappeared" true across churn.
+fn build_sticky(
+    snapshot: &GraphSnapshot,
+    keys: &owner::OwnerKeyIndex,
+    components: &OwnerComponents,
+    taint: &BTreeMap<Serial, Reason>,
+    prior: &StickyState,
+    retire_absent: bool,
+) -> StickyState {
+    let mut entries: Vec<StickyOwner> = Vec::new();
+
+    // Carry forward prior entries that still have at least one live member.
+    // An entry with none is gone for good: serials never recycle, so a
+    // vanished member can never come back — but only a *trustworthy*
+    // snapshot is allowed to conclude that a member is absent.
+    for entry in &prior.owners {
+        if !retire_absent
+            || entry
+                .members
+                .iter()
+                .any(|member| is_live(snapshot, *member))
+        {
+            entries.push(entry.clone());
+        }
+    }
+
+    for members in components.components() {
+        let node_reasons: BTreeMap<Serial, Reason> = members
+            .iter()
+            .filter_map(|serial| {
+                taint
+                    .get(serial)
+                    .filter(|reason| reason.propagates())
+                    .map(|reason| (*serial, *reason))
+            })
+            .collect();
+        if node_reasons.is_empty() {
+            continue;
+        }
+        let mut refs: BTreeSet<ObjectRef> = members.iter().map(|s| ObjectRef::Node(*s)).collect();
+        refs.extend(
+            owner::client_serials_of(snapshot, keys, members)
+                .into_iter()
+                .map(ObjectRef::Client),
+        );
+        let fingerprints = members
+            .iter()
+            .flat_map(|serial| keys.fingerprints(*serial))
+            .collect();
+        entries.push(StickyOwner {
+            members: refs,
+            fingerprints,
+            node_reasons,
+        });
+    }
+
+    StickyState {
+        owners: merge_overlapping(entries),
+    }
+}
+
+fn is_live(snapshot: &GraphSnapshot, member: ObjectRef) -> bool {
+    match member {
+        ObjectRef::Node(serial) => snapshot.node(serial).is_some(),
+        ObjectRef::Client(serial) => snapshot.clients().any(|c| c.serial == serial),
+    }
+}
+
+/// Merge entries that share any member, keeping the strongest reason.
+/// Owners fuse over time (a component that gains a leg belonging to a
+/// previously separate sticky owner is one owner now); splitting them back
+/// apart would drop taint, which is the unsafe direction.
+fn merge_overlapping(mut entries: Vec<StickyOwner>) -> Vec<StickyOwner> {
+    let mut merged: Vec<StickyOwner> = Vec::new();
+    while let Some(mut entry) = entries.pop() {
+        let mut absorbed = true;
+        while absorbed {
+            absorbed = false;
+            let mut rest = Vec::with_capacity(entries.len());
+            for other in entries.drain(..) {
+                if entry.members.is_disjoint(&other.members) {
+                    rest.push(other);
+                } else {
+                    for (serial, reason) in other.node_reasons {
+                        entry
+                            .node_reasons
+                            .entry(serial)
+                            .and_modify(|existing| {
+                                if reason.priority() < existing.priority() {
+                                    *existing = reason;
+                                }
+                            })
+                            .or_insert(reason);
+                    }
+                    entry.members.extend(other.members);
+                    entry.fingerprints.extend(other.fingerprints);
+                    absorbed = true;
+                }
+            }
+            entries = rest;
+        }
+        merged.push(entry);
+    }
+    merged.sort_by(|a, b| a.members.iter().next().cmp(&b.members.iter().next()));
+    merged
+}
+
+/// Record `reason` for `serial` if it is new or strictly stronger than what
+/// is already recorded. Returns whether anything changed — the fixpoint's
+/// termination argument rests on this being monotone.
+fn raise(taint: &mut BTreeMap<Serial, Reason>, serial: Serial, reason: Reason) -> bool {
+    match taint.get(&serial) {
+        Some(existing) if existing.priority() <= reason.priority() => false,
+        _ => {
+            taint.insert(serial, reason);
+            true
+        }
+    }
+}

src/host/taint/owner.rs

@@ -0,0 +1,390 @@
+//! The owner bridge — grouping nodes that belong to the same *owner* even
+//! though the graph shows no Link between them.
+//!
+//! This is the subtlest part of the design (v3.4 §6.1.2). Measured fact it
+//! exists to handle: a `module-loopback` forwarder's input leg and output
+//! leg have **no Link between them**, so walking Links alone from the
+//! leaking output leg finds no inbound links at all — a dead end that reads
+//! as "clean". The legs are related only by shared properties.
+//!
+//! ## The rule
+//!
+//! A union of keys, strongest first:
+//!
+//! | # | key | scope |
+//! | --- | --- | --- |
+//! | 1 | `node.link-group` | per module/filter instance |
+//! | 2 | `pulse.module.id` | per pactl module |
+//! | 3 | `client.id` | per **connection** |
+//! | 4 | `application.process.id` | per process |
+//!
+//! ⚠️ **"Resolves" means the two legs carry the key AND the values are
+//! EQUAL — not "the first key present".** A first-present implementation
+//! reproduces the exact measured leak: for `gst-launch pulsesrc ! pulsesink`
+//! both legs carry `client.id` (209 and 210) but the values *differ*, so
+//! first-present stops at key 3, sees a mismatch, and concludes "different
+//! owners". The legs are in fact one process (`application.process.id`
+//! 20172 on both). So: try each key in order, and a key resolves only if
+//! both legs carry it and the values are equal; otherwise fall through.
+//!
+//! ## Two exceptions, both guarding against mass over-exclusion
+//!
+//! 1. **Never bridge on key 4 when the value is pipewire-pulse's own PID**
+//!    (v3.4 §6.1.2). Module-created streams all carry the daemon's PID, so
+//!    bridging on it fuses every Pulse module into one owner and a single
+//!    tainted module input would exclude every module-created stream on the
+//!    box. Keys 1 and 2 already cover those cases precisely.
+//!
+//! 2. **Coarse keys (3 and 4) may not bridge nodes exported from a real
+//!    `Device`** — i.e. nodes carrying `device.id`. ⚠️ This rule is *not*
+//!    in design v3.4; it was found while implementing, and it is the exact
+//!    analogue of exception 1 for the session manager.
+//!    ✅ **MEASURED on the live graph 2026-07-21:**
+//!
+//!    | node | `client.id` | `device.id` | `factory.name` |
+//!    | --- | --- | --- | --- |
+//!    | 5 × `alsa_{output,input}.*` | **42** (`WirePlumber [export]`) | 43/45/46 | `api.alsa.pcm.{sink,source}` |
+//!    | 3 × `sink-sunshine-*` | 83 / 86 / 92 (each its own) | **absent** | `support.null-audio-sink` |
+//!
+//!    So one shared coarse key genuinely does relate every hardware device
+//!    on the box, and `device.id` cleanly separates that set from virtual
+//!    sinks. Without the rule, the hardware sink carrying peerspeak's
+//!    playback (tainted by design, every single recompute) would bridge to
+//!    *every other device node including the microphone source*, whose
+//!    readers would then taint their owners' playback legs — reproducing
+//!    precisely the §6.1.1 catastrophe ("excludes any app using a
+//!    microphone") through a different door.
+//!
+//!    ⚠️ **Keyed on `device.id`, NOT on `media.class` being `Audio/Sink`.**
+//!    The first cut suppressed coarse keys for every device-*role* node,
+//!    and Codex refuted it: a **native virtual sink** — an app that creates
+//!    an `Audio/Sink` plus a re-emitting stream on one client, with no
+//!    `link-group` and no `pulse.module.id` — would then have had its only
+//!    correlation stripped, and it would have leaked the whole call. Such a
+//!    sink has no `device.id`, so it now bridges on `client.id` as it
+//!    should.
+//!
+//! Grouping is **transitive** (union-find). That is the fail-closed
+//! direction: bigger owner components mean more taint, never less.
+
+use std::collections::BTreeMap;
+
+use super::snapshot::{GlobalId, GraphSnapshot, NodeSnapshot, Serial};
+
+/// Which key bridged two legs. Ordered strongest first; the `Ord` derive is
+/// load-bearing for "report the strongest shared key".
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
+pub enum OwnerKey {
+    LinkGroup,
+    PulseModuleId,
+    ClientId,
+    ProcessId,
+}
+
+impl OwnerKey {
+    /// Stable, machine-readable — this ends up in the phase 5 audit output
+    /// and the phase 6 status event.
+    pub fn code(self) -> &'static str {
+        match self {
+            Self::LinkGroup => "node.link-group",
+            Self::PulseModuleId => "pulse.module.id",
+            Self::ClientId => "client.id",
+            Self::ProcessId => "application.process.id",
+        }
+    }
+}
+
+/// The value a node presents for a given key, if it presents one at all.
+#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
+enum KeyValue {
+    Text(String),
+    Num(u64),
+}
+
+/// Owner keys usable on this node, strongest first.
+///
+/// A key that is present but unusable (the pipewire-pulse PID; a coarse key
+/// on a device node) is **absent** here — that is the whole mechanism of the
+/// two exceptions.
+fn keys_of(node: &NodeSnapshot, pipewire_pulse_pid: Option<u32>) -> Vec<(OwnerKey, KeyValue)> {
+    let mut out = Vec::new();
+    if let Some(group) = &node.props.link_group {
+        out.push((OwnerKey::LinkGroup, KeyValue::Text(group.clone())));
+    }
+    if let Some(module) = node.props.pulse_module_id {
+        out.push((OwnerKey::PulseModuleId, KeyValue::Num(module)));
+    }
+    // Exception 2: coarse keys never bridge passive session-manager device
+    // nodes — they all share the session manager's client.
+    if node.props.session_device {
+        return out;
+    }
+    if let Some(client) = node.props.client_id {
+        out.push((OwnerKey::ClientId, KeyValue::Num(u64::from(client.0))));
+    }
+    if let Some(pid) = node.props.process_id {
+        // Exception 1. Note the fail-closed asymmetry when the daemon PID is
+        // unknown (`None`): the exception does *not* fire, key 4 applies to
+        // everything, and Pulse modules fuse into one owner. That is broad
+        // over-exclusion — annoying and safe — which is the direction v3.4
+        // §6.1.2's failure-mode paragraph asks for.
+        if Some(pid) != pipewire_pulse_pid {
+            out.push((OwnerKey::ProcessId, KeyValue::Num(u64::from(pid))));
+        }
+    }
+    out
+}
+
+/// Can this node's owner be positively bounded — i.e. can we enumerate its
+/// sibling legs and be right?
+///
+/// ⚠️ Not the same as "has any usable key", and the difference is a leak.
+/// `client.id` alone does **not** bound an owner: that is the measured
+/// GStreamer refutation, where one process presented two different
+/// `client.id`s for its two legs. So an owner is bounded only by a strong
+/// key (link-group / pulse.module.id) or by a *usable* process id — usable
+/// meaning key 4 was not suppressed as pipewire-pulse's own PID.
+///
+/// The case this exists for is v3.4 §12's "module forwarder with neither
+/// `link-group` nor `pulse.module.id`": its process id is the daemon's and
+/// therefore suppressed, its two legs may carry different `client.id`s, and
+/// nothing else relates them. Its sibling output leg cannot be found, so
+/// the engine must fail closed rather than declare it clean
+/// (v3.4 §6.1.1, final paragraph).
+pub fn owner_is_bounded(node: &NodeSnapshot, pipewire_pulse_pid: Option<u32>) -> bool {
+    keys_of(node, pipewire_pulse_pid)
+        .iter()
+        .any(|(key, _)| *key != OwnerKey::ClientId)
+}
+
+/// Owner keys computed once per snapshot.
+///
+/// `keys_of` allocates a `Vec` and clones the `link-group` string, and the
+/// bridge asks for keys once per (tainted member × component member) pair —
+/// so recomputing was the hot spot in an otherwise linear pass.
+#[derive(Debug, Default)]
+pub struct OwnerKeyIndex {
+    keys: BTreeMap<Serial, Vec<(OwnerKey, KeyValue)>>,
+}
+
+impl OwnerKeyIndex {
+    pub fn build(snapshot: &GraphSnapshot, pipewire_pulse_pid: Option<u32>) -> Self {
+        Self {
+            keys: snapshot
+                .nodes()
+                .map(|node| (node.serial, keys_of(node, pipewire_pulse_pid)))
+                .collect(),
+        }
+    }
+
+    /// The strongest key these two nodes share directly, if any.
+    pub fn strongest_shared(&self, a: Serial, b: Serial) -> Option<OwnerKey> {
+        let (Some(a_keys), Some(b_keys)) = (self.keys.get(&a), self.keys.get(&b)) else {
+            return None;
+        };
+        // Stored strongest-first, so the first match is the strongest.
+        a_keys.iter().find_map(|(key, value)| {
+            b_keys
+                .iter()
+                .any(|(other_key, other_value)| other_key == key && other_value == value)
+                .then_some(*key)
+        })
+    }
+
+    /// Is `client.id` a usable owner key for this node?
+    ///
+    /// ⚠️ Load-bearing for sticky state. A device node's `client.id` is
+    /// suppressed by exception 2, so recording the session manager's Client
+    /// as a *member* of a tainted device's sticky owner would smuggle the
+    /// suppressed key back in: the next recompute would expand that Client
+    /// to every hardware node on the box — the microphone included — and
+    /// the §6.1.1 catastrophe would arrive one epoch late instead of never.
+    /// (Codex round 2, finding 1.)
+    pub fn uses_client_key(&self, serial: Serial) -> bool {
+        self.keys
+            .get(&serial)
+            .is_some_and(|keys| keys.iter().any(|(key, _)| *key == OwnerKey::ClientId))
+    }
+
+    /// The owner keys that are safe to remember *across* connections, for
+    /// sticky taint: the strong keys plus a usable process id.
+    ///
+    /// `client.id` is deliberately excluded — it identifies a *connection*,
+    /// and the whole point of a fingerprint is to survive one process
+    /// closing a connection and opening another. A live Client member is
+    /// what covers the same-connection case, precisely.
+    ///
+    /// These are recyclable strings and numbers, so they are only ever
+    /// applied while some **serial** member of the owner is still live
+    /// (v3.4 §6.1.3): while the process is alive, its PID cannot have been
+    /// handed to anyone else.
+    pub fn fingerprints(&self, serial: Serial) -> Vec<Fingerprint> {
+        self.keys
+            .get(&serial)
+            .map(|keys| {
+                keys.iter()
+                    .filter(|(key, _)| *key != OwnerKey::ClientId)
+                    .map(|(key, value)| Fingerprint(*key, value.clone()))
+                    .collect()
+            })
+            .unwrap_or_default()
+    }
+
+    /// Does this node currently present `fingerprint`?
+    pub fn has_fingerprint(&self, serial: Serial, fingerprint: &Fingerprint) -> bool {
+        self.keys.get(&serial).is_some_and(|keys| {
+            keys.iter()
+                .any(|(key, value)| *key == fingerprint.0 && *value == fingerprint.1)
+        })
+    }
+
+    /// See [`owner_is_bounded`].
+    pub fn is_bounded(&self, serial: Serial) -> bool {
+        self.keys
+            .get(&serial)
+            .is_some_and(|keys| keys.iter().any(|(key, _)| *key != OwnerKey::ClientId))
+    }
+}
+
+/// The strongest key two nodes share, or `None` if they share none. Used to
+/// *name* the key in a bridge decision; membership itself is transitive and
+/// comes from [`OwnerComponents`].
+pub fn strongest_shared_key(
+    a: &NodeSnapshot,
+    b: &NodeSnapshot,
+    pipewire_pulse_pid: Option<u32>,
+) -> Option<OwnerKey> {
+    let a_keys = keys_of(a, pipewire_pulse_pid);
+    let b_keys = keys_of(b, pipewire_pulse_pid);
+    // `keys_of` yields strongest-first, so the first match is the strongest.
+    a_keys.iter().find_map(|(key, value)| {
+        b_keys
+            .iter()
+            .any(|(other_key, other_value)| other_key == key && other_value == value)
+            .then_some(*key)
+    })
+}
+
+/// A remembered owner key — see [`OwnerKeyIndex::fingerprints`].
+#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Debug)]
+pub struct Fingerprint(OwnerKey, KeyValue);
+
+/// Nodes partitioned into owner components.
+#[derive(Clone, Debug, Default)]
+pub struct OwnerComponents {
+    /// node serial → component index.
+    of_node: BTreeMap<Serial, usize>,
+    /// component index → member node serials, ascending.
+    members: Vec<Vec<Serial>>,
+}
+
+impl OwnerComponents {
+    pub fn build(snapshot: &GraphSnapshot, pipewire_pulse_pid: Option<u32>) -> Self {
+        let serials: Vec<Serial> = snapshot.nodes().map(|n| n.serial).collect();
+        let index: BTreeMap<Serial, usize> =
+            serials.iter().enumerate().map(|(i, s)| (*s, i)).collect();
+        let mut uf = UnionFind::new(serials.len());
+
+        // Group by (key, value) and union within each group. Equivalent to
+        // the pairwise "some key resolves" rule, and O(n log n).
+        let mut buckets: BTreeMap<(OwnerKey, KeyValue), Vec<usize>> = BTreeMap::new();
+        for node in snapshot.nodes() {
+            let slot = index[&node.serial];
+            for (key, value) in keys_of(node, pipewire_pulse_pid) {
+                buckets.entry((key, value)).or_default().push(slot);
+            }
+        }
+        for group in buckets.values() {
+            for pair in group.windows(2) {
+                uf.union(pair[0], pair[1]);
+            }
+        }
+
+        // Compact roots into dense component indices, deterministically.
+        let mut root_to_component: BTreeMap<usize, usize> = BTreeMap::new();
+        let mut members: Vec<Vec<Serial>> = Vec::new();
+        let mut of_node = BTreeMap::new();
+        for (slot, serial) in serials.iter().enumerate() {
+            let root = uf.find(slot);
+            let component = *root_to_component.entry(root).or_insert_with(|| {
+                members.push(Vec::new());
+                members.len() - 1
+            });
+            members[component].push(*serial);
+            of_node.insert(*serial, component);
+        }
+        Self { of_node, members }
+    }
+
+    pub fn component_of(&self, serial: Serial) -> Option<usize> {
+        self.of_node.get(&serial).copied()
+    }
+
+    /// Member serials of the component containing `serial`, including it.
+    /// Empty if the node is not in this snapshot.
+    pub fn members_with(&self, serial: Serial) -> &[Serial] {
+        match self.component_of(serial) {
+            Some(component) => &self.members[component],
+            None => &[],
+        }
+    }
+
+    pub fn components(&self) -> impl Iterator<Item = &[Serial]> {
+        self.members.iter().map(Vec::as_slice)
+    }
+}
+
+struct UnionFind {
+    parent: Vec<usize>,
+}
+
+impl UnionFind {
+    fn new(len: usize) -> Self {
+        Self {
+            parent: (0..len).collect(),
+        }
+    }
+
+    fn find(&mut self, mut node: usize) -> usize {
+        while self.parent[node] != node {
+            self.parent[node] = self.parent[self.parent[node]];
+            node = self.parent[node];
+        }
+        node
+    }
+
+    fn union(&mut self, a: usize, b: usize) {
+        let (a, b) = (self.find(a), self.find(b));
+        if a != b {
+            // Lowest root wins, so components are deterministic.
+            let (low, high) = if a < b { (a, b) } else { (b, a) };
+            self.parent[high] = low;
+        }
+    }
+}
+
+/// Client objects belonging to an owner component, so sticky taint can be
+/// keyed on every object that constitutes the owner (v3.4 §6.1.3: clear the
+/// entry only once **all** member objects are gone).
+pub fn client_serials_of(
+    snapshot: &GraphSnapshot,
+    keys: &OwnerKeyIndex,
+    nodes: &[Serial],
+) -> Vec<Serial> {
+    let mut out: Vec<Serial> = nodes
+        .iter()
+        // Only nodes for which `client.id` is a *usable* owner key. See
+        // `uses_client_key`: recording a device node's shared session-manager
+        // Client here would defeat exception 2 on the next recompute.
+        .filter(|serial| keys.uses_client_key(**serial))
+        .filter_map(|serial| snapshot.node(*serial))
+        .filter_map(|node| node.props.client_id)
+        // An ambiguous client id means two Clients claim it and we cannot
+        // say which one is ours, so remember both: an entry that recorded
+        // neither could be retired while its owner was still live.
+        .flat_map(|id: GlobalId| snapshot.clients_with_id(id).map(|client| client.serial))
+        .collect();
+    out.sort_unstable();
+    out.dedup();
+    out
+}

src/host/taint/snapshot.rs

@@ -0,0 +1,332 @@
+//! The plain, owned graph model the taint engine reasons over.
+//!
+//! **No PipeWire types appear in this file, by design** (impl plan §4,
+//! phase 2). The registry observer (phase 3) translates live globals into
+//! these structs; every test builds them by hand. Nothing here ever links
+//! against libpipewire.
+//!
+//! Two id-ish things live in this model and confusing them is the bug the
+//! whole file is shaped to prevent:
+//!
+//! - [`Serial`] — `object.serial`, 64-bit, monotonic, **never reused**.
+//!   This is *identity*. Sticky taint is keyed on it.
+//! - [`GlobalId`] — the PipeWire global id, 32-bit and **recycled**. It is
+//!   a *lookup key within one snapshot* and nothing else: links name their
+//!   endpoints with it, nodes name their client with it. It must never
+//!   outlive the snapshot it was read from (design v3.4 §6.1.3).
+
+use std::collections::BTreeMap;
+
+/// `object.serial` — 64-bit, monotonic, never recycled. Identity.
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
+pub struct Serial(pub u64);
+
+/// A PipeWire global id — 32-bit and **recycled**. Snapshot-local lookup
+/// key only; see the module docs.
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
+pub struct GlobalId(pub u32);
+
+/// What a node does with audio, parsed from `media.class`.
+///
+/// Taint is computed at **node** granularity (v3.4 §6.1 edge type 2: the
+/// monitor connection is already a real Link whose output node is the sink
+/// itself, so a node-level walk crosses `app → sink → monitor-reader` for
+/// free). Ports exist in the model for link creation in phase 6 and for the
+/// `port.exclusive` predicate, not for taint.
+#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug, Hash)]
+pub enum MediaRole {
+    /// `Stream/Output/Audio` — an application playing audio. The only
+    /// fan-out candidate.
+    StreamOutput,
+    /// `Stream/Input/Audio` — an application capturing audio.
+    StreamInput,
+    /// `Audio/Sink` — a real or virtual sink.
+    Sink,
+    /// `Audio/Source` — a real or virtual source.
+    Source,
+    /// `Audio/Duplex`. ⚠️ Node granularity smears taint across both roles
+    /// of these; accepted for v1 as fail-closed over-exclusion
+    /// (v3.4 §6.1, edge type 2 caveat).
+    Duplex,
+    /// Anything else, including video and unparseable/absent `media.class`.
+    Other,
+}
+
+impl MediaRole {
+    pub fn parse(media_class: Option<&str>) -> Self {
+        match media_class {
+            Some("Stream/Output/Audio") => Self::StreamOutput,
+            Some("Stream/Input/Audio") => Self::StreamInput,
+            Some("Audio/Sink") => Self::Sink,
+            Some("Audio/Source") => Self::Source,
+            Some("Audio/Duplex") => Self::Duplex,
+            _ => Self::Other,
+        }
+    }
+
+    /// Can this node *receive* audio? This is the gate on the owner bridge:
+    /// taint crosses the intra-process hop only when the owner is actually
+    /// reading tainted audio (v3.4 §6.1.1 — "this client has both an input
+    /// and an output leg ⇒ exclude the output" is the catastrophic rule
+    /// that excludes every app with a microphone).
+    ///
+    /// `Sink` counts: EasyEffects' `ee_sink` is an `Audio/Sink` that
+    /// receives the tainted mix, and its re-emitting leg is joined to it by
+    /// `node.link-group` with no Link between them.
+    pub fn receives_audio(self) -> bool {
+        matches!(self, Self::StreamInput | Self::Sink | Self::Duplex)
+    }
+
+    /// Device-ish nodes — everything that is not a `Stream/*`. Coarse owner
+    /// keys are not allowed to bridge these; see [`super::owner`].
+    pub fn is_device_role(self) -> bool {
+        matches!(self, Self::Sink | Self::Source | Self::Duplex)
+    }
+
+    /// Only `Stream/Output/Audio` nodes are fan-out candidates (v3.4 §6.2).
+    pub fn is_candidate(self) -> bool {
+        matches!(self, Self::StreamOutput)
+    }
+}
+
+/// The subset of node properties the engine actually reasons about.
+///
+/// Deliberately a struct of parsed fields rather than a property bag: the
+/// parsing (and its failure modes) belongs at the observer boundary, and a
+/// bag invites `props.get("...")` typos that silently read `None` — which
+/// on this feature means "not tainted".
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct NodeProps {
+    /// `peerspeak.owned` is present and truthy (v3.4 §5.1). A correctness
+    /// mechanism, explicitly *not* a security boundary.
+    pub peerspeak_owned: bool,
+    /// `pulse.module.id`, parsed as `u64` — never `u32`, per v3.4 §5.2's
+    /// parse-defensively note and the phase 0a truncation bug.
+    pub pulse_module_id: Option<u64>,
+    /// `node.link-group` — owner key 1, and the `echo-cancel-` hazard
+    /// prefix (v3.4 §5.4 / D3).
+    pub link_group: Option<String>,
+    /// `client.id` — owner key 3. A **connection**, not an owner: GStreamer
+    /// opens one per stream (v3.4 §6.1.2, measured refutation).
+    pub client_id: Option<GlobalId>,
+    /// `application.process.id` **on the node** — owner key 4. For
+    /// module-created streams this is pipewire-pulse's own PID, which is
+    /// why [`super::ExclusionCtx::pipewire_pulse_pid`] exists.
+    pub process_id: Option<u32>,
+    /// The stream negotiated an encoded/passthrough format; a second link
+    /// would refuse or corrupt it (v3.4 §6.2).
+    pub passthrough: bool,
+    /// This node is a **passive device node exported by the session
+    /// manager** — a real sound card's sink or source, not something that
+    /// forwards audio.
+    ///
+    /// ⚠️ **A positive high-confidence classification the observer owes, not
+    /// a raw property** (Codex rounds 2–3). PipeWire defines `device.id`
+    /// only as "the Device this node belongs to" and `device.api` as that
+    /// Device's access API; **neither promises the node passively terminates
+    /// audio**, so a card-associated filter can satisfy both. Setting this
+    /// flag *removes* two protections at once — the node's coarse owner keys
+    /// (`owner` exception 2) and its ability to trip the fail-closed
+    /// backstop — so a false positive is a leak, not over-exclusion.
+    ///
+    /// **Phase-3 contract:**
+    /// - Set `true` only on positively-identified passive hardware
+    ///   terminals: a resolved `device.id` on a real backend
+    ///   (`device.api` present) whose `factory.name` is on an **explicit
+    ///   hardware-PCM allowlist** — `api.alsa.pcm.sink`, `api.alsa.pcm.source`,
+    ///   and the equivalent for other real backends (bluez5, v4l2 for the
+    ///   media case) as phase 3 enumerates them — never a filter, loopback,
+    ///   or `support.null-audio-sink` factory. An allowlist, not a
+    ///   substring or a denylist: an unknown factory is not a device.
+    ///   Measured discriminator on the
+    ///   target box: the five ALSA nodes carry `device.api=alsa` +
+    ///   `factory.name=api.alsa.pcm.*` and share `client.id=42`
+    ///   (`WirePlumber [export]`); the three `support.null-audio-sink` nodes
+    ///   carry neither. (`node.physical` was measured **null** on the ALSA
+    ///   nodes here, so it is *not* a usable discriminator — do not rely on
+    ///   it.)
+    /// - **Fail closed: unknown ⇒ `false`.** A node that cannot be
+    ///   positively classified keeps its owner keys and can trip the
+    ///   backstop; both are the safe direction.
+    /// - A node MUST NOT enter a snapshot with this field provisional. If
+    ///   the Device backing a node has not yet been bound, withhold the node
+    ///   and keep the epoch not-ready — otherwise a provisional `false`
+    ///   during not-ready fuses sink and mic on the shared session client
+    ///   and that fusion can persist as sticky over-exclusion (round-3
+    ///   finding 3).
+    ///
+    /// ⚠️ **A false positive is leak-capable — do not treat it as braced.**
+    /// I claimed a mis-classified filter could not leak because its legs
+    /// share a `node.link-group` (strong-key bridge) or trip the unbounded
+    /// backstop. Codex refuted it (round 4): a filter *without* a shared
+    /// strong key, marked `session_device=true`, cannot activate the
+    /// backstop from its reading leg, so a differently-keyed re-emitting leg
+    /// leaks. Those braces catch *some* shapes, not all. The only real
+    /// defence is a correct classifier — hence "positive high-confidence"
+    /// and "fail closed to false" above, without exception.
+    ///
+    /// What it is for: every real device node shares the session manager's
+    /// `client.id`, so coarse owner keys must not bridge them — else
+    /// peerspeak's playback (which taints the default sink every recompute)
+    /// would reach the microphone. See [`super::owner`] exception 2.
+    pub session_device: bool,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct NodeSnapshot {
+    pub serial: Serial,
+    pub id: GlobalId,
+    /// `node.name`, for diagnostics and for `pixelpass_capture_*` ancestry
+    /// detection (v3.4 §6.2, cycle prevention).
+    pub name: Option<String>,
+    pub role: MediaRole,
+    pub props: NodeProps,
+}
+
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub enum PortDirection {
+    In,
+    Out,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct PortSnapshot {
+    pub serial: Serial,
+    pub id: GlobalId,
+    /// Owning node, by snapshot-local id.
+    pub node: GlobalId,
+    pub direction: PortDirection,
+    /// `port.exclusive` — fan-out will be refused (v3.4 §6.2).
+    pub exclusive: bool,
+    /// `port.monitor`. Recorded for phase 6 link creation; taint does not
+    /// need it at node granularity.
+    pub monitor: bool,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct LinkSnapshot {
+    pub serial: Serial,
+    pub id: GlobalId,
+    /// `link.output.node` — the node audio flows **from**.
+    pub output_node: GlobalId,
+    /// `link.input.node` — the node audio flows **to**.
+    pub input_node: GlobalId,
+    pub output_port: Option<GlobalId>,
+    pub input_port: Option<GlobalId>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ClientSnapshot {
+    pub serial: Serial,
+    pub id: GlobalId,
+    /// `pipewire.sec.pid` — for Pulse-emulated clients this is
+    /// **pipewire-pulse's** PID, identical across every unrelated app
+    /// (v3.4 §5.2 correction 5). Phase 3 derives the daemon PID from the
+    /// consistency of this value; the engine only consumes the result.
+    pub sec_pid: Option<u32>,
+}
+
+/// How a snapshot-local id resolves.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub enum IdLookup {
+    Unique(Serial),
+    /// Two live objects in one snapshot claim the same global id — the
+    /// observer missed a removal, so the recycled id is ambiguous. Every
+    /// edge touching it is treated as unresolved, i.e. fail closed.
+    Ambiguous,
+}
+
+/// One coherent observation of the graph.
+///
+/// Built through [`GraphSnapshot::new`] so the id indexes and the ambiguity
+/// detection cannot be skipped.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct GraphSnapshot {
+    nodes: BTreeMap<Serial, NodeSnapshot>,
+    ports: BTreeMap<Serial, PortSnapshot>,
+    links: BTreeMap<Serial, LinkSnapshot>,
+    clients: BTreeMap<Serial, ClientSnapshot>,
+    node_ids: BTreeMap<GlobalId, IdLookup>,
+    client_ids: BTreeMap<GlobalId, IdLookup>,
+}
+
+impl GraphSnapshot {
+    pub fn new(
+        nodes: Vec<NodeSnapshot>,
+        ports: Vec<PortSnapshot>,
+        links: Vec<LinkSnapshot>,
+        clients: Vec<ClientSnapshot>,
+    ) -> Self {
+        let node_ids = index_ids(nodes.iter().map(|n| (n.id, n.serial)));
+        let client_ids = index_ids(clients.iter().map(|c| (c.id, c.serial)));
+        Self {
+            nodes: nodes.into_iter().map(|n| (n.serial, n)).collect(),
+            ports: ports.into_iter().map(|p| (p.serial, p)).collect(),
+            links: links.into_iter().map(|l| (l.serial, l)).collect(),
+            clients: clients.into_iter().map(|c| (c.serial, c)).collect(),
+            node_ids,
+            client_ids,
+        }
+    }
+
+    pub fn nodes(&self) -> impl Iterator<Item = &NodeSnapshot> {
+        self.nodes.values()
+    }
+
+    pub fn node(&self, serial: Serial) -> Option<&NodeSnapshot> {
+        self.nodes.get(&serial)
+    }
+
+    pub fn links(&self) -> impl Iterator<Item = &LinkSnapshot> {
+        self.links.values()
+    }
+
+    pub fn ports(&self) -> impl Iterator<Item = &PortSnapshot> {
+        self.ports.values()
+    }
+
+    pub fn clients(&self) -> impl Iterator<Item = &ClientSnapshot> {
+        self.clients.values()
+    }
+
+    /// Resolve a snapshot-local node id. `None` means "no such node in this
+    /// snapshot", which for a link endpoint means unresolved ancestry.
+    pub fn node_by_id(&self, id: GlobalId) -> Option<IdLookup> {
+        self.node_ids.get(&id).copied()
+    }
+
+    pub fn client_by_id(&self, id: GlobalId) -> Option<IdLookup> {
+        self.client_ids.get(&id).copied()
+    }
+
+    /// Every node claiming a global id. More than one means the id is
+    /// [`IdLookup::Ambiguous`] and each claimant must be treated as a
+    /// possible endpoint of any link naming it.
+    pub fn nodes_with_id(&self, id: GlobalId) -> impl Iterator<Item = &NodeSnapshot> {
+        self.nodes.values().filter(move |node| node.id == id)
+    }
+
+    /// Every client claiming a global id — same fail-closed reasoning.
+    pub fn clients_with_id(&self, id: GlobalId) -> impl Iterator<Item = &ClientSnapshot> {
+        self.clients.values().filter(move |client| client.id == id)
+    }
+
+    /// Ports belonging to a node, by the node's snapshot-local id.
+    pub fn ports_of(&self, node: GlobalId) -> impl Iterator<Item = &PortSnapshot> {
+        self.ports.values().filter(move |p| p.node == node)
+    }
+}
+
+fn index_ids(entries: impl Iterator<Item = (GlobalId, Serial)>) -> BTreeMap<GlobalId, IdLookup> {
+    let mut out: BTreeMap<GlobalId, IdLookup> = BTreeMap::new();
+    for (id, serial) in entries {
+        out.entry(id)
+            .and_modify(|slot| {
+                if *slot != IdLookup::Unique(serial) {
+                    *slot = IdLookup::Ambiguous;
+                }
+            })
+            .or_insert(IdLookup::Unique(serial));
+    }
+    out
+}

src/interactive.rs

@@ -281,8 +281,10 @@ impl Player {
                 &[
                     // No `--untimed`: it ignores audio timestamps and drifts a
                     // shared video out of sync. Pacing to audio keeps A/V synced.
+                    // Also leave hwdec at the `low-latency` default (software
+                    // decode): forcing `--hwdec=auto` froze some viewers on
+                    // frame 1 while audio kept playing.
                     "--profile=low-latency",
-                    "--hwdec=auto",
                     "--audio-buffer=0.2",
                     "--demuxer-max-bytes=2M",
                     "--demuxer-readahead-secs=0.5",

src/main.rs

@@ -1,5 +1,6 @@
 mod cli;
 mod common;
+mod doctor;
 #[cfg(feature = "gui")]
 mod gui;
 mod host;
@@ -36,6 +37,13 @@ async fn main() -> Result<()> {
         }
     }
 
+    // Diagnostics run before pipewire::init() (they don't need it) and work
+    // regardless of the `gui` feature, so a headless tester can probe their box.
+    if cli.doctor {
+        let relay = common::endpoint::relay_override(cli.relay.as_deref());
+        return doctor::run(relay).await;
+    }
+
     // libpipewire requires global init before any pw_* call. Idempotent;
     // safe to call even when the per-app audio thread never spawns.
     pipewire::init();

src/viewer/mod.rs

@@ -102,7 +102,7 @@ fn print_viewer_banner(url: &str) {
     eprintln!("│  Connected to host. Open the stream in your player:");
     eprintln!("│");
     eprintln!(
-        "│    mpv --profile=low-latency --hwdec=auto --audio-buffer=0.2 --demuxer-max-bytes=2M --demuxer-readahead-secs=0.5 {url}"
+        "│    mpv --profile=low-latency --audio-buffer=0.2 --demuxer-max-bytes=2M --demuxer-readahead-secs=0.5 {url}"
     );
     eprintln!("│    vlc --network-caching=200 --live-caching=200 {url}");
     eprintln!("│");