host/taint: close five leaks found in Codex round 1

All five were reachable, all five now have a regression test, and each
test was verified by injecting the mutation that reverts its fix.

1. Sticky taint ignored surviving Client members. An app can close every
   stream while keeping its PipeWire connection open and then open a new
   one — Firefox does this constantly — and the new leg came back
   Eligible while the owner's buffers still held the call. Sticky seeding
   now resolves live Client serials to their current nodes.
2. "Receives audio" was inferred from `media.class` alone, so a node with
   an absent or unexpected class sitting on a real inbound link could not
   start an owner bridge and its sibling re-emitted the call. A node is
   now a receiver if it appears as a resolved `link.input.node` OR has a
   receiving role.
3. The device-node coarse-key exception was keyed on `media.class` being
   `Audio/Sink|Source|Duplex`, which also stripped the only correlation a
   *native virtual sink* has (own client, no link-group, no module id).
   Now keyed on `device.id`, measured on the live graph as the exact
   discriminator: the 5 ALSA nodes carry device.id 43/45/46 and share
   `client.id` 42 (`WirePlumber [export]`); the 3 `support.null-audio-sink`
   nodes carry no device.id and hold their own clients.
4. The unbounded-owner backstop required the tainted *reader* to be
   unbounded. Properties can be asymmetric — a reader with a link-group
   whose re-emitting leg has none is bounded while its sibling is not
   findable — so that condition is dropped; targets stay restricted to
   unbounded output legs, which keeps the blast radius small.
5. A not-ready snapshot could retire sticky owners, erasing taint history
   on the strength of a graph already declared untrustworthy. `evaluate`
   now returns the prior state unchanged while `!graph_ready`.

Test-quality findings, also fixed:
- a single pass of each rule survived all 32 tests (every fixture needed
  at most one owner hop) → two-chained-forwarder test with a clean
  control, plus a 60-layer chain to catch an accidental blow-up
- first-write-wins `raise()` survived → a node reached by bridge on one
  pass and by a direct link on the next must report the stronger reason
- `drop_clients` left the fixture's client caches stale, so "a fresh
  client after teardown" was really a dangling id; the recycling row now
  reuses node id, client id AND `pulse.module.id` verbatim

Also corrected the cost claim: this is O((V+E)·D) for owner-bridge depth
D, not O(V+E) as v3.4 §6.4 states. Owner keys are now computed once per
snapshot instead of per candidate pair.

42 tests.

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

src/host/taint/fixture.rs

@@ -90,18 +90,25 @@ impl Graph {
         self.node(name, role, app(client, pid))
     }
 
-    /// A leg of a pactl-loaded module: one client per module, and the
-    /// node's `application.process.id` is **pipewire-pulse's own**, because
-    /// pipewire-pulse genuinely is the client.
-    pub fn module_node(&mut self, name: &str, role: MediaRole, module: u64) -> NodeRef {
-        let client = match self.client_by_module.get(&module) {
+    /// The client a pactl module holds. Measured: each module gets its own
+    /// (`sink-sunshine-*` were clients 83/86/92), which is why one tainted
+    /// module does not fuse with the next.
+    pub fn client_of_module(&mut self, module: u64) -> GlobalId {
+        match self.client_by_module.get(&module) {
             Some(id) => *id,
             None => {
                 let id = self.client(Some(PULSE_PID));
                 self.client_by_module.insert(module, id);
                 id
             }
-        };
+        }
+    }
+
+    /// A leg of a pactl-loaded module: one client per module, and the
+    /// node's `application.process.id` is **pipewire-pulse's own**, because
+    /// pipewire-pulse genuinely is the client.
+    pub fn module_node(&mut self, name: &str, role: MediaRole, module: u64) -> NodeRef {
+        let client = self.client_of_module(module);
         self.node(name, role, pulse_module(client, module, PULSE_PID))
     }
 
@@ -112,8 +119,9 @@ impl Graph {
         self.node(name, role, link_group(group, client, pid))
     }
 
-    /// A device node as the session manager creates it: no strong key, and
-    /// WirePlumber's client and PID — shared with every other device.
+    /// A device node as the session manager creates it: no strong key,
+    /// WirePlumber's client and PID — shared with every other device — and
+    /// a `device.id`, which is what marks it as session-manager-exported.
     pub fn device_node(&mut self, name: &str, role: MediaRole) -> NodeRef {
         let session = match self.session_client {
             Some(id) => id,
@@ -123,7 +131,17 @@ impl Graph {
                 id
             }
         };
-        self.node(name, role, device(session, SESSION_PID))
+        let device_id = self.id();
+        self.node(name, role, device(session, SESSION_PID, device_id))
+    }
+
+    /// A **virtual** sink an application created natively: an `Audio/Sink`
+    /// with no `device.id` and no strong key, sharing one client with the
+    /// stream that re-emits what it receives. Coarse keys must still bridge
+    /// these two, or the whole call leaks through the re-emitting leg.
+    pub fn native_virtual_node(&mut self, name: &str, role: MediaRole, pid: u32) -> NodeRef {
+        let client = self.client_of_app(pid);
+        self.node(name, role, app(client, pid))
     }
 
     pub fn peerspeak_node(&mut self, name: &str, pid: u32) -> NodeRef {
@@ -228,8 +246,31 @@ impl Graph {
     }
 
     /// Drop clients too — full owner teardown.
+    ///
+    /// Invalidates the per-app/per-module caches as well: leaving them
+    /// stale made a later `client_of_app` hand back the *removed* client's
+    /// id, so a test that meant "a brand-new client after teardown" was
+    /// really building a node pointing at a client object that no longer
+    /// existed (Codex round 1, finding 8).
     pub fn drop_clients(&mut self, ids: &[GlobalId]) {
         self.clients.retain(|c| !ids.contains(&c.id));
+        self.client_by_app.retain(|_, id| !ids.contains(id));
+        self.client_by_module.retain(|_, id| !ids.contains(id));
+        if self.session_client.is_some_and(|id| ids.contains(&id)) {
+            self.session_client = None;
+        }
+    }
+
+    /// A client that reuses a global id a dead client had — the recycling
+    /// case, with a fresh serial.
+    pub fn client_with_id(&mut self, id: GlobalId, sec_pid: Option<u32>) -> GlobalId {
+        let serial = self.serial();
+        self.clients.push(ClientSnapshot {
+            serial,
+            id,
+            sec_pid,
+        });
+        id
     }
 }
 
@@ -266,10 +307,17 @@ pub fn link_group(group: &str, client: GlobalId, pid: u32) -> NodeProps {
 
 /// A device node as the session manager creates it: no strong key, and the
 /// session manager's own client and PID — shared with every other device.
-pub fn device(session_client: GlobalId, session_pid: u32) -> NodeProps {
+///
+/// Measured 2026-07-21: real ALSA device nodes carry the shared
+/// `client.id` but **no** `application.process.id` at all. Giving them one
+/// here is deliberately *more* pessimistic than reality — it hands the
+/// engine a second coarse key it could fuse devices on, so a test that
+/// passes here also passes against the real props.
+pub fn device(session_client: GlobalId, session_pid: u32, device_id: GlobalId) -> NodeProps {
     NodeProps {
         client_id: Some(session_client),
         process_id: Some(session_pid),
+        device_id: Some(device_id),
         ..NodeProps::default()
     }
 }

src/host/taint/mod.rs

@@ -305,14 +305,25 @@ impl Decisions {
 
 /// Recompute eligibility for the whole graph.
 ///
-/// Full recompute per graph event is the v1 design (O(V+E) over a ~40-node
-/// desktop graph); there is deliberately no incremental dirty-set.
+/// Full recompute per graph event is the v1 design; there is deliberately
+/// no incremental dirty-set.
+///
+/// ⚠️ **Cost is not O(V+E), despite what v3.4 §6.4 says.** Each fixpoint
+/// pass re-runs a full link BFS *and* a full owner scan, and one pass is
+/// consumed per layer of owner-bridge hops, so it is O((V+E)·D) for a
+/// bridge depth D. D is 1 for every topology observed so far and 2 for a
+/// forwarder feeding a forwarder; a 60-layer chain is covered by a test
+/// purely to catch an accidental blow-up. Phase 5 records the real
+/// recompute-duration distribution and maximum, which is what the "full
+/// recompute is fine for v1" claim should actually rest on — a measured
+/// headroom, not a node count.
 pub fn evaluate(
     snapshot: &GraphSnapshot,
     ctx: &ExclusionCtx,
     prior: &StickyState,
 ) -> (Decisions, StickyState) {
     let components = OwnerComponents::build(snapshot, ctx.pipewire_pulse_pid);
+    let keys = owner::OwnerKeyIndex::build(snapshot, ctx.pipewire_pulse_pid);
 
     let mut taint: BTreeMap<Serial, Reason> = BTreeMap::new();
     let mut sticky_serials: BTreeSet<Serial> = BTreeSet::new();
@@ -331,19 +342,29 @@ pub fn evaluate(
     // owner bridge feed each other — a bridged output leg has downstream
     // links, and a downstream monitor reader bridges to its own siblings —
     // so neither can be run once.
-    let downstream = downstream_edges(snapshot, &mut taint);
+    let edges = downstream_edges(snapshot, &mut taint);
     loop {
         let mut changed = false;
-        changed |= propagate_links(&downstream, &mut taint);
-        changed |= propagate_owner_bridge(snapshot, ctx, &components, &mut taint);
-        changed |= propagate_unresolved_owner(snapshot, ctx, &mut taint);
+        changed |= propagate_links(&edges.edges, &mut taint);
+        changed |= propagate_owner_bridge(&keys, &components, &edges, &mut taint);
+        changed |= propagate_unresolved_owner(snapshot, &keys, &edges, &mut taint);
         if !changed {
             break;
         }
     }
 
     let decisions = build_decisions(snapshot, ctx, &taint, &sticky_serials);
-    let next_sticky = build_sticky(snapshot, &components, &taint, prior);
+    let next_sticky = if ctx.graph_ready {
+        build_sticky(snapshot, &components, &taint, prior)
+    } else {
+        // ⚠️ Never retire a sticky owner on the strength of a snapshot we
+        // have already declared untrustworthy (Codex round 1). An object
+        // missing from a partial graph has not been observed to disappear,
+        // and `build_sticky` drops entries whose members are all absent —
+        // so one not-ready recompute could erase the taint history and the
+        // next ready one would hand back a clean bill of health.
+        prior.clone()
+    };
     (decisions, next_sticky)
 }
 
@@ -402,14 +423,25 @@ fn seed_sticky(
     sticky_serials: &mut BTreeSet<Serial>,
 ) {
     for entry in &prior.owners {
-        let live_nodes: Vec<Serial> = entry
-            .members
-            .iter()
-            .filter_map(|member| match member {
-                ObjectRef::Node(serial) => snapshot.node(*serial).map(|_| *serial),
-                ObjectRef::Client(_) => None,
-            })
-            .collect();
+        let mut live_nodes: Vec<Serial> = Vec::new();
+        for member in &entry.members {
+            match member {
+                ObjectRef::Node(serial) => {
+                    if snapshot.node(*serial).is_some() {
+                        live_nodes.push(*serial);
+                    }
+                }
+                // ⚠️ A surviving **Client** re-seeds too. An app can close
+                // every stream it had while keeping its PipeWire connection
+                // open, then open a fresh one — Firefox does exactly this.
+                // Seeding only from live nodes let that new leg come back
+                // Eligible while the owner was still, by v3.4 §6.1.3's own
+                // rule, tainted (its buffers outlive its streams).
+                ObjectRef::Client(serial) => {
+                    live_nodes.extend(nodes_of_client(snapshot, *serial));
+                }
+            }
+        }
         // The owner is sticky, not the individual node: a leg that appears
         // later in the same still-live owner inherits the taint.
         for serial in live_nodes {
@@ -423,24 +455,45 @@ fn seed_sticky(
     }
 }
 
+/// Nodes currently attached to a client, by the client's **serial**. The
+/// client's snapshot-local id is resolved fresh each time, so a recycled id
+/// can never resurrect a dead owner.
+fn nodes_of_client(snapshot: &GraphSnapshot, client: Serial) -> Vec<Serial> {
+    let Some(id) = snapshot
+        .clients()
+        .find(|c| c.serial == client)
+        .map(|c| c.id)
+    else {
+        return Vec::new();
+    };
+    // If that id is ambiguous, two clients claim it and we cannot say which
+    // nodes belong to the sticky one — so claim them all, which is the
+    // fail-closed direction.
+    snapshot
+        .nodes()
+        .filter(|node| node.props.client_id == Some(id))
+        .map(|node| node.serial)
+        .collect()
+}
+
 /// `output node → input nodes`, resolving snapshot-local ids. An endpoint
 /// that does not resolve taints the *other* end as unresolved ancestry when
 /// that other end is the input side — we cannot know what is feeding it.
-fn downstream_edges(
-    snapshot: &GraphSnapshot,
-    taint: &mut BTreeMap<Serial, Reason>,
-) -> BTreeMap<Serial, Vec<Serial>> {
+fn downstream_edges(snapshot: &GraphSnapshot, taint: &mut BTreeMap<Serial, Reason>) -> Edges {
     let mut edges: BTreeMap<Serial, Vec<Serial>> = BTreeMap::new();
+    let mut receivers: BTreeSet<Serial> = BTreeSet::new();
     for link in snapshot.links() {
         let from = snapshot.node_by_id(link.output_node);
         let to = snapshot.node_by_id(link.input_node);
         match (from, to) {
             (Some(IdLookup::Unique(from)), Some(IdLookup::Unique(to))) => {
                 edges.entry(from).or_default().push(to);
+                receivers.insert(to);
             }
             (_, Some(IdLookup::Unique(to))) => {
                 // Something feeds this node and we cannot say what.
                 raise(taint, to, Reason::UnresolvedAncestry);
+                receivers.insert(to);
             }
             _ => {}
         }
@@ -449,7 +502,29 @@ fn downstream_edges(
         targets.sort_unstable();
         targets.dedup();
     }
-    edges
+    // A node that receives audio by *role* counts even with no inbound link
+    // yet: a pixelpass capture sink is a taint root the moment it exists,
+    // and its owner's re-emitting leg must be bridged from it immediately.
+    receivers.extend(
+        snapshot
+            .nodes()
+            .filter(|node| node.role.receives_audio())
+            .map(|node| node.serial),
+    );
+    Edges { edges, receivers }
+}
+
+/// Resolved signal edges plus the set of nodes that can receive audio.
+struct Edges {
+    edges: BTreeMap<Serial, Vec<Serial>>,
+    /// ⚠️ Membership is "appears as a resolved `link.input.node`" **or**
+    /// "has a receiving role" — deliberately not role alone. Codex round 1:
+    /// a node whose `media.class` is absent or unexpected (`Other`), or an
+    /// `Audio/Source` that is really a filter output, can sit on an inbound
+    /// link carrying tainted audio; inferring "receives audio" from the role
+    /// alone left such a node unable to start an owner bridge, and its
+    /// sibling output leg stayed Eligible while re-emitting the call.
+    receivers: BTreeSet<Serial>,
 }
 
 fn propagate_links(
@@ -482,21 +557,19 @@ fn propagate_links(
 /// leg ⇒ exclude the output" rule would exclude every app using a
 /// microphone, Firefox in a video call included.
 fn propagate_owner_bridge(
-    snapshot: &GraphSnapshot,
-    ctx: &ExclusionCtx,
+    keys: &owner::OwnerKeyIndex,
     components: &OwnerComponents,
+    edges: &Edges,
     taint: &mut BTreeMap<Serial, Reason>,
 ) -> bool {
     let mut changed = false;
     for members in components.components() {
-        let sources: Vec<Serial> = members
+        let sources: BTreeSet<Serial> = members
             .iter()
             .copied()
             .filter(|serial| {
                 taint.get(serial).is_some_and(|r| r.propagates())
-                    && snapshot
-                        .node(*serial)
-                        .is_some_and(|n| n.role.receives_audio())
+                    && edges.receivers.contains(serial)
             })
             .collect();
         if sources.is_empty() {
@@ -506,17 +579,11 @@ fn propagate_owner_bridge(
             if sources.contains(target) {
                 continue;
             }
-            let Some(target_node) = snapshot.node(*target) else {
-                continue;
-            };
             // Name the strongest key shared directly with any tainted
             // member; `None` means the two are only transitively related.
             let key = sources
                 .iter()
-                .filter_map(|source| snapshot.node(*source))
-                .filter_map(|source| {
-                    owner::strongest_shared_key(source, target_node, ctx.pipewire_pulse_pid)
-                })
+                .filter_map(|source| keys.strongest_shared(*source, *target))
                 .min();
             changed |= raise(taint, *target, Reason::TaintedOwnerBridge { key });
         }
@@ -525,40 +592,44 @@ fn propagate_owner_bridge(
 }
 
 /// Fail-closed backstop for an owner we cannot bound (v3.4 §6.1.1, final
-/// paragraph): a capture stream is reading tainted audio and nothing about
-/// it lets us enumerate its sibling output legs, so we cannot know which
-/// output leg is re-emitting what it read. Exclude the output legs that are
-/// equally unbounded.
+/// paragraph): something read tainted audio and nothing about the output
+/// legs on this box lets us enumerate which of them are its siblings, so we
+/// cannot know which one is re-emitting what it read. Exclude the output
+/// legs that are equally unbounded.
 ///
-/// Two deliberate restrictions keep the blast radius at "small", as the
-/// design promises:
+/// Two restrictions keep the blast radius at "small", as the design
+/// promises — and one earlier restriction was **removed** as unsafe:
 ///
-/// - **Source must be a `Stream/Input/Audio`.** A tainted *device* sink is
-///   the normal case, not an anomaly — peerspeak's own playback taints the
-///   default sink on literally every recompute — and letting that trip this
-///   rule would exclude the entire desktop.
+/// - **Source must not be a real device node.** A tainted hardware sink is
+///   the normal case, not an anomaly: peerspeak's own playback taints the
+///   default sink on literally every recompute, and letting that trip this
+///   rule would exclude the whole desktop.
 /// - **Targets must themselves be unbounded.** Ordinary application streams
 ///   carry a real `application.process.id` and are bounded, so they are
-///   never swept up; in practice only other daemon-owned keyless module
-///   streams are.
+///   never swept up; in practice only daemon-owned keyless module streams
+///   are.
+/// - ⚠️ **Removed (Codex round 1): the source no longer has to be unbounded
+///   itself.** Properties can be asymmetric — a reader carrying a
+///   `node.link-group` whose re-emitting leg carries none is *bounded* while
+///   its sibling is not findable — and requiring an unbounded source let
+///   exactly that shape through as Eligible.
 fn propagate_unresolved_owner(
     snapshot: &GraphSnapshot,
-    ctx: &ExclusionCtx,
+    keys: &owner::OwnerKeyIndex,
+    edges: &Edges,
     taint: &mut BTreeMap<Serial, Reason>,
 ) -> bool {
-    let unbounded_reader = snapshot.nodes().any(|node| {
-        node.role == MediaRole::StreamInput
+    let tainted_reader = snapshot.nodes().any(|node| {
+        node.props.device_id.is_none()
+            && edges.receivers.contains(&node.serial)
             && taint.get(&node.serial).is_some_and(|r| r.propagates())
-            && !owner::owner_is_bounded(node, ctx.pipewire_pulse_pid)
     });
-    if !unbounded_reader {
+    if !tainted_reader {
         return false;
     }
     let mut changed = false;
     for node in snapshot.nodes() {
-        if node.role == MediaRole::StreamOutput
-            && !owner::owner_is_bounded(node, ctx.pipewire_pulse_pid)
-        {
+        if node.role == MediaRole::StreamOutput && !keys.is_bounded(node.serial) {
             changed |= raise(taint, node.serial, Reason::UnresolvedOwner);
         }
     }

src/host/taint/owner.rs

@@ -35,18 +35,34 @@
 //!    tainted module input would exclude every module-created stream on the
 //!    box. Keys 1 and 2 already cover those cases precisely.
 //!
-//! 2. **Coarse keys (3 and 4) may not bridge device-role nodes.** ⚠️ This
-//!    rule is *not* in design v3.4 — it was found while implementing, and
-//!    it is the exact analogue of exception 1 for the session manager:
-//!    every ALSA `Audio/Sink` and `Audio/Source` on the box is created by
-//!    WirePlumber and therefore shares one `client.id` and one
-//!    `application.process.id`. Without this rule, the hardware sink
-//!    carrying peerspeak's playback (tainted by design, every single time)
-//!    would bridge to *every other device node including the microphone
-//!    source*, whose readers would then taint their owners' playback legs —
-//!    reproducing precisely the §6.1.1 catastrophe ("excludes any app using
-//!    a microphone") through a different door. Device nodes that genuinely
-//!    belong to a module still bridge, via keys 1 and 2.
+//! 2. **Coarse keys (3 and 4) may not bridge nodes exported from a real
+//!    `Device`** — i.e. nodes carrying `device.id`. ⚠️ This rule is *not*
+//!    in design v3.4; it was found while implementing, and it is the exact
+//!    analogue of exception 1 for the session manager.
+//!    ✅ **MEASURED on the live graph 2026-07-21:**
+//!
+//!    | node | `client.id` | `device.id` | `factory.name` |
+//!    | --- | --- | --- | --- |
+//!    | 5 × `alsa_{output,input}.*` | **42** (`WirePlumber [export]`) | 43/45/46 | `api.alsa.pcm.{sink,source}` |
+//!    | 3 × `sink-sunshine-*` | 83 / 86 / 92 (each its own) | **absent** | `support.null-audio-sink` |
+//!
+//!    So one shared coarse key genuinely does relate every hardware device
+//!    on the box, and `device.id` cleanly separates that set from virtual
+//!    sinks. Without the rule, the hardware sink carrying peerspeak's
+//!    playback (tainted by design, every single recompute) would bridge to
+//!    *every other device node including the microphone source*, whose
+//!    readers would then taint their owners' playback legs — reproducing
+//!    precisely the §6.1.1 catastrophe ("excludes any app using a
+//!    microphone") through a different door.
+//!
+//!    ⚠️ **Keyed on `device.id`, NOT on `media.class` being `Audio/Sink`.**
+//!    The first cut suppressed coarse keys for every device-*role* node,
+//!    and Codex refuted it: a **native virtual sink** — an app that creates
+//!    an `Audio/Sink` plus a re-emitting stream on one client, with no
+//!    `link-group` and no `pulse.module.id` — would then have had its only
+//!    correlation stripped, and it would have leaked the whole call. Such a
+//!    sink has no `device.id`, so it now bridges on `client.id` as it
+//!    should.
 //!
 //! Grouping is **transitive** (union-find). That is the fail-closed
 //! direction: bigger owner components mean more taint, never less.
@@ -98,8 +114,9 @@ fn keys_of(node: &NodeSnapshot, pipewire_pulse_pid: Option<u32>) -> Vec<(OwnerKe
     if let Some(module) = node.props.pulse_module_id {
         out.push((OwnerKey::PulseModuleId, KeyValue::Num(module)));
     }
-    // Exception 2: coarse keys never bridge device-role nodes.
-    if node.role.is_device_role() {
+    // Exception 2: coarse keys never bridge nodes exported from a real
+    // Device — they all share the session manager's client.
+    if node.props.device_id.is_some() {
         return out;
     }
     if let Some(client) = node.props.client_id {
@@ -140,6 +157,48 @@ pub fn owner_is_bounded(node: &NodeSnapshot, pipewire_pulse_pid: Option<u32>) ->
         .any(|(key, _)| *key != OwnerKey::ClientId)
 }
 
+/// Owner keys computed once per snapshot.
+///
+/// `keys_of` allocates a `Vec` and clones the `link-group` string, and the
+/// bridge asks for keys once per (tainted member × component member) pair —
+/// so recomputing was the hot spot in an otherwise linear pass.
+#[derive(Debug, Default)]
+pub struct OwnerKeyIndex {
+    keys: BTreeMap<Serial, Vec<(OwnerKey, KeyValue)>>,
+}
+
+impl OwnerKeyIndex {
+    pub fn build(snapshot: &GraphSnapshot, pipewire_pulse_pid: Option<u32>) -> Self {
+        Self {
+            keys: snapshot
+                .nodes()
+                .map(|node| (node.serial, keys_of(node, pipewire_pulse_pid)))
+                .collect(),
+        }
+    }
+
+    /// The strongest key these two nodes share directly, if any.
+    pub fn strongest_shared(&self, a: Serial, b: Serial) -> Option<OwnerKey> {
+        let (Some(a_keys), Some(b_keys)) = (self.keys.get(&a), self.keys.get(&b)) else {
+            return None;
+        };
+        // Stored strongest-first, so the first match is the strongest.
+        a_keys.iter().find_map(|(key, value)| {
+            b_keys
+                .iter()
+                .any(|(other_key, other_value)| other_key == key && other_value == value)
+                .then_some(*key)
+        })
+    }
+
+    /// See [`owner_is_bounded`].
+    pub fn is_bounded(&self, serial: Serial) -> bool {
+        self.keys
+            .get(&serial)
+            .is_some_and(|keys| keys.iter().any(|(key, _)| *key != OwnerKey::ClientId))
+    }
+}
+
 /// The strongest key two nodes share, or `None` if they share none. Used to
 /// *name* the key in a bridge decision; membership itself is transitive and
 /// comes from [`OwnerComponents`].

src/host/taint/snapshot.rs

@@ -116,6 +116,16 @@ pub struct NodeProps {
     /// The stream negotiated an encoded/passthrough format; a second link
     /// would refuse or corrupt it (v3.4 §6.2).
     pub passthrough: bool,
+    /// `device.id` — set only on nodes exported from a real `Device`
+    /// object by the session manager.
+    ///
+    /// ✅ **Measured 2026-07-21**: this is the discriminator between "a
+    /// hardware device node, one of a set that all share WirePlumber's
+    /// `client.id`" and "a virtual sink with its own client". The five
+    /// ALSA nodes carry `device.id` 43/45/46 with `device.api=alsa`; the
+    /// three `support.null-audio-sink` nodes carry none. See
+    /// [`super::owner`] exception 2, which turns on exactly this.
+    pub device_id: Option<GlobalId>,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]

src/host/taint/tests.rs

@@ -924,3 +924,354 @@ fn recycled_ids_module_indices_and_link_groups_do_not_inherit_taint() {
         &[("call", call, "peerspeak-owned")],
     );
 }
+
+// ──────────────────────────────────────────────────────────────────────
+// Regressions from Codex round 1 — each of these was a leak or a lost
+// taint history, and each mutation-tests the fix that closed it.
+// ──────────────────────────────────────────────────────────────────────
+
+#[test]
+fn sticky_taint_survives_when_only_the_client_object_remains() {
+    // Finding 1. An app can close every stream it holds while keeping its
+    // PipeWire connection open, then open a fresh one — Firefox does this
+    // constantly. Seeding sticky taint only from live *nodes* let the new
+    // leg come back Eligible while the owner's buffers were still full of
+    // the call.
+    let (mut graph, call, rec_in, rec_out, firefox) = sticky_scene();
+    let c = ctx();
+    let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default());
+
+    // Both legs vanish; the client object does not.
+    let late = graph.app_node("rec-out-late", MediaRole::StreamOutput, 8080);
+    let (next, _) = evaluate(&graph.build_without(&[rec_in, rec_out]), &c, &sticky);
+    assert_partition(
+        &next,
+        &[("firefox", firefox)],
+        &[
+            ("call", call, "peerspeak-owned"),
+            ("late", late, "tainted-owner-bridge"),
+        ],
+    );
+}
+
+#[test]
+fn an_inbound_linked_node_of_unknown_role_still_bridges() {
+    // Finding 2. "Receives audio" cannot be inferred from `media.class`
+    // alone: a node with an absent or unexpected class sits on a real
+    // inbound link carrying the call, and if it cannot start an owner
+    // bridge its sibling output leg re-emits the call as Eligible.
+    let mut graph = Graph::new();
+    let sink = graph.device_node("hw-sink", MediaRole::Sink);
+    let call = graph.peerspeak_node("peerspeak", 7);
+    graph.link(call, sink);
+
+    let client = graph.client_of_app(8080);
+    let unknown_in = graph.node("unknown-in", MediaRole::Other, app(client, 8080));
+    let leaked_out = graph.node("leaked-out", MediaRole::StreamOutput, app(client, 8080));
+    graph.link(sink, unknown_in);
+
+    let decisions = run(&graph, &ctx());
+    assert_partition(
+        &decisions,
+        &[],
+        &[
+            ("call", call, "peerspeak-owned"),
+            ("leaked-out", leaked_out, "tainted-owner-bridge"),
+        ],
+    );
+    assert_tainted(&decisions, unknown_in, "tainted-upstream");
+}
+
+#[test]
+fn a_candidate_that_is_itself_a_receiver_still_bridges_to_its_siblings() {
+    // The other half of finding 2: a `Stream/Output/Audio` on an inbound
+    // link is excluded by the link walk, but it must also be able to carry
+    // taint across the owner bridge to its siblings.
+    let mut graph = Graph::new();
+    let sink = graph.device_node("hw-sink", MediaRole::Sink);
+    let call = graph.peerspeak_node("peerspeak", 7);
+    graph.link(call, sink);
+
+    let odd_in = graph.app_node("odd-in", MediaRole::StreamOutput, 9100);
+    let sibling = graph.app_node("sibling-out", MediaRole::StreamOutput, 9100);
+    graph.link(sink, odd_in);
+
+    assert_partition(
+        &run(&graph, &ctx()),
+        &[],
+        &[
+            ("call", call, "peerspeak-owned"),
+            ("odd-in", odd_in, "tainted-upstream"),
+            ("sibling", sibling, "tainted-owner-bridge"),
+        ],
+    );
+}
+
+#[test]
+fn a_native_virtual_sink_bridges_on_its_client() {
+    // Finding 3. The device exception is keyed on `device.id`, not on the
+    // node's role: an app that creates its own `Audio/Sink` plus a
+    // re-emitting stream on one client has no `device.id`, no link-group
+    // and no `pulse.module.id`, so `client.id` is the *only* thing relating
+    // the two. Suppressing coarse keys by role stripped it and leaked the
+    // whole call.
+    let mut graph = Graph::new();
+    let hw = graph.device_node("hw", MediaRole::Sink);
+    let call = graph.peerspeak_node("peerspeak", 7);
+    let virtual_sink = graph.native_virtual_node("native-virtual-sink", MediaRole::Sink, 4321);
+    let reemitter = graph.native_virtual_node("native-output", MediaRole::StreamOutput, 4321);
+    graph.link(call, virtual_sink);
+    graph.link(reemitter, hw);
+
+    let decisions = run(&graph, &ctx());
+    assert_partition(
+        &decisions,
+        &[],
+        &[
+            ("call", call, "peerspeak-owned"),
+            ("reemitter", reemitter, "tainted-owner-bridge"),
+        ],
+    );
+    assert_eq!(
+        decisions.taint[&reemitter.serial].reason,
+        Reason::TaintedOwnerBridge {
+            key: Some(OwnerKey::ClientId)
+        }
+    );
+}
+
+#[test]
+fn an_asymmetric_forwarder_fails_closed() {
+    // Finding 4. The reader carries a `node.link-group` and is therefore
+    // "bounded", but its re-emitting leg carries none, so the strong key
+    // finds no sibling. Requiring the *source* to be unbounded let this
+    // through as Eligible.
+    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 in_client = graph.client(Some(PULSE_PID));
+    let out_client = graph.client(Some(PULSE_PID));
+    let fwd_in = graph.node(
+        "fwd-in",
+        MediaRole::StreamInput,
+        NodeProps {
+            link_group: Some("asymmetric-forwarder".into()),
+            client_id: Some(in_client),
+            process_id: Some(PULSE_PID),
+            ..NodeProps::default()
+        },
+    );
+    let fwd_out = graph.node(
+        "fwd-out",
+        MediaRole::StreamOutput,
+        app(out_client, PULSE_PID),
+    );
+    graph.link(hw, fwd_in);
+
+    // Blast radius control: a real app is bounded by its own PID.
+    let firefox = graph.app_node("firefox", MediaRole::StreamOutput, 11114);
+    graph.link(firefox, hw);
+
+    assert_partition(
+        &run(&graph, &ctx()),
+        &[("firefox", firefox)],
+        &[
+            ("call", call, "peerspeak-owned"),
+            ("fwd-out", fwd_out, "unresolved-owner"),
+        ],
+    );
+}
+
+#[test]
+fn a_not_ready_snapshot_never_retires_sticky_owners() {
+    // Finding 5. `graph_ready == false` says "this view of the graph is not
+    // trustworthy". Retiring sticky owners because their members are absent
+    // from an untrustworthy snapshot erases the taint history, and the next
+    // ready recompute hands back a clean bill of health for an owner that
+    // never went away.
+    let (graph, call, rec_in, rec_out, firefox) = sticky_scene();
+    let c = ctx();
+    let (_, sticky) = evaluate(&graph.build(), &c, &StickyState::default());
+
+    let not_ready = ExclusionCtx {
+        graph_ready: false,
+        ..ctx()
+    };
+    let (_, carried) = evaluate(
+        &super::snapshot::GraphSnapshot::default(),
+        &not_ready,
+        &sticky,
+    );
+    assert_eq!(
+        carried, sticky,
+        "a not-ready recompute must not edit history"
+    );
+
+    let (after, _) = evaluate(&graph.build_without(&[rec_in]), &c, &carried);
+    assert_partition(
+        &after,
+        &[("firefox", firefox)],
+        &[
+            ("call", call, "peerspeak-owned"),
+            ("rec-out", rec_out, "tainted-owner-bridge"),
+        ],
+    );
+}
+
+#[test]
+fn taint_crosses_two_chained_forwarders() {
+    // Finding 6. Every earlier fixture needed at most one owner hop, so
+    // replacing the fixpoint with a single pass of each rule survived the
+    // whole suite. Two chained forwarders need two.
+    let mut graph = Graph::new();
+    let hw = graph.device_node("hw-sink", MediaRole::Sink);
+    let call = graph.peerspeak_node("peerspeak", 7);
+    let sink1 = graph.module_node("sink1", MediaRole::Sink, 100);
+    graph.link(call, sink1);
+
+    let fwd1_in = graph.module_node("fwd1-in", MediaRole::StreamInput, 101);
+    let fwd1_out = graph.module_node("fwd1-out", MediaRole::StreamOutput, 101);
+    graph.link(sink1, fwd1_in);
+
+    let sink2 = graph.module_node("sink2", MediaRole::Sink, 102);
+    graph.link(fwd1_out, sink2);
+    let fwd2_in = graph.module_node("fwd2-in", MediaRole::StreamInput, 103);
+    let fwd2_out = graph.module_node("fwd2-out", MediaRole::StreamOutput, 103);
+    graph.link(sink2, fwd2_in);
+    graph.link(fwd2_out, hw);
+
+    // Negative control: an unrelated clean forwarder chain of the same shape.
+    let clean_sink = graph.device_node("clean-sink", MediaRole::Sink);
+    let clean_in = graph.module_node("clean-in", MediaRole::StreamInput, 200);
+    let clean_out = graph.module_node("clean-out", MediaRole::StreamOutput, 200);
+    graph.link(clean_sink, clean_in);
+    graph.link(clean_out, hw);
+
+    assert_partition(
+        &run(&graph, &ctx()),
+        &[("clean-out", clean_out)],
+        &[
+            ("call", call, "peerspeak-owned"),
+            ("fwd1-out", fwd1_out, "tainted-owner-bridge"),
+            ("fwd2-out", fwd2_out, "tainted-owner-bridge"),
+        ],
+    );
+}
+
+#[test]
+fn a_stronger_reason_replaces_a_weaker_one_regardless_of_arrival_order() {
+    // Finding 7. The old determinism test reversed two links that never
+    // competed, so a first-write-wins `raise()` survived it. Here a node is
+    // reached by the owner bridge on the first fixpoint pass and by a
+    // direct tainted link on the second; the direct link is the stronger
+    // reason and must win.
+    let build = |reverse: bool| {
+        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 rec_in = graph.app_node("rec-in", MediaRole::StreamInput, 5000);
+        let mid = graph.app_node("mid-out", MediaRole::StreamOutput, 5000);
+        let owned_sink = graph.native_virtual_node("owned-sink", MediaRole::Sink, 5000);
+        if reverse {
+            graph.link(mid, owned_sink);
+            graph.link(hw, rec_in);
+        } else {
+            graph.link(hw, rec_in);
+            graph.link(mid, owned_sink);
+        }
+        let decisions = run(&graph, &ctx());
+        (
+            decisions.taint[&owned_sink.serial].reason.code(),
+            decisions.excluded(),
+        )
+    };
+    let forward = build(false);
+    assert_eq!(
+        forward.0, "tainted-upstream",
+        "bridged first, then reached directly — the direct link is stronger"
+    );
+    assert_eq!(forward, build(true), "and it must not depend on link order");
+}
+
+#[test]
+fn a_recycled_client_id_and_module_index_do_not_inherit_taint() {
+    // Finding 8. The original recycling test never reused a
+    // `pulse.module.id`, and its "fresh" client silently reused the dead
+    // client's global id because the fixture cache was stale — so the node
+    // pointed at a client object that no longer existed.
+    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 old_in = graph.module_node("old-in", MediaRole::StreamInput, 536_870_919);
+    let old_out = graph.module_node("old-out", MediaRole::StreamOutput, 536_870_919);
+    graph.link(hw, old_in);
+
+    let c = ctx();
+    let (first, sticky) = evaluate(&graph.build(), &c, &StickyState::default());
+    assert_eq!(
+        first.candidates[&old_out.serial].reason().map(Reason::code),
+        Some("tainted-owner-bridge")
+    );
+
+    // Full teardown, then a module that reuses the node id, the client id
+    // and the module index verbatim. Only the serials are new — which is
+    // the entire reason stickiness is keyed on them.
+    let old_client = graph.client_of_module(536_870_919);
+    graph.drop_clients(&[old_client]);
+    let reborn_client = graph.client_with_id(old_client, Some(PULSE_PID));
+    let reborn = graph.node_with_id(
+        "reborn-out",
+        MediaRole::StreamOutput,
+        old_out.id,
+        super::fixture::pulse_module(reborn_client, 536_870_919, PULSE_PID),
+    );
+    let (second, _) = evaluate(&graph.build_without(&[old_in, old_out]), &c, &sticky);
+    assert_partition(
+        &second,
+        &[("reborn", reborn)],
+        &[("call", call, "peerspeak-owned")],
+    );
+}
+
+#[test]
+fn a_deep_forwarder_chain_converges() {
+    // Finding 9. The fixpoint is monotone so it terminates, but each pass
+    // re-runs a full link BFS and a full owner scan, so a chain of owner
+    // hops costs one pass per layer. This is the shape that would expose an
+    // accidental blow-up; phase 5 measures the real distribution.
+    const LAYERS: u64 = 60;
+    let mut graph = Graph::new();
+    let hw = graph.device_node("hw-sink", MediaRole::Sink);
+    let call = graph.peerspeak_node("peerspeak", 7);
+    let mut upstream = graph.module_node("sink-0", MediaRole::Sink, 1);
+    graph.link(call, upstream);
+    let mut last_out = None;
+    for layer in 1..=LAYERS {
+        let module = 1000 + layer;
+        let leg_in = graph.module_node(&format!("fwd-{layer}-in"), MediaRole::StreamInput, module);
+        let leg_out =
+            graph.module_node(&format!("fwd-{layer}-out"), MediaRole::StreamOutput, module);
+        graph.link(upstream, leg_in);
+        let next = graph.module_node(&format!("sink-{layer}"), MediaRole::Sink, 2000 + layer);
+        graph.link(leg_out, next);
+        upstream = next;
+        last_out = Some(leg_out);
+    }
+    graph.link(last_out.expect("at least one layer"), hw);
+
+    let decisions = run(&graph, &ctx());
+    assert_eq!(
+        decisions.eligible(),
+        Vec::new(),
+        "every leg of the chain carries the call"
+    );
+    assert_eq!(
+        decisions.candidates.len(),
+        LAYERS as usize + 1,
+        "the whole chain plus peerspeak's own playback"
+    );
+}