repair: say "once per (pid, attribution)", because that is what it is now

Codex's non-blocking round-7 nit. The comment and test name still claimed liveness
is asked once per pid, which stopped being true when one pid became two questions.
No behaviour change; the wording was the last thing pointing at the old model.

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

Cargo.lock

@@ -2958,6 +2958,33 @@ version = "0.2.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
 
+[[package]]
+name = "libpulse-binding"
+version = "2.30.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "909eb3049e16e373680fe65afe6e2a722ace06b671250cc4849557bc57d6a397"
+dependencies = [
+ "bitflags 2.13.0",
+ "libc",
+ "libpulse-sys",
+ "num-derive",
+ "num-traits",
+ "winapi",
+]
+
+[[package]]
+name = "libpulse-sys"
+version = "1.23.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d74371848b22e989f829cc1621d2ebd74960711557d8b45cfe740f60d0a05e61"
+dependencies = [
+ "libc",
+ "num-derive",
+ "num-traits",
+ "pkg-config",
+ "winapi",
+]
+
 [[package]]
 name = "libredox"
 version = "0.1.18"
@@ -3539,6 +3566,17 @@ version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441"
 
+[[package]]
+name = "num-derive"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed3955f1a9c7c0c15e092f9c887db08b1fc683305fdf6eb6684f22555355e202"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "num-traits"
 version = "0.2.19"
@@ -4163,6 +4201,7 @@ dependencies = [
  "iroh",
  "iroh-tickets",
  "ksni",
+ "libpulse-binding",
  "nix 0.30.1",
  "notify-rust",
  "pipewire",

Cargo.toml

@@ -46,6 +46,16 @@ serde_json = "1"
 directories = "5"
 ashpd = { version = "0.9", default-features = false, features = ["tokio"] }
 pipewire = "0.9"
+# `--repair` reads and unloads Pulse modules through libpulse introspection rather
+# than by parsing `pactl` output. `pa_module_info` carries index, name and the exact
+# argument in one record, and `pa_context_is_local()` answers whether the server we
+# actually reached is local — neither of which the text listings can do (an argument
+# may contain tabs and newlines that the short format cannot escape, the JSON
+# listing carries no module index at all, and `PULSE_SERVER` is a fallback list, so
+# it never proved locality). Vetted at 2.30.1: MIT/Apache-2.0, no build script
+# beyond a pkg-config probe, no network or subprocess use, and all three historical
+# RustSec advisories (2018-0020, 2018-0021, 2019-0038) fixed by 2.6.0.
+libpulse-binding = "2.30"
 x11rb = { version = "0.13", default-features = false, features = ["allow-unsafe-code"] }
 uuid = { version = "1", features = ["v4"] }
 iroh-tickets = "1.0.0"

src/cli.rs

@@ -105,6 +105,18 @@ pub struct Cli {
     #[arg(long)]
     pub repair: bool,
 
+    /// With `--repair`: also clean up modules that carry no ownership token,
+    /// judging them by process id alone.
+    ///
+    /// Modules loaded by pixelpass versions before ownership tokens existed cannot
+    /// be attributed to a machine, boot or pid namespace, so `--repair` refuses them
+    /// by default: a process id means different processes in different namespaces,
+    /// and acting on the wrong one unloads a *running* host's audio. Use this only
+    /// on the machine that ran the crashed host, and only when the reported
+    /// candidates look right.
+    #[arg(long, requires = "repair")]
+    pub repair_legacy_untagged: 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

src/host/audio.rs

@@ -39,6 +39,7 @@ use std::sync::{Arc, Mutex};
 use std::thread::JoinHandle;
 
 use crate::cli::HostOpts;
+use crate::repair::plan::{self as repair_plan, Shape};
 
 /// Owns the pactl-loaded modules plus, when filtering is active, the
 /// libpipewire stream-router thread. Drop unloads modules as a backstop;
@@ -64,9 +65,14 @@ impl Routing {
     /// also spawn the libpipewire thread that reroutes matching streams.
     pub async fn start(opts: &HostOpts) -> Result<Self> {
         let pid = std::process::id();
-        let sink_name = format!("pixelpass_capture_{pid}");
+        let sink_name = repair_plan::sink_name_for(pid);
 
-        let sink_module = load_module(&["module-null-sink", &format!("sink_name={sink_name}")])
+        // Every module this host loads carries an ownership token, minted per
+        // load, so `--repair` can tell whose pid the name refers to instead of
+        // assuming the number means the same thing everywhere. Without it a repair
+        // run in another pid namespace can unload a live host's audio; see
+        // `repair::plan::OwnerToken`.
+        let sink_module = load_module(Shape::LegacyCaptureSink, pid)
             .context("failed to load module-null-sink")?;
 
         // In strict per-app mode we never mirror the default sink: the viewer
@@ -82,13 +88,8 @@ impl Routing {
             None
         } else {
             Some(
-                load_module(&[
+                load_module(Shape::LoopbackIntoCapture, pid)
-                    "module-loopback",
+                    .context("failed to load module-loopback (null-sink cleaned up on Drop)")?,
-                    "source=@DEFAULT_SINK@.monitor",
-                    &format!("sink={sink_name}"),
-                    "latency_msec=20",
-                ])
-                .context("failed to load module-loopback (null-sink cleaned up on Drop)")?,
             )
         };
 
@@ -115,7 +116,6 @@ impl Routing {
             let (router, mut event_rx) = StreamRouter::spawn(app.clone(), sink_name.clone())?;
             let loopback_for_task = Arc::clone(&loopback_arc);
             let local_monitor_for_task = Arc::clone(&local_monitor_arc);
-            let sink_name_for_task = sink_name.clone();
             let strict = opts.strict_audio;
             let event_task = tokio::spawn(async move {
                 use crate::common::output::{self, AppAudioState};
@@ -137,12 +137,7 @@ impl Routing {
                             // only, never the desktop/call — so it can't echo into
                             // the capture.
                             if local_monitor_for_task.lock().unwrap().is_none() {
-                                match load_module(&[
+                                match load_module(Shape::LoopbackOutOfCapture, pid) {
-                                    "module-loopback",
-                                    &format!("source={sink_name_for_task}.monitor"),
-                                    "sink=@DEFAULT_SINK@",
-                                    "latency_msec=20",
-                                ]) {
                                     Ok(id) => {
                                         tracing::info!(
                                             module = id,
@@ -195,12 +190,7 @@ impl Routing {
                             tracing::info!(
                                 "audio routing: last routed stream gone → restoring default-sink loopback"
                             );
-                            match load_module(&[
+                            match load_module(Shape::LoopbackIntoCapture, pid) {
-                                "module-loopback",
-                                "source=@DEFAULT_SINK@.monitor",
-                                &format!("sink={sink_name_for_task}"),
-                                "latency_msec=20",
-                            ]) {
                                 Ok(id) => {
                                     *loopback_for_task.lock().unwrap() = Some(id);
                                 }
@@ -349,10 +339,47 @@ struct SinkInputProperties {
 // pactl module helpers
 // ──────────────────────────────────────────────────────────────────────
 
-fn load_module(args: &[&str]) -> Result<u32> {
+/// Mint an ownership token for one module load.
+///
+/// **Per load, not per session.** The nonce is what makes two loads by the same pid
+/// render different arguments, which is what lets a fingerprint tell a module from
+/// its replacement at the same index. A token minted once and reused for every
+/// reload would be a host-session nonce and would not do that, so the counter is
+/// bumped on every call and mixed with the clock.
+fn owner_token(pid: u32) -> Result<repair_plan::OwnerToken> {
+    use std::sync::atomic::{AtomicU64, Ordering};
+    static LOADS: AtomicU64 = AtomicU64::new(0);
+
+    let local = crate::repair::local_identity()?;
+    // A nonce only has to be unlikely to repeat, not unguessable. The counter makes
+    // two loads within the same clock tick distinct; the clock keeps two runs of the
+    // same process distinct.
+    let counter = LOADS.fetch_add(1, Ordering::Relaxed);
+    let nanos = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_nanos() as u64)
+        .unwrap_or(0);
+    Ok(repair_plan::OwnerToken {
+        machine: local.machine,
+        boot: local.boot,
+        pid_ns: local.pid_ns,
+        nonce: nanos ^ (counter << 48) ^ (u64::from(pid) << 32),
+    })
+}
+
+/// Load the Pulse module for one [`Shape`] and return its index.
+///
+/// Both the module name and its arguments come from the shape itself
+/// ([`crate::repair::plan::Shape`]) rather than being written out here, so that
+/// `--repair`'s exact-form matcher and this loader are one source of truth. A
+/// latency or argument change that moved only one of them would leave repair
+/// silently unable to recognise the modules this build loads.
+fn load_module(shape: Shape, pid: u32) -> Result<u32> {
+    let owner = owner_token(pid).context("could not build an audio ownership token")?;
     let output = Command::new("pactl")
         .arg("load-module")
-        .args(args)
+        .arg(shape.module_name())
+        .args(shape.render_args(pid, Some(&owner)))
         .output()
         .context("failed to run pactl load-module")?;
     if !output.status.success() {

src/host/audit/mod.rs

@@ -74,8 +74,9 @@ use serde::Serialize;
 
 use crate::host::aec::{AecConfig, AecState, AecValidator};
 use crate::host::observer::{EventKind, Millis, Projection, Readiness};
+use crate::host::taint::owner::OwnerKey;
 use crate::host::taint::snapshot::Serial;
-use crate::host::taint::{Decisions, Eligibility, ExclusionCtx, StickyState, evaluate};
+use crate::host::taint::{Decisions, Eligibility, ExclusionCtx, Reason, StickyState, evaluate};
 
 /// How long the AEC validator may sit in `Validating` after the graph first
 /// reports ready before failing closed. Generous relative to the observer's own
@@ -181,6 +182,16 @@ pub struct AuditRow {
     pub eligible: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub reason: Option<&'static str>,
+    /// The owner key that carried the taint across, when the reason is
+    /// `tainted-owner-bridge` *and* the tainted member shared a key directly.
+    ///
+    /// §5.1 row 1 asserts "reason = owner bridge, **naming the key**" — the
+    /// point being that the exclusion is provably the owner bridge on a
+    /// specific key rather than an incidental link walk that happens to reach
+    /// the same verdict. [`Reason::code`] collapses the payload, so without
+    /// this field that row cannot be asserted from the record at all.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub owner_key: Option<&'static str>,
     /// The exclusion was carried over from a previous snapshot rather than
     /// derived from the current topology (phase-2 stickiness).
     pub sticky: bool,
@@ -195,9 +206,25 @@ pub struct TaintRow {
     pub serial: u64,
     pub name: Option<String>,
     pub reason: &'static str,
+    /// As [`AuditRow::owner_key`]. Present here too because the bridge that
+    /// matters for a row's diagnosis is often on a non-candidate node.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub owner_key: Option<&'static str>,
     pub sticky: bool,
 }
 
+/// The owner key a `tainted-owner-bridge` reason resolved on, if it named one.
+///
+/// `None` for every other reason, and also for a bridge whose tainted member
+/// shared no key *directly* — the taint reached it transitively, so there is no
+/// single key to name and inventing one would be a false diagnosis.
+fn owner_key_of(reason: Reason) -> Option<&'static str> {
+    match reason {
+        Reason::TaintedOwnerBridge { key } => key.map(OwnerKey::code),
+        _ => None,
+    }
+}
+
 /// A node carrying a peerspeak ownership carrier on a role the engine does not
 /// honour it on (round 10, R10-1). `role` is the point of the row: it says
 /// which non-producer role the tag turned up on, which is what distinguishes a
@@ -424,11 +451,13 @@ fn build_body(
             // engine would have passed — otherwise a shut gate would erase every
             // reason code in the record and the matrix would stop constraining
             // the engine at all.
-            let (eligible, reason, sticky) = match decision.eligibility {
+            let (eligible, reason, owner_key, sticky) = match decision.eligibility {
-                Eligibility::NotEligible { reason, sticky } => (false, Some(reason.code()), sticky),
+                Eligibility::NotEligible { reason, sticky } => {
+                    (false, Some(reason.code()), owner_key_of(reason), sticky)
+                }
                 Eligibility::Eligible => match gate_reason {
-                    Some(gate) => (false, Some(gate.code()), false),
+                    Some(gate) => (false, Some(gate.code()), None, false),
-                    None => (true, None, false),
+                    None => (true, None, None, false),
                 },
             };
             AuditRow {
@@ -436,6 +465,7 @@ fn build_body(
                 name: decision.name.clone(),
                 eligible,
                 reason,
+                owner_key,
                 sticky,
             }
         })
@@ -450,6 +480,7 @@ fn build_body(
             serial: serial.0,
             name: node_name(projection, serial),
             reason: entry.reason.code(),
+            owner_key: owner_key_of(entry.reason),
             sticky: entry.sticky,
         })
         .collect();

src/host/audit/tests.rs

@@ -529,6 +529,53 @@ fn row_1_owner_bridge_forwarder_with_an_untainted_control() {
     );
 }
 
+/// §5.1 row 1's other half: the record must **name the key** the bridge
+/// resolved on, not merely say "owner bridge".
+///
+/// Without this the row is unassertable from the record: `Reason::code`
+/// collapses `TaintedOwnerBridge { key }` to one string, so an exclusion that
+/// arrived by an incidental link walk and one that arrived across a named owner
+/// key are indistinguishable — and the row exists precisely to tell them apart.
+/// The fixture's forwarder legs are joined by `pulse.module.id`, so that is the
+/// key that must be reported.
+#[test]
+fn row_1_names_the_owner_key_the_bridge_resolved_on() {
+    let mut graph = Graph::new();
+    let call = graph.peerspeak_node("peerspeak-call", 200);
+    let sink = graph.module_node("tainted-null-sink", MediaRole::Sink, 30);
+    graph.link(call, sink);
+    let capture = graph.module_node("tainted-loopback-capture", MediaRole::StreamInput, 30);
+    let _playback = graph.module_node("tainted-loopback-playback", MediaRole::StreamOutput, 30);
+    graph.link(sink, capture);
+
+    let body = observe(&mut auditor_off(), &ready(graph.build()), 0)
+        .record
+        .body;
+
+    let playback = body
+        .candidates
+        .iter()
+        .find(|row| row.name.as_deref() == Some("tainted-loopback-playback"))
+        .expect("the forwarder's playback leg is a candidate");
+    assert_eq!(playback.reason, Some("tainted-owner-bridge"));
+    assert_eq!(
+        playback.owner_key,
+        Some("pulse.module.id"),
+        "the bridge key must be named in the record, not collapsed into the reason code"
+    );
+
+    // And it stays absent everywhere it would be a false diagnosis: the tag
+    // exclusion is not a bridge at all.
+    let call_name = owned_name("peerspeak-call", 200);
+    let tagged = body
+        .candidates
+        .iter()
+        .find(|row| row.name.as_deref() == Some(call_name.as_str()))
+        .expect("the tagged call playback is a candidate");
+    assert_eq!(tagged.reason, Some("peerspeak-owned"));
+    assert_eq!(tagged.owner_key, None);
+}
+
 /// §5.1 row 3: two Pulse modules, one tainted input. **The other module's output
 /// must be eligible** — this is the row that makes a wrong pipewire-pulse-PID
 /// fusion observable, because fusing all Pulse-created nodes into one owner

src/host/observer/adapter.rs

@@ -16,7 +16,7 @@ use crate::host::taint::{PEERSPEAK_OWNED_PROP, PEERSPEAK_OWNED_VALUE};
 use anyhow::{Context, Result};
 use pipewire::{self as pw, types::ObjectType};
 use std::cell::{Cell, RefCell};
-use std::collections::{BTreeMap, VecDeque};
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
 use std::rc::Rc;
 use std::sync::{Arc, Mutex};
 use std::thread::JoinHandle;
@@ -130,7 +130,9 @@ struct LiveGlobal {
 struct ObserverState {
     model: RegistryModel,
     latest: Arc<Mutex<Option<Projection>>>,
-    last_candidate: Option<u32>,
+    /// The pulse-PID candidate set as of the last probe, so only PIDs entering
+    /// it are read from `/proc`.
+    last_candidates: BTreeSet<u32>,
     live_globals: BTreeMap<GlobalId, VecDeque<LiveGlobal>>,
     sink: Option<Box<dyn ProjectionSink>>,
     started_at: Instant,
@@ -148,7 +150,7 @@ impl ObserverState {
         Self {
             model: RegistryModel::new(0, READINESS_TIMEOUT_MILLIS),
             latest,
-            last_candidate: None,
+            last_candidates: BTreeSet::new(),
             live_globals: BTreeMap::new(),
             sink,
             started_at,
@@ -163,10 +165,17 @@ impl ObserverState {
         let event_outcome = self.model.apply(event);
         let mut outcome = event_outcome;
 
-        let candidate = self.model.pulse_pid_candidate();
+        // Round 10: a *set* of candidates, because repetition across Clients
-        if candidate != self.last_candidate {
+        // turned out not to identify pipewire-pulse (see `pulse_pid`'s module
-            self.last_candidate = candidate;
+        // docs — WirePlumber repeats a PID too, which made the old single
-            if let Some(pid) = candidate {
+        // candidate permanently ambiguous on this host).
+        let candidates = self.model.pulse_pid_candidates();
+        if candidates != self.last_candidates {
+            // Only PIDs *entering* the set are probed. A PID that left and came
+            // back is "entering" again and so is re-probed, which is what keeps
+            // the PID-reuse guard honest rather than answering from a cached
+            // `comm` for a number that now belongs to someone else.
+            for &pid in candidates.difference(&self.last_candidates) {
                 let comm = std::fs::read_to_string(format!("/proc/{pid}/comm"))
                     .ok()
                     .map(|comm| comm.trim_end_matches(['\r', '\n']).to_string());
@@ -178,6 +187,8 @@ impl ObserverState {
                     outcome = Outcome::Applied;
                 }
             }
+            self.model.retain_probed_comms(&candidates);
+            self.last_candidates = candidates;
         }
 
         // v3.5 §6.7 decision 2: a projection the model proved identical is not

src/host/observer/mod.rs

@@ -91,7 +91,7 @@ use crate::host::taint::snapshot::{
     PortSnapshot, Serial,
 };
 use classify::{Classification, DeviceClaim, DeviceProps};
-use std::collections::{BTreeMap, VecDeque};
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
 
 /// A monotonic millisecond clock value, supplied by the adapter via
 /// [`RegEvent::Tick`]. Kept as a bare integer rather than
@@ -399,12 +399,25 @@ impl RegistryModel {
         matches!(self.readiness, Readiness::Complete) && !self.obligations_outstanding()
     }
 
-    /// The pulse-PID candidate the adapter should be probing (`None` = no
+    /// The pulse-PID candidates the adapter should be probing — every distinct
-    /// repeated `sec_pid`, nothing to probe). Exposed so the adapter re-probes
+    /// `sec_pid` on the current Clients. Exposed so the adapter re-probes only
-    /// only when the candidate changes.
+    /// the PIDs *entering* the set rather than all of them on every event.
-    pub fn pulse_pid_candidate(&self) -> Option<u32> {
+    pub fn pulse_pid_candidates(&self) -> BTreeSet<u32> {
         let clients: Vec<ClientSnapshot> = self.clients.values().cloned().collect();
-        pulse_pid::candidate(&clients)
+        pulse_pid::candidates(&clients)
+    }
+
+    /// Forget probed `comm`s for PIDs no Client presents any more.
+    ///
+    /// Cannot change the projection — [`Self::pulse_pid`] only ever reads
+    /// `comm`s for PIDs in the current candidate set — so it is deliberately
+    /// not an [`Outcome`]-returning `apply` arm: it must not publish, and it
+    /// must not count as a graph event for O5. Its purpose is to bound the map
+    /// (one entry per live Client PID) in a host process that runs for hours,
+    /// and to guarantee a PID that leaves and returns is re-probed rather than
+    /// answered from a stale `comm`.
+    pub fn retain_probed_comms(&mut self, live: &BTreeSet<u32>) {
+        self.probed_comm.retain(|pid, _| live.contains(pid));
     }
 
     /// Fold one observation into the model. The returned [`Outcome`] tells the
@@ -701,9 +714,9 @@ impl RegistryModel {
     /// probed `comm`. `None` whenever anything is ambiguous or unconfirmed —
     /// the safe answer (key 4 unusable).
     fn pulse_pid(&self) -> Option<u32> {
-        let candidate = self.pulse_pid_candidate()?;
+        pulse_pid::resolve(&self.pulse_pid_candidates(), |pid| {
-        let comm = self.probed_comm.get(&candidate).and_then(|c| c.as_deref());
+            self.probed_comm.get(&pid).cloned().flatten()
-        pulse_pid::validate(candidate, comm)
+        })
     }
 
     /// Project the current state into the taint engine's inputs.

src/host/observer/pulse_pid.rs

@@ -11,20 +11,49 @@
 //! The derivation is split into two pure stages so the I/O — reading
 //! `/proc/<pid>/comm` — stays in the adapter:
 //!
-//! 1. [`candidate`] finds the PID that *looks* like pulse from the graph
+//! 1. [`candidates`] lists the PIDs worth probing from the graph alone: every
-//!    alone: the `pipewire.sec.pid` value shared across multiple Clients.
+//!    distinct `pipewire.sec.pid` any Client presents.
-//!    Native PipeWire clients carry their own distinct PID; only the
+//! 2. [`resolve`] picks the one whose `comm`, as read from `/proc` by the
-//!    Pulse shim repeats one value, so a repeated value is the signal.
+//!    adapter, is exactly pipewire-pulse's. This is also what closes **PID
-//! 2. [`validate`] confirms that candidate against the `comm` the adapter
+//!    reuse**: a recycled PID is rejected because `/proc/<pid>/comm` now names
-//!    read from `/proc`. This is what closes **PID reuse**: a recycled PID
+//!    a different process.
-//!    that coincidentally repeats in the graph is rejected because
-//!    `/proc/<pid>/comm` now names a different process.
 //!
-//! Any failure at either stage — no repeated value, two repeated values,
+//! Any failure — no Client carries the property, no `comm` matches, `/proc`
-//! the property missing, `/proc` gone, a `comm` mismatch — yields `None`.
+//! gone, or *several* PIDs claim to be pipewire-pulse — yields `None`.
+//!
+//! ## ⚠️ Round 10 (MEASURED): repetition is not the signal
+//!
+//! Stage 1 used to return a single candidate: the one `sec_pid` value shared by
+//! two or more Clients, reasoning that "native PipeWire clients carry their own
+//! distinct PID; only the Pulse shim repeats one value". **That is false on a
+//! stock desktop, and the phase-5 §5.1 matrix caught it on row 1.** Measured on
+//! this host (PipeWire 1.6.8 / WirePlumber 0.5.15): WirePlumber holds *two*
+//! Clients — `WirePlumber` and `WirePlumber [export]` — both carrying
+//! `sec_pid` 1747. So two values repeated (1747 and pipewire-pulse's 2528), the
+//! old rule called that ambiguous and returned `None`, and the consequence was
+//! not a missing optimisation but a machine-wide over-exclusion cascade: with
+//! the daemon PID unknown, key 4's suppression never fires, every
+//! Pulse-emulated node fuses into one owner, and the eligible half of every row
+//! empties out (see `owner::keys_of`'s fail-closed asymmetry note).
+//!
+//! The rule failed in *both* directions, which is why the prefilter is gone
+//! rather than patched:
+//!
+//! - **False ambiguity** — any second process holding two Clients defeats it.
+//!   WirePlumber always does, so this was permanent, not a corner case.
+//! - **False absence** — a session where pipewire-pulse happens to hold exactly
+//!   one Client (one Pulse app running) never repeats a value at all, so the
+//!   candidate is missed and the same cascade follows.
+//!
+//! `comm` was always the authoritative check; repetition was a heuristic
+//! standing in front of it, and it was wrong. Probing every distinct `sec_pid`
+//! costs one `/proc` read per *distinct* PID (single digits — bounded by the
+//! Client count, cached, and re-read only when the candidate set changes),
+//! which is a cheap price for a signal that does not encode an assumption about
+//! how many Clients anyone else opens.
 
 use crate::host::taint::snapshot::ClientSnapshot;
-use std::collections::BTreeMap;
+use std::collections::BTreeSet;
 
 /// The kernel `comm` of the pipewire-pulse process. `comm` is truncated to
 /// 15 bytes by the kernel; `pipewire-pulse` is 14 bytes, so it is exact —
@@ -32,40 +61,19 @@ use std::collections::BTreeMap;
 /// recycled PID belonging to e.g. `pipewire-pulseX`.
 const PULSE_COMM: &str = "pipewire-pulse";
 
-/// Stage 1: the PID that looks like pipewire-pulse from the client graph.
+/// Stage 1: every PID worth probing — the distinct `pipewire.sec.pid` values
+/// the Clients present.
 ///
-/// Returns `Some(pid)` only when **exactly one** `pipewire.sec.pid` value is
+/// No filtering, and deliberately so (see the module docs): any rule applied
-/// shared by two or more clients. Rationale, matched to the failure matrix:
+/// here is a guess about other processes' Client counts, while stage 2 has the
-///
+/// kernel's own answer. A `BTreeSet` because the adapter diffs successive
-/// - **consistent** — one value repeats, the rest (native clients) are
+/// candidate sets to decide what to re-probe, and that diff must not depend on
-///   distinct ⇒ that value.
+/// Client iteration order.
-/// - **inconsistent** — two or more values each repeat ⇒ we cannot tell which
+pub fn candidates(clients: &[ClientSnapshot]) -> BTreeSet<u32> {
-///   is pulse ⇒ `None`.
+    clients.iter().filter_map(|client| client.sec_pid).collect()
-/// - **missing property** — the Pulse clients carry no `sec_pid` ⇒ nothing
-///   repeats ⇒ `None`.
-///
-/// A count threshold of two is deliberate: a single client carrying a PID is
-/// indistinguishable from a lone native app, and pulse always mints many.
-pub fn candidate(clients: &[ClientSnapshot]) -> Option<u32> {
-    let mut counts: BTreeMap<u32, usize> = BTreeMap::new();
-    for client in clients {
-        if let Some(pid) = client.sec_pid {
-            *counts.entry(pid).or_insert(0) += 1;
-        }
-    }
-
-    // Every PID seen on 2+ clients is a pulse candidate. If there is exactly
-    // one such PID we trust it; zero or several ⇒ fail closed.
-    let mut repeated = counts.iter().filter(|&(_, &n)| n >= 2).map(|(&pid, _)| pid);
-    let first = repeated.next()?;
-    if repeated.next().is_some() {
-        // Ambiguous: more than one value repeats.
-        return None;
-    }
-    Some(first)
 }
 
-/// Stage 2: confirm the candidate against the `comm` read from
+/// Stage 2: confirm one candidate against the `comm` read from
 /// `/proc/<candidate>/comm`.
 ///
 /// `comm` is `None` when the adapter's read failed — the `/proc` entry is
@@ -79,11 +87,36 @@ pub fn validate(candidate: u32, comm: Option<&str>) -> Option<u32> {
     }
 }
 
-/// The two stages composed, for callers that already hold the probed `comm`.
+/// Stage 2 across the whole candidate set: the *unique* PID whose `comm` is
-/// The model keeps them separate (it recomputes the candidate as clients
+/// pipewire-pulse's.
-/// churn, and only re-probes when the candidate *changes*), so this is a
+///
-/// convenience for tests and for the fully-resolved path.
+/// `None` when none matches (nothing to suppress that we can prove) and also
-pub fn derive(clients: &[ClientSnapshot], comm_of: impl Fn(u32) -> Option<String>) -> Option<u32> {
+/// when **several** do. Several means either two pipewire-pulse daemons are
-    let candidate = candidate(clients)?;
+/// live — a nested or sandboxed session — or a `comm` collision, and a single
-    validate(candidate, comm_of(candidate).as_deref())
+/// `Option<u32>` cannot suppress two owners. Failing closed here lands on the
+/// over-exclusion side, matching the asymmetry `owner::keys_of` already
+/// documents: broad over-exclusion is annoying, a missed suppression is an
+/// echo. Suppressing a *set* of daemon PIDs is the real answer if a
+/// multi-daemon host ever turns up; it is not v1, and it is recorded rather
+/// than silently approximated.
+pub fn resolve(candidates: &BTreeSet<u32>, comm_of: impl Fn(u32) -> Option<String>) -> Option<u32> {
+    let mut found = None;
+    for &pid in candidates {
+        if validate(pid, comm_of(pid).as_deref()).is_some() {
+            if found.is_some() {
+                return None;
+            }
+            found = Some(pid);
+        }
+    }
+    found
+}
+
+/// Both stages composed, for callers that can probe on demand.
+///
+/// The model keeps them separate — it recomputes the candidate set as Clients
+/// churn and only re-probes PIDs entering it — so this is a convenience for
+/// tests and for the fully-resolved path.
+pub fn derive(clients: &[ClientSnapshot], comm_of: impl Fn(u32) -> Option<String>) -> Option<u32> {
+    resolve(&candidates(clients), comm_of)
 }

src/host/observer/tests.rs

@@ -418,29 +418,107 @@ fn clients_with(pids: &[Option<u32>]) -> Vec<ClientSnapshot> {
         .collect()
 }
 
+/// A `comm` probe of `pipewire-pulse` identifies the daemon regardless of how
+/// many Clients it holds.
 #[test]
-fn pid_candidate_consistent_repeated_value() {
+fn pid_resolves_on_comm_not_on_repetition() {
     let clients = clients_with(&[Some(4137), Some(4137), Some(9001)]);
-    assert_eq!(pulse_pid::candidate(&clients), Some(4137));
+    let pulse = pulse_pid::derive(&clients, |pid| {
+        Some(
+            if pid == 4137 {
+                "pipewire-pulse"
+            } else {
+                "firefox"
+            }
+            .to_string(),
+        )
+    });
+    assert_eq!(pulse, Some(4137));
+}
+
+/// 🔴 **The round-10 regression, measured on this host and caught by the §5.1
+/// row-1 matrix run.** WirePlumber holds two Clients (`WirePlumber` and
+/// `WirePlumber [export]`) sharing one `sec_pid`, so two values repeat. The old
+/// stage 1 called that ambiguous and returned `None`, which switched key 4's
+/// suppression off and fused every Pulse-emulated node into a single owner —
+/// a machine-wide over-exclusion cascade, on a stock desktop, permanently.
+#[test]
+fn a_second_process_holding_two_clients_does_not_defeat_the_derivation() {
+    // 1747 = WirePlumber x2, 2528 = pipewire-pulse x2, plus a native app.
+    let clients = clients_with(&[Some(1747), Some(1747), Some(2528), Some(2528), Some(9001)]);
+    let pulse = pulse_pid::derive(&clients, |pid| {
+        Some(
+            match pid {
+                1747 => "wireplumber",
+                2528 => "pipewire-pulse",
+                _ => "firefox",
+            }
+            .to_string(),
+        )
+    });
+    assert_eq!(
+        pulse,
+        Some(2528),
+        "the WirePlumber pair must not make this ambiguous"
+    );
+}
+
+/// The other direction the old rule failed in: pipewire-pulse holding exactly
+/// one Client (a session with one Pulse app) repeated nothing, so it was never
+/// even a candidate — same cascade, opposite cause.
+#[test]
+fn a_daemon_holding_a_single_client_is_still_found() {
+    let clients = clients_with(&[Some(2528), Some(9001)]);
+    let pulse = pulse_pid::derive(&clients, |pid| {
+        Some(
+            if pid == 2528 {
+                "pipewire-pulse"
+            } else {
+                "kwin_wayland"
+            }
+            .to_string(),
+        )
+    });
+    assert_eq!(pulse, Some(2528));
 }
 
 #[test]
-fn pid_candidate_inconsistent_two_repeats_is_none() {
+fn pid_candidates_are_every_distinct_sec_pid() {
-    let clients = clients_with(&[Some(4137), Some(4137), Some(9001), Some(9001)]);
+    let clients = clients_with(&[Some(4137), Some(4137), Some(9001), None]);
-    assert_eq!(pulse_pid::candidate(&clients), None);
+    assert_eq!(
+        pulse_pid::candidates(&clients),
+        [4137, 9001].into_iter().collect()
+    );
 }
 
 #[test]
-fn pid_candidate_missing_property_is_none() {
+fn pid_missing_property_leaves_nothing_to_probe() {
     let clients = clients_with(&[None, None, None]);
-    assert_eq!(pulse_pid::candidate(&clients), None);
+    assert!(pulse_pid::candidates(&clients).is_empty());
+    assert_eq!(pulse_pid::derive(&clients, |_| None), None);
 }
 
+/// No Client's `comm` is pipewire-pulse's: nothing to suppress that we can
+/// prove, so `None` — and key 4 stays coarse rather than wrong.
 #[test]
-fn pid_candidate_single_occurrence_is_none() {
+fn pid_resolve_no_match_is_none() {
-    // One client per pid: nothing repeats, so nothing is pipewire-pulse.
     let clients = clients_with(&[Some(4137), Some(9001)]);
-    assert_eq!(pulse_pid::candidate(&clients), None);
+    assert_eq!(
+        pulse_pid::derive(&clients, |_| Some("firefox".to_string())),
+        None
+    );
+}
+
+/// Two live pipewire-pulse daemons: a single `Option<u32>` cannot suppress
+/// both, so fail closed to over-exclusion rather than pick one and leak the
+/// other's fusion.
+#[test]
+fn pid_resolve_two_daemons_is_none() {
+    let clients = clients_with(&[Some(4137), Some(9001)]);
+    assert_eq!(
+        pulse_pid::derive(&clients, |_| Some("pipewire-pulse".to_string())),
+        None
+    );
 }
 
 #[test]
@@ -473,9 +551,8 @@ fn pid_validate_reuse_named_other_process_is_none() {
 #[test]
 fn pid_derive_end_to_end_valid() {
     let clients = clients_with(&[Some(4137), Some(4137)]);
-    let candidate = pulse_pid::candidate(&clients).expect("candidate");
     assert_eq!(
-        pulse_pid::validate(candidate, Some("pipewire-pulse")),
+        pulse_pid::derive(&clients, |_| Some("pipewire-pulse".to_string())),
         Some(4137)
     );
 }
@@ -490,7 +567,7 @@ fn model_pulse_pid_valid_through_projection() {
     m.apply(client(1, 200, Some(4137)));
     m.apply(client(2, 201, Some(4137)));
     m.apply(client(3, 202, Some(9001)));
-    assert_eq!(m.pulse_pid_candidate(), Some(4137));
+    assert_eq!(m.pulse_pid_candidates(), [4137, 9001].into_iter().collect());
     m.apply(RegEvent::ProcCommProbed {
         pid: 4137,
         comm: Some("pipewire-pulse".to_string()),
@@ -507,6 +584,40 @@ fn model_pulse_pid_none_until_probed() {
     assert_eq!(m.project().pipewire_pulse_pid, None);
 }
 
+/// Probed `comm`s are dropped once no Client presents the PID any more.
+///
+/// Two reasons, and the second is the load-bearing one: the map is bounded by
+/// the live Client count in a process that runs for hours, **and** a PID that
+/// leaves and returns is re-probed rather than answered from the `comm` of
+/// whoever held that number before. Pruning cannot change the projection —
+/// `pulse_pid` only reads PIDs in the current candidate set — which is why it
+/// is not an `apply` arm and must not publish.
+#[test]
+fn a_departed_pid_does_not_keep_its_probed_comm() {
+    let mut m = model();
+    m.apply(client(1, 200, Some(4137)));
+    m.apply(RegEvent::ProcCommProbed {
+        pid: 4137,
+        comm: Some("pipewire-pulse".to_string()),
+    });
+    assert_eq!(m.project().pipewire_pulse_pid, Some(4137));
+
+    // The daemon's Client goes away; the adapter prunes to the live set.
+    let live = m.pulse_pid_candidates();
+    assert!(live.contains(&4137));
+    m.retain_probed_comms(&std::collections::BTreeSet::new());
+
+    // A *different* process now holds 4137 and opens a Client. Without the
+    // prune this would answer from the stale `comm` and suppress a real app's
+    // owner key.
+    m.apply(client(2, 201, Some(4137)));
+    assert_eq!(
+        m.project().pipewire_pulse_pid,
+        None,
+        "the stale comm must not survive its PID leaving the graph"
+    );
+}
+
 #[test]
 fn model_pulse_pid_none_on_comm_mismatch() {
     let mut m = model();

src/host/taint/mod.rs

@@ -904,7 +904,10 @@ fn propagate_owner_bridge(
 /// 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
+/// - A *bounded* tainted reader has a strong key, or a usable PID **backed by
+///   a resolved Client** (F11-1 — a node's self-claimed
+///   `application.process.id` no longer bounds anything on its own; see
+///   [`owner::owner_is_bounded`]), 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

src/host/taint/owner.rs

@@ -130,6 +130,22 @@ impl OwnerCtx {
     fn client_pid(&self, node: &NodeSnapshot) -> Option<u32> {
         self.client_pids.get(&node.props.client_id?).copied()
     }
+
+    /// Does this node have **protected provenance** — an unambiguous Client
+    /// yielding `Some(pipewire.sec.pid)`?
+    ///
+    /// ⚠️ Read **before** the pipewire-pulse suppression in [`keys_of`], and
+    /// that ordering is the whole rule (F11-1, below). A Pulse-emulated app's
+    /// Client resolves to the daemon's PID; the value is then omitted from the
+    /// bridge keys as too coarse to *group* on, but it is still a protected
+    /// `pipewire.*` answer to "who is this", so the app keeps its provenance.
+    ///
+    /// ❌ Not "a unique Client object exists". A unique Client with
+    /// `sec_pid = None` satisfies that and carries no protected identity at
+    /// all, which is exactly the hole [`owner_is_bounded`] closes.
+    fn client_is_resolved(&self, node: &NodeSnapshot) -> bool {
+        self.client_pid(node).is_some()
+    }
 }
 
 /// Which key bridged two legs. Ordered strongest first; the `Ord` derive is
@@ -257,10 +273,59 @@ fn keys_of(node: &NodeSnapshot, ctx: &OwnerCtx) -> Vec<(OwnerKey, KeyValue)> {
 /// the engine must fail closed rather than declare it clean
 /// (v3.4 §6.1.1, final paragraph).
 ///
-/// # 🔴 OPEN, phase-6 blocking — the key union can *reduce* taint here
+/// # F11-1 — CLOSED. A self-claimed PID is not provenance
 ///
-/// **Round 11 review, finding 1. Verified correct; deliberately not fixed in
+/// **The rule, implemented below:** a strong key (`node.link-group`,
-/// that round.** Round 10 made key 4 a union of the node's
+/// `pulse.module.id`) bounds an owner on its own; **key 4 bounds an owner only
+/// when the node's Client resolves** — see [`OwnerCtx::client_is_resolved`].
+/// A node whose Client cannot be resolved at all is *unbounded*, whatever
+/// `application.process.id` it puts on itself.
+///
+/// The five Client cases, which is the matrix this needed (two of them are
+/// indistinguishable under the wrong reading of "resolves"):
+///
+/// | Client | node's own pid | bounded? | why |
+/// | --- | --- | --- | --- |
+/// | **absent** | claimed | **no** | nothing corroborates the claim |
+/// | **ambiguous** (two Clients, one id) | claimed | **no** | "we do not know who owns this" must not be papered over |
+/// | **unique but pid-less** | claimed | **no** | a Client object is not an identity; `sec_pid` is |
+/// | **resolved-native** (`sec_pid` = the app's) | absent | **yes** | protected pid, and it *is* key 4 |
+/// | **resolved-to-pipewire-pulse** | claimed | **yes** | protected provenance; the daemon pid is suppressed as a *grouping* key only |
+///
+/// The last row is what keeps this from being the blunt fix. Applying
+/// "self-claims are not sound" without the provenance test unbounds every
+/// Pulse-emulated app — their Client's `sec_pid` is the daemon's and
+/// suppressed, so the node's own claim is their only per-app identity — which
+/// re-triggers the §6.1.1 mass over-exclusion the whole design exists to avoid
+/// and empties the eligible half of the §5.1 matrix.
+///
+/// **Cost, measured on the live graph** (2026-07-26): **zero**. The
+/// before- and after-binaries audited the *same* graph simultaneously — both
+/// are read-only observers, which is the only way to A/B a partition without
+/// churn between runs — with a tagged producer feeding the default sink,
+/// `parec` on its monitor as a real tainted reader (so the sweep was armed,
+/// not merely present in the code), and Firefox, `aplay` and `pacat` as
+/// bystanders. **181 records each, the same 14 distinct decision states, none
+/// exclusive to either side, no `unresolved-owner` on either.** The eligible
+/// half stayed non-empty throughout: native (`aplay`), Pulse-emulated
+/// (`pacat`) and Firefox all eligible. O5 is unmoved: identical p50 (15 µs)
+/// and busy fraction (0.0012), and the after-binary's worst per-record
+/// recompute was *lower* (217 µs vs 243 µs — noise, same debug build, same
+/// concurrent load).
+///
+/// Why it costs nothing here: every real app on this box is either native
+/// (Client `sec_pid` = its own pid) or Pulse-emulated (Client `sec_pid` = the
+/// daemon's), and **both resolve**. Sweeping all 18 live nodes for the
+/// predicate's inputs directly, the only unresolved-Client nodes were
+/// `Dummy-Driver` and `Freewheel-Driver`, which carry no pid key to lose;
+/// session-manager device nodes are unresolved too (their Client is pid-less)
+/// but exception 2 already strips key 4 from them. That is the answer the
+/// deferral was waiting for: the rule bites exactly the anomalous shapes, and
+/// this host has none.
+///
+/// ## The leak it closes (round 11 review, finding 1)
+///
+/// Round 10 made key 4 a union of the node's
 /// `application.process.id` and its Client's `pipewire.sec.pid`, and the claim
 /// that this was "strictly additive" was too strong: the same key list also
 /// feeds *this* predicate, so adding a value can move a node from unbounded to
@@ -277,55 +342,50 @@ fn keys_of(node: &NodeSnapshot, ctx: &OwnerCtx) -> Vec<(OwnerKey, KeyValue)> {
 /// 3. Neither the bridge nor the sweep fires, and the output stays eligible
 ///    while re-emitting the call.
 ///
-/// It cannot leak today: `evaluate()` is reached only by the dry-run audit,
+/// Step 2 is now unbounded ⇒ the sweep fires ⇒ the leg is excluded. Note it
-/// which creates no links. It becomes live when phase 6 consumes eligibility.
+/// could not leak *yet* when it was filed — `evaluate()` is reached only by the
+/// dry-run audit, which creates no links — and that is why the fix waited for
+/// the §5.1 measurement instead of guessing at its cost.
 ///
-/// **Why it is not fixed yet.** The principled repair is provenance: a
+/// ## What this is deliberately NOT
-/// self-claimed `application.process.id` is not a *sound* bound, only the
-/// protected keys are. But applying that bluntly makes every Pulse-emulated
-/// app unbounded — their Client's `sec_pid` is the daemon's and suppressed, so
-/// the node's own claim is their only per-app identity — which re-triggers the
-/// §6.1.1 mass over-exclusion the whole design is built to avoid, and would
-/// make the eligible half of the §5.1 matrix empty.
 ///
-/// The targeted rule that closes the path above without that cost: **a node
+/// It is not a claim that `application.process.id` is now unused: it still
-/// whose Client cannot be resolved at all must not be bounded by its own
+/// bridges (a self-claim is fine as *evidence that two legs are related* —
-/// self-claimed PID.** An ambiguous Client already means "we do not know who
+/// the fail-closed direction), and a resolved-Client node is still bounded by
-/// owns this", and a self-claim must not paper over it; a Pulse app's Client
+/// whichever key-4 value survives suppression. Only *boundedness* — the
-/// *is* resolved (to the daemon's PID, then suppressed), so it keeps its
+/// permission to say "I can enumerate this owner's other legs, so a
-/// bound. Implementing it needs `OwnerCtx` to distinguish "resolved" from
+/// differently-keyed output is provably someone else" — now demands a
-/// "absent", and `OwnerKeyIndex` to carry boundedness separately from the key
+/// `pipewire.*` answer to "who is this".
-/// set, since bridging must keep using the full union.
 ///
-/// ⚠️ Do this **with the §5.1 matrix data in hand**, not before: the whole
+/// ⚠️ Bridging must keep using the **full** union, so boundedness is carried
-/// question is how much over-exclusion the rule actually causes on a real
+/// separately from the key set in [`OwnerKeyIndex`] rather than being
-/// graph, and that is measurable rather than arguable.
+/// re-derived from it.
-///
-/// ## Round 12 — the deferral holds, and "resolved" has a trap in it
-///
-/// Codex re-examined this and agreed the deferral is defensible while
-/// `evaluate()` is audit-only, and that the rule above closes the recorded path
-/// without unbounding normal Pulse-emulated apps — **but only under one
-/// reading of "resolves"**, and the wrong reading reintroduces the hole:
-///
-/// - ✅ "Resolved" must mean **an unambiguous Client that yields
-///   `Some(pipewire.sec.pid)`**, taken *before* the pipewire-pulse suppression
-///   step. A Pulse app then still has the daemon's protected PID as
-///   provenance, even though that value is omitted from the bridge keys, so it
-///   stays bounded and the eligible half survives.
-/// - ❌ **Do not** implement it as "a unique Client object exists". A unique
-///   Client with `sec_pid = None` would satisfy that test while providing no
-///   protected identity at all, leaving exactly the self-claimed-PID hole this
-///   rule is meant to close.
-///
-/// So the matrix needs five Client cases, not two: **absent**, **ambiguous**,
-/// **unique but pid-less**, **resolved-native**, and
-/// **resolved-to-pipewire-pulse**. The third is the one that distinguishes the
-/// two readings, and it is the row a two-case matrix would silently skip.
 pub fn owner_is_bounded(node: &NodeSnapshot, ctx: &OwnerCtx) -> bool {
-    keys_of(node, ctx)
+    bounded_by(&keys_of(node, ctx), node, ctx)
-        .iter()
+}
-        .any(|(key, _)| *key != OwnerKey::ClientId)
+
+/// [`owner_is_bounded`]'s rule, over an already-computed key list.
+///
+/// The single implementation: [`OwnerKeyIndex::build`] has the keys in hand and
+/// must not recompute them, and two copies of a predicate this load-bearing is
+/// how the two spellings drift apart.
+fn bounded_by(keys: &[(OwnerKey, KeyValue)], node: &NodeSnapshot, ctx: &OwnerCtx) -> bool {
+    let mut has_process_key = false;
+    for (key, _) in keys {
+        match key {
+            // Strong keys are per-instance and name the sibling set directly.
+            OwnerKey::LinkGroup | OwnerKey::PulseModuleId => return true,
+            OwnerKey::ProcessId => has_process_key = true,
+            // Never: one process can present two `client.id`s (the measured
+            // GStreamer refutation, above).
+            OwnerKey::ClientId => {}
+        }
+    }
+    // F11-1. The key may be the node's own claim, the Client's protected pid,
+    // or both — `keys_of` does not record which, and it does not need to: a
+    // resolved Client is provenance for the node *whatever* value key 4 ends
+    // up carrying, and without one there is no protected identity to stand on.
+    has_process_key && ctx.client_is_resolved(node)
 }
 
 /// Owner keys computed once per snapshot.
@@ -336,16 +396,27 @@ pub fn owner_is_bounded(node: &NodeSnapshot, ctx: &OwnerCtx) -> bool {
 #[derive(Debug, Default)]
 pub struct OwnerKeyIndex {
     keys: BTreeMap<Serial, Vec<(OwnerKey, KeyValue)>>,
+    /// Nodes whose owner is positively bounded — see [`owner_is_bounded`].
+    ///
+    /// ⚠️ **Stored, not derived from `keys`.** Since F11-1 the predicate needs
+    /// the node's Client as well as its key list, and the two answers are
+    /// deliberately different: the full union still bridges, while a
+    /// self-claimed pid no longer bounds.
+    bounded: BTreeSet<Serial>,
 }
 
 impl OwnerKeyIndex {
     pub fn build(snapshot: &GraphSnapshot, ctx: &OwnerCtx) -> Self {
-        Self {
+        let mut keys = BTreeMap::new();
-            keys: snapshot
+        let mut bounded = BTreeSet::new();
-                .nodes()
+        for node in snapshot.nodes() {
-                .map(|node| (node.serial, keys_of(node, ctx)))
+            let node_keys = keys_of(node, ctx);
-                .collect(),
+            if bounded_by(&node_keys, node, ctx) {
+                bounded.insert(node.serial);
+            }
+            keys.insert(node.serial, node_keys);
         }
+        Self { keys, bounded }
     }
 
     /// The strongest key these two nodes share directly, if any.
@@ -409,11 +480,10 @@ impl OwnerKeyIndex {
         })
     }
 
-    /// See [`owner_is_bounded`].
+    /// See [`owner_is_bounded`]. A node outside this snapshot is unbounded,
+    /// which is the fail-closed answer.
     pub fn is_bounded(&self, serial: Serial) -> bool {
-        self.keys
+        self.bounded.contains(&serial)
-            .get(&serial)
-            .is_some_and(|keys| keys.iter().any(|(key, _)| *key != OwnerKey::ClientId))
     }
 }
 

src/host/taint/tests.rs

@@ -1088,6 +1088,226 @@ fn the_nodes_own_process_id_wins_over_its_clients() {
     );
 }
 
+// ──────────────────────────────────────────────────────────────────────
+// F11-1 — a self-claimed pid is not provenance: the five Client cases
+// ──────────────────────────────────────────────────────────────────────
+
+/// The scaffold every F11-1 row needs: peerspeak's call reaching the hardware
+/// sink, a **bounded** tainted reader, and an ordinary bystander.
+///
+/// ⚠️ The reader must be *bounded* (`sunshine` carries a real pid). An
+/// unbounded tainted reader trips `propagate_unresolved_owner`'s other tier,
+/// which sweeps **every** output candidate on the box regardless of its own
+/// keys — the three "unbounded" rows below would then pass without testing
+/// anything. The bystander is the other half of that guard: it is bounded via
+/// the ordinary Pulse shape, so an implementation that unbounded everything
+/// fails every row instead of passing three of them.
+///
+/// Returns the graph, the hardware sink to hang nodes off, and the two nodes
+/// every row must name in its partition.
+fn armed_with_a_bounded_reader() -> (Graph, NodeRef, NodeRef, NodeRef) {
+    let mut graph = Graph::new();
+    let hw = graph.device_node("hw-sink", MediaRole::Sink);
+    let call = graph.peerspeak_node("peerspeak", 7);
+    graph.link(call, hw);
+    let sunshine = graph.app_node("sunshine", MediaRole::StreamInput, 3_838);
+    graph.link(hw, sunshine);
+    let bystander = graph.app_node("mpv", MediaRole::StreamOutput, 9_001);
+    graph.link(bystander, hw);
+    (graph, hw, call, bystander)
+}
+
+/// Case 1 of 5 — **Client absent.** A node that names no Client at all has
+/// nothing but its own word for who owns it, so it cannot be bounded.
+#[test]
+fn an_absent_client_leaves_a_self_claimed_pid_unbounded() {
+    let (mut graph, hw, call, bystander) = armed_with_a_bounded_reader();
+    let orphan = graph.node(
+        "no-client",
+        MediaRole::StreamOutput,
+        NodeProps {
+            process_id: Some(70_001),
+            ..NodeProps::default()
+        },
+    );
+    graph.link(orphan, hw);
+
+    assert_partition(
+        &run(&graph, &ctx()),
+        &[("mpv", bystander)],
+        &[
+            ("call", call, "peerspeak-owned"),
+            ("no-client", orphan, "unresolved-owner"),
+        ],
+    );
+}
+
+/// Case 2 of 5 — **Client ambiguous.** Two live Clients claim the id, so the
+/// observer missed a removal and we do not know who owns this node. A
+/// self-claimed pid must not paper over that: this is step 2 of the recorded
+/// leak path, and before F11-1 the claim bounded the node and spared it.
+#[test]
+fn an_ambiguous_client_leaves_a_self_claimed_pid_unbounded() {
+    let (mut graph, hw, call, bystander) = armed_with_a_bounded_reader();
+    let shared_id = graph.client(Some(70_010));
+    graph.client_with_id(shared_id, Some(70_011));
+    let app = graph.node(
+        "ambiguous-client",
+        MediaRole::StreamOutput,
+        NodeProps {
+            client_id: Some(shared_id),
+            process_id: Some(70_012),
+            ..NodeProps::default()
+        },
+    );
+    graph.link(app, hw);
+
+    assert_partition(
+        &run(&graph, &ctx()),
+        &[("mpv", bystander)],
+        &[
+            ("call", call, "peerspeak-owned"),
+            ("ambiguous-client", app, "unresolved-owner"),
+        ],
+    );
+}
+
+/// Case 3 of 5 — **Client unique but pid-less**, and *the row that decides
+/// which rule is implemented*.
+///
+/// A unique Client object exists, so "resolved = a unique Client exists" would
+/// call this node bounded — leaving the self-claimed-pid hole wide open under a
+/// rule that looks like it closed it. `sec_pid` is what carries protected
+/// identity, so `None` means unresolved, and pid-less Clients are ordinary
+/// (the session manager's is one).
+///
+/// A two-case absent/resolved matrix skips this silently. That is why it is
+/// written out.
+#[test]
+fn a_unique_but_pidless_client_leaves_a_self_claimed_pid_unbounded() {
+    let (mut graph, hw, call, bystander) = armed_with_a_bounded_reader();
+    let pidless = graph.client(None);
+    let app = graph.node(
+        "pidless-client",
+        MediaRole::StreamOutput,
+        NodeProps {
+            client_id: Some(pidless),
+            process_id: Some(70_020),
+            ..NodeProps::default()
+        },
+    );
+    graph.link(app, hw);
+
+    assert_partition(
+        &run(&graph, &ctx()),
+        &[("mpv", bystander)],
+        &[
+            ("call", call, "peerspeak-owned"),
+            ("pidless-client", app, "unresolved-owner"),
+        ],
+    );
+}
+
+/// Case 4 of 5 — **Client resolved, native.** `pipewire.sec.pid` is the app's
+/// own, so provenance and key 4 are the same value and the node is bounded
+/// without claiming anything itself.
+#[test]
+fn a_resolved_native_client_bounds_its_node() {
+    let (mut graph, hw, call, bystander) = armed_with_a_bounded_reader();
+    let mpv2 = graph.native_client_node("mpv-native", MediaRole::StreamOutput, 70_030);
+    graph.link(mpv2, hw);
+
+    assert_partition(
+        &run(&graph, &ctx()),
+        &[("mpv", bystander), ("mpv-native", mpv2)],
+        &[("call", call, "peerspeak-owned")],
+    );
+}
+
+/// Case 5 of 5 — **Client resolved to pipewire-pulse.** The row that stops
+/// this rule from being the blunt fix.
+///
+/// Every Pulse-emulated app looks like this: the Client's `sec_pid` is the
+/// daemon's — suppressed as a *grouping* key, because it would fuse fifteen
+/// unrelated apps — while the node's own `application.process.id` is the app's.
+/// Provenance is read **before** that suppression, so the app keeps its bound
+/// and stays eligible. Reading it after would unbound every Pulse app on the
+/// box and empty the eligible half of the §5.1 matrix, which is the §6.1.1
+/// catastrophe arriving through the boundedness door.
+#[test]
+fn a_client_resolving_to_pipewire_pulse_still_bounds_its_node() {
+    let (mut graph, hw, call, bystander) = armed_with_a_bounded_reader();
+    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 70_040);
+    graph.link(firefox, hw);
+
+    assert_partition(
+        &run(&graph, &ctx()),
+        &[("mpv", bystander), ("firefox", firefox)],
+        &[("call", call, "peerspeak-owned")],
+    );
+}
+
+/// **The recorded leak path, end to end** (round 11 review, finding 1).
+///
+/// One process, two Clients. Its reading leg claims the daemon's pid — which
+/// exception 1 suppresses — while its Client holds a real protected pid `A`, so
+/// the union bounds the reader by `A` and the *unbounded-reader* tier never
+/// arms. Its re-emitting leg sits on a second Client whose id is **ambiguous**
+/// (one of the two claimants even holds `A`, so this is not "the guess would
+/// have been wrong" — it is "a guess is not evidence"), and claims a pid of its
+/// own. The two legs share no key, so the bridge does not fire either.
+///
+/// Before F11-1 the self-claim bounded the output leg, both tiers stayed quiet,
+/// and it re-emitted the call while eligible. Now the leg is unbounded, the
+/// bounded-reader tier sweeps it, and `mpv` shows the sweep is still targeted.
+#[test]
+fn a_self_claimed_pid_cannot_spare_an_output_leg_the_bridge_cannot_reach() {
+    let mut graph = Graph::new();
+    let hw = graph.device_node("hw-sink", MediaRole::Sink);
+    let call = graph.peerspeak_node("peerspeak", 7);
+    graph.link(call, hw);
+
+    let reader_client = graph.client(Some(80_000));
+    let reader = graph.node(
+        "forwarder-in",
+        MediaRole::StreamInput,
+        NodeProps {
+            client_id: Some(reader_client),
+            process_id: Some(PULSE_PID),
+            ..NodeProps::default()
+        },
+    );
+    graph.link(hw, reader);
+
+    let ambiguous = graph.client(Some(80_000));
+    graph.client_with_id(ambiguous, Some(80_001));
+    let output = graph.node(
+        "forwarder-out",
+        MediaRole::StreamOutput,
+        NodeProps {
+            client_id: Some(ambiguous),
+            process_id: Some(80_002),
+            ..NodeProps::default()
+        },
+    );
+    graph.link(output, hw);
+
+    let bystander = graph.app_node("mpv", MediaRole::StreamOutput, 9_001);
+    graph.link(bystander, hw);
+
+    // `mpv` staying eligible is what proves the reader is bounded: an
+    // unbounded tainted reader sweeps **every** output candidate, `mpv`
+    // included, and this row would then be testing the wrong tier.
+    assert_partition(
+        &run(&graph, &ctx()),
+        &[("mpv", bystander)],
+        &[
+            ("call", call, "peerspeak-owned"),
+            ("forwarder-out", output, "unresolved-owner"),
+        ],
+    );
+}
+
 #[test]
 fn an_unknown_pipewire_pulse_pid_over_excludes_rather_than_leaks() {
     // v3.4 §6.1.2's failure-mode paragraph: if pixelpass cannot identify
@@ -2463,5 +2683,12 @@ fn an_ambiguous_client_id_remembers_every_claimant_for_stickiness() {
         client_members >= 2,
         "both ambiguous-id clients should be remembered: {sticky:#?}"
     );
-    let _ = (ClientSnapshot { serial: Serial(0), id: GlobalId(0), sec_pid: None }, firefox);
+    let _ = (
+        ClientSnapshot {
+            serial: Serial(0),
+            id: GlobalId(0),
+            sec_pid: None,
+        },
+        firefox,
+    );
 }

src/main.rs

@@ -49,7 +49,7 @@ async fn main() -> Result<()> {
     pipewire::init();
 
     if cli.repair {
-        return repair::run().await;
+        return repair::run(cli.repair_legacy_untagged).await;
     }
 
     // Read-only diagnostic: observe the graph, report what the audio-exclusion

src/repair.rs

@@ -1,236 +0,0 @@
-//! `--repair`: clean up null-sinks and loopbacks left behind by a crashed
-//! pixelpass host. Identifies orphans by the `pixelpass_capture_<pid>`
-//! name pattern + dead-PID check, then unloads paired loopbacks first
-//! (mirrors `Routing::shutdown`'s order so PipeWire doesn't leave zombie
-//! links). Live PIDs — including this process and any other running
-//! pixelpass — are left alone.
-
-use anyhow::{Context, Result, bail};
-use std::collections::HashSet;
-use std::path::Path;
-use std::process::Command;
-
-const SINK_NAME_PREFIX: &str = "pixelpass_capture_";
-
-pub async fn run() -> Result<()> {
-    let modules = list_modules().context("failed to list pactl modules")?;
-
-    let mut dead_sinks: Vec<OrphanSink> = Vec::new();
-    let mut dead_pids: HashSet<u32> = HashSet::new();
-    let mut live_skipped: u32 = 0;
-
-    for m in &modules {
-        if m.name != "module-null-sink" {
-            continue;
-        }
-        let Some(sink_name) = extract_kv(&m.args, "sink_name") else {
-            continue;
-        };
-        let Some(pid_str) = sink_name.strip_prefix(SINK_NAME_PREFIX) else {
-            continue;
-        };
-        let Ok(pid) = pid_str.parse::<u32>() else {
-            continue;
-        };
-
-        if is_pid_alive(pid) {
-            live_skipped += 1;
-            continue;
-        }
-        dead_pids.insert(pid);
-        dead_sinks.push(OrphanSink {
-            id: m.id,
-            sink_name: sink_name.to_string(),
-            pid,
-        });
-    }
-
-    let mut dead_loopbacks: Vec<u32> = Vec::new();
-    for m in &modules {
-        if m.name != "module-loopback" {
-            continue;
-        }
-        // A pixelpass loopback references a capture sink either as its
-        // destination (`sink=pixelpass_capture_<pid>` — the default→null
-        // mirror) or as its source (`source=pixelpass_capture_<pid>.monitor`
-        // — the local monitor that lets the sharer hear the app). Match both.
-        let Some(pid) = loopback_capture_pid(&m.args) else {
-            continue;
-        };
-        if dead_pids.contains(&pid) {
-            dead_loopbacks.push(m.id);
-        }
-    }
-
-    if dead_sinks.is_empty() && dead_loopbacks.is_empty() {
-        if live_skipped > 0 {
-            println!(
-                "[pixelpass] --repair: nothing to clean up ({live_skipped} live pixelpass host(s) left alone)."
-            );
-        } else {
-            println!("[pixelpass] --repair: nothing to clean up.");
-        }
-        return Ok(());
-    }
-
-    let mut unloaded = 0u32;
-    let mut failed = 0u32;
-
-    for id in &dead_loopbacks {
-        match unload_module(*id) {
-            Ok(()) => {
-                println!("[pixelpass] --repair: unloaded loopback module #{id}");
-                unloaded += 1;
-            }
-            Err(e) => {
-                eprintln!("[pixelpass] --repair: failed to unload loopback #{id}: {e:#}");
-                failed += 1;
-            }
-        }
-    }
-    for orphan in &dead_sinks {
-        match unload_module(orphan.id) {
-            Ok(()) => {
-                println!(
-                    "[pixelpass] --repair: unloaded {} (orphaned from pid {})",
-                    orphan.sink_name, orphan.pid
-                );
-                unloaded += 1;
-            }
-            Err(e) => {
-                eprintln!(
-                    "[pixelpass] --repair: failed to unload {} (#{}): {e:#}",
-                    orphan.sink_name, orphan.id
-                );
-                failed += 1;
-            }
-        }
-    }
-
-    if live_skipped > 0 {
-        println!("[pixelpass] --repair: left {live_skipped} live pixelpass host(s) alone.");
-    }
-
-    if failed > 0 {
-        bail!("--repair: {failed} module(s) failed to unload (see errors above)");
-    }
-    println!("[pixelpass] --repair: cleaned up {unloaded} module(s).");
-    Ok(())
-}
-
-struct Module {
-    id: u32,
-    name: String,
-    args: String,
-}
-
-struct OrphanSink {
-    id: u32,
-    sink_name: String,
-    pid: u32,
-}
-
-fn list_modules() -> Result<Vec<Module>> {
-    let output = Command::new("pactl")
-        .args(["list", "short", "modules"])
-        .output()
-        .context("failed to run `pactl list short modules`")?;
-    if !output.status.success() {
-        bail!(
-            "pactl list short modules failed: {}",
-            String::from_utf8_lossy(&output.stderr).trim()
-        );
-    }
-    let text = String::from_utf8(output.stdout).context("pactl returned non-UTF-8")?;
-    let mut modules = Vec::new();
-    // `pactl list short modules` is tab-separated, but some modules have
-    // multi-line `{ ... }` argument blocks that wrap onto continuation
-    // lines starting with whitespace. The wrap lines never parse as a
-    // u32 ID, so the simple per-line + parse-id filter is robust.
-    for line in text.lines() {
-        let mut parts = line.splitn(4, '\t');
-        let Some(id_str) = parts.next() else { continue };
-        let Ok(id) = id_str.parse::<u32>() else {
-            continue;
-        };
-        let Some(name) = parts.next() else { continue };
-        let args = parts.next().unwrap_or("").to_string();
-        modules.push(Module {
-            id,
-            name: name.to_string(),
-            args,
-        });
-    }
-    Ok(modules)
-}
-
-/// The `pixelpass_capture_<pid>` PID a loopback references, whether the capture
-/// sink is its destination (`sink=pixelpass_capture_<pid>`) or its source
-/// (`source=pixelpass_capture_<pid>.monitor`). `None` for unrelated loopbacks.
-fn loopback_capture_pid(args: &str) -> Option<u32> {
-    let from_sink = extract_kv(args, "sink").and_then(|v| v.strip_prefix(SINK_NAME_PREFIX));
-    let from_source = extract_kv(args, "source")
-        .and_then(|v| v.strip_prefix(SINK_NAME_PREFIX))
-        .and_then(|rest| rest.strip_suffix(".monitor"));
-    from_sink
-        .or(from_source)
-        .and_then(|pid| pid.parse::<u32>().ok())
-}
-
-fn extract_kv<'a>(args: &'a str, key: &str) -> Option<&'a str> {
-    for token in args.split_whitespace() {
-        if let Some(rest) = token.strip_prefix(key)
-            && let Some(value) = rest.strip_prefix('=')
-        {
-            return Some(value);
-        }
-    }
-    None
-}
-
-fn is_pid_alive(pid: u32) -> bool {
-    Path::new(&format!("/proc/{pid}")).exists()
-}
-
-fn unload_module(id: u32) -> Result<()> {
-    let output = Command::new("pactl")
-        .arg("unload-module")
-        .arg(id.to_string())
-        .output()
-        .context("failed to run pactl unload-module")?;
-    if !output.status.success() {
-        bail!(
-            "pactl unload-module #{id}: {}",
-            String::from_utf8_lossy(&output.stderr).trim()
-        );
-    }
-    Ok(())
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn loopback_pid_matches_default_null_mirror_by_sink() {
-        // The default→null loopback: capture sink is the destination.
-        let args = "source=@DEFAULT_SINK@.monitor sink=pixelpass_capture_4242 latency_msec=20";
-        assert_eq!(loopback_capture_pid(args), Some(4242));
-    }
-
-    #[test]
-    fn loopback_pid_matches_local_monitor_by_source() {
-        // The local monitor: capture sink's monitor is the source, and the
-        // destination is the real default sink (not a pixelpass name).
-        let args = "source=pixelpass_capture_4242.monitor sink=@DEFAULT_SINK@ latency_msec=20";
-        assert_eq!(loopback_capture_pid(args), Some(4242));
-    }
-
-    #[test]
-    fn loopback_pid_ignores_unrelated_loopback() {
-        assert_eq!(
-            loopback_capture_pid("source=alsa_output.pci.monitor sink=some_other_sink"),
-            None
-        );
-    }
-}

src/repair/introspect.rs

@@ -0,0 +1,297 @@
+//! Structured Pulse introspection: the observation and destruction layer for
+//! `--repair`.
+//!
+//! # Why this replaced parsing `pactl`
+//!
+//! Three separate defects, all of them consequences of reading a human-oriented
+//! text format rather than the protocol:
+//!
+//! 1. **Record boundaries were unprovable.** `pactl list short modules` prints a
+//!    module's argument raw into a tab-and-newline-delimited format with no
+//!    escaping. A *genuine* module whose argument contains a newline — say
+//!    `…latency_msec=20\nremix=false`, and `remix` is a real loopback option —
+//!    renders a first line that reads byte-exactly like one of our canonical
+//!    forms, with the remainder dropped as an unparseable continuation. No index
+//!    is forged, so no duplicate-index check can see it: repair would classify and
+//!    unload a module it had never actually seen in full. A tab in the same
+//!    position instead hides a sink reference, which is worse, because the gate
+//!    that protects a still-referenced sink then cannot see the reference.
+//! 2. **Index and argument could be mis-paired.** The one listing that carries the
+//!    exact argument (`-f json`) carries **no index** at all on pactl 17, and the
+//!    one that carries the index cannot carry the argument faithfully. Combining
+//!    them by position is unsound whenever module names repeat: another client
+//!    loading one module and unloading another between the two calls leaves the
+//!    counts and names aligned while every argument has shifted by one.
+//! 3. **Locality was a guess.** `PULSE_SERVER` is a *fallback list*, so
+//!    `unix:/missing tcp:remote:4713` passes any "starts with unix:" test and then
+//!    connects to another machine — where our local pids mean nothing and a live
+//!    remote host's modules look dead.
+//!
+//! `pa_module_info` carries index, name and argument together in one structured
+//! record, so (1) and (2) cannot arise. `pa_context_is_local()` answers (3) about
+//! the connection that actually got established rather than about a string we
+//! hoped described it. And because unloading goes back through the *same*
+//! connection, there is no window in which listing and destruction could disagree
+//! about which server they are talking to.
+//!
+//! # What is deliberately not here
+//!
+//! No decisions. This module observes and destroys; every judgement about what may
+//! be destroyed lives in [`super::plan`], which is pure and needs no Pulse server
+//! to test. The one policy this layer owns is *refusing to talk to the wrong
+//! server at all*.
+
+use anyhow::{Context as _, Result, bail};
+use libpulse_binding::callbacks::ListResult;
+use libpulse_binding::context::{Context, FlagSet as ContextFlagSet, State as ContextState};
+use libpulse_binding::mainloop::standard::{IterateResult, Mainloop};
+use libpulse_binding::operation::{Operation, State as OperationState};
+use libpulse_binding::proplist::{Proplist, properties};
+use std::cell::RefCell;
+use std::rc::Rc;
+use std::time::{Duration, Instant};
+
+use super::plan::ModuleObservation;
+
+/// How long to wait for the connection to reach `Ready`. A one-shot CLI must not
+/// hang on an unresponsive server; failing closed here costs the user a re-run.
+const CONNECT_BUDGET: Duration = Duration::from_secs(3);
+
+/// How long any single introspection request may take.
+///
+/// ⚠️ On timeout the `Operation` wrapper is dropped while still running. In
+/// libpulse-binding 2.30.1 that only unrefs the C operation — the boxed callback
+/// and the `Rc`s it captured leak until the context cancels the operation at
+/// disconnect. That is bounded and harmless *here*, because `--repair` is a
+/// one-shot process that exits immediately afterwards, and it cannot become a
+/// use-after-free (the closure owns its clones, and the context clears callbacks
+/// before the mainloop is touched). **It would not be acceptable in the long-lived
+/// host**, so this module must not be reused for host-side loading until that
+/// binding bug is fixed or worked around; `op.cancel()` does not help.
+const REQUEST_BUDGET: Duration = Duration::from_secs(3);
+
+/// How long to sleep between mainloop iterations while waiting. Non-blocking
+/// iteration plus a short sleep keeps the deadline enforceable, which
+/// `iterate(true)` would not.
+const POLL_INTERVAL: Duration = Duration::from_millis(2);
+
+/// A live, verified-local connection to the Pulse server.
+///
+/// Both listing and unloading run through this one connection, so everything
+/// repair sees and everything it destroys provably belong to the same server.
+///
+/// ⚠️ **Field order is load-bearing, and this was not theoretical.** Rust drops
+/// fields in declaration order, and the context's teardown frees IO events that
+/// live *in* the mainloop. With `mainloop` declared first, `--repair` did its work
+/// correctly and then died on the way out:
+///
+/// ```text
+/// Assertion '!e->dead' failed at ../pulseaudio/src/pulse/mainloop.c:207,
+/// function mainloop_io_free(). Aborting.
+/// ```
+///
+/// SIGABRT, a core dump, and exit 134 — so a completely successful repair reported
+/// failure to its caller. This is the same invariant phase 0b's
+/// `ScreenshareTeardown` exists for, met again one layer down.
+///
+/// Rather than leave that resting on where the fields happen to be written, [`Drop`]
+/// **explicitly** takes and drops the context first, so the ordering survives a
+/// future reorder of this struct. The declaration order below is still correct, and
+/// now it is also not load-bearing.
+pub struct PulseSession {
+    /// `Option` only so that `Drop` can `take()` it and destroy it *before* the
+    /// mainloop. Always `Some` for the whole of the session's usable life.
+    context: Option<Context>,
+    mainloop: Mainloop,
+}
+
+impl Drop for PulseSession {
+    fn drop(&mut self) {
+        // Disconnect, then destroy the context while the mainloop it registered its
+        // IO events with is still alive. The mainloop then drops after us.
+        //
+        // Nothing is drained afterwards on purpose. An earlier version iterated the
+        // mainloop a few times here to "let teardown settle", which was a ritual
+        // rather than a barrier: a fixed number of non-blocking polls cannot
+        // guarantee that any particular event became ready. It is also unnecessary —
+        // PulseAudio's context unlink cancels outstanding operations and removes the
+        // context's socket machinery synchronously, so by the time `drop(context)`
+        // returns there is no obligation left for the mainloop to service.
+        if let Some(mut context) = self.context.take() {
+            context.disconnect();
+            drop(context);
+        }
+    }
+}
+
+impl PulseSession {
+    /// The live context. Infallible in practice: only `Drop` ever clears it, and
+    /// nothing can call this afterwards.
+    fn context(&mut self) -> &mut Context {
+        self.context
+            .as_mut()
+            .expect("the context is only taken during Drop")
+    }
+    /// Connect, wait for readiness, and refuse anything but a local server.
+    pub fn connect() -> Result<Self> {
+        let mut proplist = Proplist::new().context("could not allocate a Pulse proplist")?;
+        // `set_str` fails only on an invalid key, and these keys are constants.
+        let _ = proplist.set_str(properties::APPLICATION_NAME, "pixelpass --repair");
+        let _ = proplist.set_str(properties::APPLICATION_ID, "xyz.pixelpass.repair");
+
+        let mut mainloop = Mainloop::new().context("could not create a Pulse mainloop")?;
+        let mut context = Context::new_with_proplist(&mainloop, "pixelpass --repair", &proplist)
+            .context("could not create a Pulse context")?;
+        context
+            .connect(None, ContextFlagSet::NOFLAGS, None)
+            .context("could not connect to the Pulse server")?;
+
+        let deadline = Instant::now() + CONNECT_BUDGET;
+        loop {
+            iterate_once(&mut mainloop)?;
+            match context.get_state() {
+                ContextState::Ready => break,
+                ContextState::Failed => {
+                    bail!("the Pulse server refused the connection");
+                }
+                ContextState::Terminated => {
+                    bail!("the Pulse connection terminated before it was ready");
+                }
+                _ => {
+                    if Instant::now() >= deadline {
+                        bail!(
+                            "the Pulse server did not become ready within {:?}",
+                            CONNECT_BUDGET
+                        );
+                    }
+                    std::thread::sleep(POLL_INTERVAL);
+                }
+            }
+        }
+
+        // The connection is established, so this is about the server we actually
+        // reached — not about what a server string appeared to promise. A remote
+        // server's module table belongs to another machine's processes, where our
+        // pids mean nothing, so repair must not touch it.
+        match context.is_local() {
+            Some(true) => {}
+            Some(false) => bail!(
+                "connected to a REMOTE Pulse server; --repair only ever operates on the local \
+                 server, because it decides what to unload from local process liveness"
+            ),
+            None => bail!(
+                "could not determine whether the Pulse server is local; refusing to unload \
+                 anything"
+            ),
+        }
+
+        Ok(Self {
+            context: Some(context),
+            mainloop,
+        })
+    }
+
+    /// Every loaded module, with its exact argument.
+    pub fn list_modules(&mut self) -> Result<Vec<ModuleObservation>> {
+        // `Rc<RefCell<…>>` because the callback is owned by the C library and may
+        // be invoked many times before the operation completes.
+        let collected: Rc<RefCell<Vec<ModuleObservation>>> = Rc::new(RefCell::new(Vec::new()));
+        let failed: Rc<RefCell<bool>> = Rc::new(RefCell::new(false));
+
+        let sink = Rc::clone(&collected);
+        let error_flag = Rc::clone(&failed);
+        let op = self
+            .context()
+            .introspect()
+            .get_module_info_list(move |result| match result {
+                ListResult::Item(info) => {
+                    // A module with no name is not one we can identify, and an
+                    // argumentless module is simply one loaded without arguments.
+                    let name = info.name.as_deref().unwrap_or_default();
+                    let args = info.argument.as_deref().unwrap_or_default();
+                    sink.borrow_mut()
+                        .push(ModuleObservation::new(info.index, name, args));
+                }
+                ListResult::End => {}
+                ListResult::Error => *error_flag.borrow_mut() = true,
+            });
+
+        self.run_to_completion(op, "list modules")?;
+        if *failed.borrow() {
+            bail!("the Pulse server returned an error while listing modules");
+        }
+
+        let modules = collected.borrow().clone();
+        // Impossible per the protocol — an index identifies one module — so this is
+        // a sanity check on external input, not a safety boundary. It fails closed
+        // because an ambiguous index is one we could unload wrongly.
+        for (i, module) in modules.iter().enumerate() {
+            if modules[..i].iter().any(|earlier| earlier.id == module.id) {
+                bail!(
+                    "the Pulse server reported module index #{} twice; refusing to unload \
+                     anything",
+                    module.id
+                );
+            }
+        }
+        Ok(modules)
+    }
+
+    /// Unload one module, over the same connection it was observed on.
+    pub fn unload_module(&mut self, index: u32) -> Result<()> {
+        let succeeded: Rc<RefCell<Option<bool>>> = Rc::new(RefCell::new(None));
+        let outcome = Rc::clone(&succeeded);
+        let op = self
+            .context()
+            .introspect()
+            .unload_module(index, move |success| *outcome.borrow_mut() = Some(success));
+
+        self.run_to_completion(op, "unload module")?;
+        match *succeeded.borrow() {
+            Some(true) => Ok(()),
+            Some(false) => bail!("the Pulse server rejected unloading module #{index}"),
+            // The operation completed without the callback running, which we cannot
+            // read as success.
+            None => bail!("no result was reported for unloading module #{index}"),
+        }
+    }
+
+    /// Drive the mainloop until `op` finishes, or the budget expires.
+    fn run_to_completion<T: ?Sized>(&mut self, op: Operation<T>, what: &str) -> Result<()> {
+        let deadline = Instant::now() + REQUEST_BUDGET;
+        loop {
+            iterate_once(&mut self.mainloop)?;
+            match op.get_state() {
+                OperationState::Done => return Ok(()),
+                OperationState::Cancelled => {
+                    bail!("the Pulse server cancelled the request to {what}");
+                }
+                OperationState::Running => {
+                    // A connection that dies mid-request would otherwise be waited
+                    // out to the full budget.
+                    match self.context().get_state() {
+                        ContextState::Ready => {}
+                        state => {
+                            bail!("the Pulse connection became {state:?} while trying to {what}")
+                        }
+                    }
+                    if Instant::now() >= deadline {
+                        bail!("the Pulse server did not {what} within {REQUEST_BUDGET:?}");
+                    }
+                    std::thread::sleep(POLL_INTERVAL);
+                }
+            }
+        }
+    }
+}
+
+/// One non-blocking mainloop iteration, with quit and error surfaced as errors.
+fn iterate_once(mainloop: &mut Mainloop) -> Result<()> {
+    match mainloop.iterate(false) {
+        IterateResult::Success(_) => Ok(()),
+        IterateResult::Quit(code) => {
+            bail!("the Pulse mainloop quit unexpectedly (code {})", code.0)
+        }
+        IterateResult::Err(e) => Err(e).context("the Pulse mainloop failed"),
+    }
+}

src/repair/mod.rs

@@ -0,0 +1,553 @@
+//! `--repair`: clean up the Pulse modules left behind by a crashed pixelpass
+//! host.
+//!
+//! All of the judgement lives in [`plan`], which is pure. What remains here is
+//! I/O plus the two rules that cannot be expressed in a plan:
+//!
+//! - **Re-verify immediately before destroying anything.** Pulse module indices
+//!   are reused verbatim, and a host can die (or come back) between the snapshot
+//!   and the unload, so the plan is treated as evidence that expires — never as a
+//!   licence.
+//! - **Gate the sink on what is still attached to it**, not on the plan's ordering
+//!   having succeeded. An unload can fail or be skipped, and a loopback can appear
+//!   after the plan was made.
+//!
+//! Repair never touches native PipeWire nodes. Since phase 0c the capture sink is
+//! connection-owned and removes itself when its host dies, so there is nothing
+//! there for repair to do and no safe way for it to help.
+//!
+//! # Where the observations come from
+//!
+//! Structured Pulse introspection over one verified-local connection — see
+//! [`introspect`], which also documents the three defects that parsing `pactl`'s
+//! text output turned out to have. Listing *and* unloading both go through that
+//! same connection.
+
+pub mod introspect;
+pub mod plan;
+
+use anyhow::{Context, Result, bail};
+use std::path::Path;
+
+use introspect::PulseSession;
+use plan::{Fingerprint, Liveness, Shape};
+
+pub async fn run(clean_untagged: bool) -> Result<()> {
+    let liveness = LivenessProbe::new();
+    if let Some(reason) = liveness.degraded_reason() {
+        // Scoped deliberately: modules carrying a token that matches this machine,
+        // boot and pid namespace are still cleaned, because the token establishes
+        // what these signals can only guess at. Saying "refusing to unload
+        // anything" here would be false in exactly the container-recovery case the
+        // token was added for.
+        eprintln!(
+            "[pixelpass] --repair: cannot independently determine process liveness \
+             ({reason}); modules WITHOUT an ownership token will be left alone."
+        );
+    }
+
+    let local = local_identity().context("could not establish this process's own identity")?;
+    let policy = plan::Policy {
+        local,
+        untagged: if clean_untagged {
+            plan::UntaggedPolicy::CleanByPidAlone
+        } else {
+            plan::UntaggedPolicy::Refuse
+        },
+    };
+    if clean_untagged {
+        eprintln!(
+            "[pixelpass] --repair: --repair-legacy-untagged given; untagged modules will be \
+             judged by process id ALONE. That is only safe on the machine and in the pid \
+             namespace that ran the crashed host."
+        );
+    }
+
+    let mut pulse = PulseSession::connect().context("could not observe the Pulse module table")?;
+    let modules = pulse
+        .list_modules()
+        .context("could not list Pulse modules")?;
+
+    // Say so loudly when something names our sinks but matches no shape we know:
+    // that is either a third party using our names, or a newer pixelpass whose
+    // modules this build cannot recognise. The second is how repair would go
+    // silently blind, so it never gets inferred from a clean exit.
+    let unrecognised = plan::unrecognised_pixelpass_modules(&modules);
+    if !unrecognised.is_empty() {
+        eprintln!(
+            "[pixelpass] --repair: {} module(s) name a pixelpass capture sink but do not match \
+             any shape this build knows; they are being LEFT ALONE:",
+            unrecognised.len()
+        );
+        for obs in &unrecognised {
+            eprintln!(
+                "[pixelpass] --repair:   #{} {} {}",
+                obs.id, obs.name, obs.args
+            );
+        }
+    }
+
+    let planned = plan::plan(&modules, &policy, |pid, attribution| {
+        liveness_for(&liveness, attribution, pid)
+    });
+
+    // Ours by shape, but carrying no proof of whose pid they name. Never unloaded by
+    // default — listed, so an explicit legacy run has something to look at first.
+    if !planned.untagged.is_empty() {
+        eprintln!(
+            "[pixelpass] --repair: {} module(s) are pixelpass's but carry no ownership token, so \
+             the process id in their name cannot be attributed to this machine or pid namespace. \
+             LEFT ALONE. Re-run with --repair-legacy-untagged to clean them by pid alone:",
+            planned.untagged.len()
+        );
+        for fp in &planned.untagged {
+            eprintln!(
+                "[pixelpass] --repair:   #{} {} (claims pid {})",
+                fp.id,
+                fp.shape.label(),
+                fp.pid
+            );
+        }
+    }
+    // Tokened, but the token belongs to another machine, boot or namespace.
+    if !planned.foreign.is_empty() {
+        eprintln!(
+            "[pixelpass] --repair: {} module(s) belong to another machine, boot or pid namespace; \
+             their process ids mean nothing here. LEFT ALONE:",
+            planned.foreign.len()
+        );
+        for fp in &planned.foreign {
+            eprintln!(
+                "[pixelpass] --repair:   #{} {} (claims pid {})",
+                fp.id,
+                fp.shape.label(),
+                fp.pid
+            );
+        }
+    }
+
+    if planned.is_empty() {
+        let mut held = Vec::new();
+        if !planned.live_pids.is_empty() {
+            held.push(format!(
+                "{} live pixelpass host(s)",
+                planned.live_pids.len()
+            ));
+        }
+        if !planned.unknown_pids.is_empty() {
+            held.push(format!(
+                "{} pid(s) of undeterminable liveness",
+                planned.unknown_pids.len()
+            ));
+        }
+        if held.is_empty() {
+            println!("[pixelpass] --repair: nothing to clean up.");
+        } else {
+            println!(
+                "[pixelpass] --repair: nothing to clean up ({} left alone).",
+                held.join(", ")
+            );
+        }
+        return Ok(());
+    }
+
+    let mut unloaded = 0u32;
+    let mut skipped = 0u32;
+    let mut failed = 0u32;
+
+    for fp in &planned.unload {
+        // ORDER MATTERS, and it is the opposite of what reads naturally.
+        //
+        // Liveness is asked FIRST, and the fresh snapshot is taken AFTER it. The
+        // tempting order — verify the module, then check liveness, then unload —
+        // leaves the dangerous window wide open: between `kill` returning ESRCH and
+        // the unload, this process can be descheduled long enough for the planned
+        // module to vanish, a new host to inherit both the pid and the module index,
+        // and its differently-nonced arguments to occupy that index. Nothing would
+        // re-read those arguments, so the reused index gets unloaded.
+        //
+        // Asking liveness first and re-verifying the fingerprint after it means a
+        // replacement arriving in that window is caught by the argument comparison,
+        // and only the irreducible snapshot-to-unload interval remains.
+        match attributed_liveness(&liveness, fp) {
+            Liveness::Dead => {}
+            Liveness::Alive => {
+                println!(
+                    "[pixelpass] --repair: pid {} is alive again; leaving module #{} alone",
+                    fp.pid, fp.id
+                );
+                skipped += 1;
+                continue;
+            }
+            Liveness::Unknown => {
+                eprintln!(
+                    "[pixelpass] --repair: pid {}'s liveness became undeterminable; \
+                     leaving module #{} alone",
+                    fp.pid, fp.id
+                );
+                skipped += 1;
+                continue;
+            }
+        }
+
+        // Fresh snapshot per action, taken after the liveness answer. Deliberately
+        // not hoisted out of the loop: each unload changes the module table, and the
+        // point is to decide against the table as it is *now*.
+        let current = pulse
+            .list_modules()
+            .context("could not re-list Pulse modules")?;
+        let Some(obs) = current.iter().find(|m| m.id == fp.id) else {
+            println!(
+                "[pixelpass] --repair: module #{} is already gone; skipping",
+                fp.id
+            );
+            skipped += 1;
+            continue;
+        };
+        if !fp.still_matches(obs) {
+            // The index now names something else, or the same module's
+            // arguments changed. Either way we no longer know what we would be
+            // destroying, so we do not destroy it.
+            eprintln!(
+                "[pixelpass] --repair: module #{} no longer matches what was planned \
+                 (index reused?); refusing to unload it",
+                fp.id
+            );
+            skipped += 1;
+            continue;
+        }
+        // The sink goes last in the plan, but "last" is not the same as "nothing
+        // is attached any more": a loopback unload may have failed or been
+        // skipped, or a new one may have arrived since. Ask the fresh snapshot.
+        if fp.shape == Shape::LegacyCaptureSink
+            && let Some(holder) = plan::sink_still_referenced(&current, fp.pid, fp.id)
+        {
+            eprintln!(
+                "[pixelpass] --repair: module #{} (capture sink for pid {}) is still referenced \
+                 by module #{}; leaving the sink loaded",
+                fp.id, fp.pid, holder
+            );
+            skipped += 1;
+            continue;
+        }
+
+        match pulse.unload_module(fp.id) {
+            Ok(()) => {
+                println!("[pixelpass] --repair: {}", describe(fp));
+                unloaded += 1;
+            }
+            Err(e) => {
+                eprintln!("[pixelpass] --repair: failed to unload #{}: {e:#}", fp.id);
+                failed += 1;
+            }
+        }
+    }
+
+    if !planned.live_pids.is_empty() {
+        println!(
+            "[pixelpass] --repair: left {} live pixelpass host(s) alone.",
+            planned.live_pids.len()
+        );
+    }
+    if !planned.unknown_pids.is_empty() {
+        println!(
+            "[pixelpass] --repair: left {} pid(s) alone whose liveness could not be determined.",
+            planned.unknown_pids.len()
+        );
+    }
+    if skipped > 0 {
+        // Deliberately not "changed under us": a skip can also mean the module is
+        // still referenced, or its owner's liveness stopped being decidable. The
+        // per-module reason was printed above.
+        println!("[pixelpass] --repair: skipped {skipped} module(s) (reasons above).");
+    }
+    if failed > 0 {
+        bail!("--repair: {failed} module(s) failed to unload (see errors above)");
+    }
+    println!("[pixelpass] --repair: cleaned up {unloaded} module(s).");
+    Ok(())
+}
+
+fn describe(fp: &Fingerprint) -> String {
+    format!(
+        "unloaded {} #{} (orphaned from pid {})",
+        fp.shape.label(),
+        fp.id,
+        fp.pid
+    )
+}
+
+/// Ask liveness with the module's *attribution* in hand.
+///
+/// A module whose token matches this machine, boot and pid namespace has already
+/// proven that its pid is a number meaningful here — that is the token's entire
+/// job. Running such a module through the probe's degradation checks would defeat
+/// it in exactly the situation it exists for: a host crashing inside a container
+/// leaves a token that matches perfectly, while a container marker or a multi-entry
+/// `NSpid` makes the probe answer `Unknown` for everything, so token-qualified
+/// repair would do nothing precisely where it is now safe.
+///
+/// The degradation signals therefore guard only the *untagged* path, where a bare
+/// pid is all there is and those signals are the only protection left.
+fn liveness_for(probe: &LivenessProbe, attribution: plan::Attribution, pid: u32) -> Liveness {
+    match attribution {
+        plan::Attribution::Tokened => probe.of_attributed(pid),
+        plan::Attribution::Untagged => probe.of(pid),
+    }
+}
+
+/// The same rule, for a fingerprint at execution time.
+fn attributed_liveness(probe: &LivenessProbe, fp: &Fingerprint) -> Liveness {
+    let attribution = match fp.owner {
+        Some(_) => plan::Attribution::Tokened,
+        None => plan::Attribution::Untagged,
+    };
+    liveness_for(probe, attribution, fp.pid)
+}
+
+// ──────────────────────────────────────────────────────────────────────
+// Identity
+// ──────────────────────────────────────────────────────────────────────
+
+/// This process's machine, boot and pid-namespace identity.
+///
+/// Read from the kernel and the system, never guessed: without all three, a token
+/// cannot be compared and no module can be attributed. Dashes are stripped so every
+/// component is safe inside a single unquoted Pulse property value.
+pub fn local_identity() -> Result<plan::LocalIdentity> {
+    let machine = read_identity_file("/etc/machine-id")
+        .or_else(|_| read_identity_file("/var/lib/dbus/machine-id"))
+        .context("could not read a machine id")?;
+    let boot = read_identity_file("/proc/sys/kernel/random/boot_id")
+        .context("could not read the boot id")?;
+    let pid_ns = pid_namespace_id().context("could not read this process's pid namespace")?;
+    Ok(plan::LocalIdentity {
+        machine,
+        boot,
+        pid_ns,
+    })
+}
+
+fn read_identity_file(path: &str) -> Result<String> {
+    let raw = std::fs::read_to_string(path).with_context(|| format!("could not read {path}"))?;
+    let cleaned: String = raw
+        .trim()
+        .chars()
+        .filter(|c| c.is_ascii_hexdigit())
+        .collect();
+    if cleaned.is_empty() {
+        bail!("{path} held no usable identity");
+    }
+    Ok(cleaned)
+}
+
+/// The inode of `/proc/self/ns/pid` — the kernel's identity for a pid namespace.
+///
+/// This is the value that makes a pid meaningful: two processes in different pid
+/// namespaces can hold the same number, and only this distinguishes them.
+fn pid_namespace_id() -> Result<u64> {
+    use std::os::unix::fs::MetadataExt;
+    let meta =
+        std::fs::metadata("/proc/self/ns/pid").context("could not stat /proc/self/ns/pid")?;
+    Ok(meta.ino())
+}
+
+// ──────────────────────────────────────────────────────────────────────
+// Liveness
+// ──────────────────────────────────────────────────────────────────────
+
+/// Answers "is this pid still around", and knows when it must refuse to answer.
+///
+/// **An error is not an absence.** `Path::exists()` folds permission errors and a
+/// missing `/proc` into `false`, which here would read as "dead, go ahead and
+/// unload". Liveness is asked with `kill(pid, 0)` instead, where `EPERM` *proves*
+/// existence.
+///
+/// # The limit of what this can prove, stated rather than papered over
+///
+/// A pid can be alive and invisible. Inside a pid namespace — a container, a
+/// distrobox — `/proc/self` is perfectly visible while every process in the
+/// *parent* namespace is not, and `hidepid` has the same self-visible,
+/// others-invisible shape. Repair in such a place can reach the host's Pulse
+/// socket, see a live host's modules, get `ESRCH` for its pid and unload a running
+/// host's audio.
+///
+/// The signals below are **negative** ones: they detect *some* cases where pid
+/// numbers cannot be trusted, and every one of them fails closed. What they cannot
+/// do is prove the converse. `NSpid` reports this process's pid in each namespace
+/// that its procfs can see, and its leftmost value is relative to the pid namespace
+/// that mounted that procfs — so a nested namespace with its own `/proc` reports a
+/// single entry quite legitimately. `NSpid > 1` therefore means "definitely
+/// nested", while `NSpid == 1` means only "not detectably nested".
+///
+/// Closing that properly needs the module itself to carry an owner token (machine
+/// and boot identity plus pid-namespace identity) written at load time, with
+/// token-less modules treated as `Unknown`. That changes what pixelpass writes into
+/// the graph and how far back `--repair` can clean up, so it is a design decision
+/// recorded in the impl plan rather than guessed at here.
+struct LivenessProbe {
+    /// `None` when no signal says pid numbers are untrustworthy; `Some(reason)`
+    /// when every answer must be [`Liveness::Unknown`].
+    degraded: Option<String>,
+}
+
+impl LivenessProbe {
+    fn new() -> Self {
+        Self {
+            degraded: Self::detect_degradation(),
+        }
+    }
+
+    fn detect_degradation() -> Option<String> {
+        // Locality is deliberately NOT checked here. `PULSE_SERVER` is a fallback
+        // *list*, so `unix:/missing tcp:remote:4713` starts with "unix:" and still
+        // connects to another machine, and a remote server can be selected by client
+        // configuration with the variable unset entirely. The authoritative answer
+        // comes from `pa_context_is_local()` on the connection that actually got
+        // established — see `introspect::PulseSession::connect`.
+        match std::fs::read_to_string("/proc/self/status") {
+            Ok(status) => {
+                let nspid = status
+                    .lines()
+                    .find_map(|line| line.strip_prefix("NSpid:"))
+                    .map(|rest| rest.split_whitespace().count());
+                match nspid {
+                    Some(n) if n > 1 => {
+                        return Some(format!(
+                            "this process is in a nested pid namespace (NSpid has {n} entries), \
+                             so pids in module names may belong to processes it cannot see"
+                        ));
+                    }
+                    // NB: a single entry is not proof of the initial namespace — see
+                    // the type's doc comment. It only means nothing detected it.
+                    // A kernel too old to report NSpid cannot rule nesting out.
+                    None => {
+                        return Some(
+                            "/proc/self/status does not report NSpid, so pid-namespace identity \
+                             cannot be established"
+                                .to_string(),
+                        );
+                    }
+                    Some(_) => {}
+                }
+            }
+            Err(e) => return Some(format!("/proc/self/status could not be read: {e}")),
+        }
+        // Belt and braces: container runtimes that leave a marker.
+        for marker in ["/run/.containerenv", "/.dockerenv"] {
+            if Path::new(marker).try_exists().unwrap_or(false) {
+                return Some(format!("{marker} exists, so this is a container"));
+            }
+        }
+        None
+    }
+
+    fn degraded_reason(&self) -> Option<&str> {
+        self.degraded.as_deref()
+    }
+
+    /// Liveness for a pid this process has **no** independent reason to trust —
+    /// an untagged module. Here the degradation signals are the only protection.
+    fn of(&self, pid: u32) -> Liveness {
+        if self.degraded.is_some() {
+            return Liveness::Unknown;
+        }
+        self.of_attributed(pid)
+    }
+
+    /// Liveness for a pid already proven to belong to this machine, boot and pid
+    /// namespace by an [`plan::OwnerToken`].
+    ///
+    /// The degradation checks are deliberately skipped: they exist to guess at
+    /// whether a bare pid is meaningful, and here that is not a guess any more.
+    fn of_attributed(&self, pid: u32) -> Liveness {
+        // `kill(0, …)` signals our whole process group and a negative pid signals
+        // another group, so neither may ever reach `kill`. Neither is a pid we
+        // could have written into a sink name anyway.
+        if pid == 0 || pid > i32::MAX as u32 {
+            return Liveness::Unknown;
+        }
+        match nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid as i32), None) {
+            Ok(()) => Liveness::Alive,
+            // The process exists; we merely may not signal it.
+            Err(nix::errno::Errno::EPERM) => Liveness::Alive,
+            Err(nix::errno::Errno::ESRCH) => Liveness::Dead,
+            Err(_) => Liveness::Unknown,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // The `pactl`-text parser that used to live here is gone, and so are its
+    // tests: `introspect` gets index, name and argument as structured fields, so
+    // there is no format left to mis-parse. What replaced those tests is the live
+    // field gate, since the remaining risk is in talking to the server, which no
+    // unit test can exercise. The decisions all live in `plan`, which is pure and
+    // tested there.
+
+    /// The probe must never say `Dead` when it cannot see the whole pid space, and
+    /// must never ask `kill` about a pid that would signal something other than one
+    /// process.
+    #[test]
+    fn a_degraded_probe_never_reports_dead() {
+        let degraded = LivenessProbe {
+            degraded: Some("test".to_string()),
+        };
+        assert_eq!(degraded.of(1), Liveness::Unknown);
+        assert_eq!(degraded.of(u32::MAX), Liveness::Unknown);
+
+        let probe = LivenessProbe::new();
+        // Our own pid is alive by construction — unless this test itself runs
+        // somewhere the probe must abstain, which is exactly the other branch.
+        let me = std::process::id();
+        match probe.degraded_reason() {
+            None => assert_eq!(probe.of(me), Liveness::Alive),
+            Some(_) => assert_eq!(probe.of(me), Liveness::Unknown),
+        }
+        // `kill(0, …)` would signal our whole process group, and a pid past
+        // `i32::MAX` cannot be expressed to `kill` at all.
+        assert_eq!(probe.of(0), Liveness::Unknown);
+        assert_eq!(probe.of(u32::MAX), Liveness::Unknown);
+    }
+
+    /// A token proves the pid is meaningful here, so the probe's namespace
+    /// guesswork must not veto it. Without this, a host crashing inside a
+    /// container leaves a perfectly matching token while a container marker makes
+    /// every answer `Unknown` — and token-qualified repair does nothing in exactly
+    /// the situation the token was built for.
+    ///
+    /// This runs against a *degraded* probe deliberately: on an ordinary desktop
+    /// the two paths agree, so a test using the real probe's state would pass
+    /// whether or not the distinction exists.
+    #[test]
+    fn a_token_beats_the_degradation_signals_but_a_bare_pid_does_not() {
+        let degraded = LivenessProbe {
+            degraded: Some("pretending to be in a container".to_string()),
+        };
+        let me = std::process::id();
+
+        assert_eq!(
+            degraded.of_attributed(me),
+            Liveness::Alive,
+            "an attributed pid must still be answered when the probe is degraded"
+        );
+        assert_eq!(
+            degraded.of(me),
+            Liveness::Unknown,
+            "a bare pid must not be, since the signals are all it has"
+        );
+
+        // And the routing between them, which is what the caller actually uses.
+        assert_eq!(
+            liveness_for(&degraded, plan::Attribution::Tokened, me),
+            Liveness::Alive
+        );
+        assert_eq!(
+            liveness_for(&degraded, plan::Attribution::Untagged, me),
+            Liveness::Unknown
+        );
+    }
+}