host/taint: close the partial fixes found in Codex round 2

The verification round earned its place: five of the six round-1 fixes
were partial, and two of the gaps were worse than the bugs they replaced.

1. ⚠️ The round-1 sticky fix smuggled the suppressed key back in.
   `client_serials_of` recorded the shared `WirePlumber [export]` client as
   a member of a tainted hardware sink's owner, so the *second* recompute
   expanded that client to every sound card on the box, tainted the
   microphone, and excluded every app holding one — the §6.1.1 catastrophe
   arriving one epoch late instead of never. `client.id` may now only be
   recorded, or expanded, for nodes where it is a usable owner key.
   The regression test evaluates an unchanged snapshot three times: a
   correct engine's answer must not drift when nothing has.
2. Sticky followed a surviving *connection*, not a surviving *owner*. A
   process can leave one client idle and open a second — GStreamer opens
   one per stream as a matter of course — and the new leg escaped.
   `StickyOwner` now carries owner **fingerprints** (strong keys and a
   usable PID, never `client.id`), applied only while some serial member
   is still live, so a recyclable key cannot resurrect a dead owner.
3. An **ambiguous** link input endpoint tainted every claimant but made
   none of them a receiver, so their sibling output legs stayed eligible.
   Taint without receiver status cannot start an owner bridge.
4. `device.id` is a raw observation, not the classification the coarse-key
   exception needs — PipeWire defines it only as "the Device this node
   belongs to", so a forwarding node carrying one would have lost both its
   owner keys and its ability to trip the backstop. Replaced by
   `session_device`, a phase-3 obligation (`device.id` AND `device.api`)
   documented to fail closed when it cannot classify.
5. Readiness now gates sticky **retirement only**. Round 1 stopped a
   not-ready epoch erasing history; it also stopped it recording any, so a
   reader could consume and buffer the call during that epoch, vanish
   before readiness, and leave its output eligible.
6. Added the unresolved-output-plus-unknown-role fixture: deleting one
   `receivers.insert` survived all 42 previous tests.

Mutation-verified: 7/7 reverts killed by their intended test. Two attempts
did not land first time and both were my error, not the engine's — the
client-key guard is applied at two sites so removing one is not a revert
(removing the pair is, and that is killed), and the fingerprint-lifetime
test put the recycled node in a snapshot *after* the entry had already
been retired, so the guard was never consulted. Rewritten to place it in
the same snapshot that first sees the owner gone.

Cost comment corrected again, to O(D·(V+E+Σ|sources|·|targets|)).

49 tests.

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

src/host/taint/mod.rs

@@ -202,6 +202,17 @@ pub struct StickyOwner {
     /// accumulates: that is what makes "clear only once all member objects
     /// have disappeared" true across churn.
     pub members: BTreeSet<ObjectRef>,
+    /// Owner keys remembered across connections — strong keys and a usable
+    /// process id, never `client.id`. Applied only while some serial member
+    /// above is still live, which is what keeps a recyclable key from
+    /// resurrecting a dead owner.
+    ///
+    /// Needed because a live Client is not the same thing as a live owner:
+    /// a process can leave one connection idle and open a second, and
+    /// GStreamer opens one connection per stream as a matter of course, so
+    /// following connections alone lets the next leg escape (Codex round 2,
+    /// finding 2).
+    pub fingerprints: BTreeSet<owner::Fingerprint>,
     /// The reason recorded for each node that was tainted in its own right.
     /// Kept per node rather than collapsed to one owner-wide reason, or a
     /// forwarder's output leg inherits its *input* leg's `tainted-upstream`
@@ -309,14 +320,16 @@ impl Decisions {
 /// 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.
+/// pass re-runs a full link BFS *and* a full owner scan, and the bridge
+/// scans every tainted source in a component for each target, so the bound
+/// is `O(D · (V + E + Σ_C |sources_C|·|targets_C|))` — worst case
+/// `O(D · (V² + E))` — for an owner-bridge depth D. D is 1 for every
+/// topology observed so far and 2 for a forwarder feeding a forwarder, and
+/// components on a real desktop are two or three nodes; the quadratic term
+/// needs one owner with many legs. A 60-layer chain test guards the depth
+/// dimension only. Phase 5 records the real recompute-duration
+/// distribution and maximum, which is what "full recompute is fine for v1"
+/// should rest on — measured headroom, not a node count.
 pub fn evaluate(
     snapshot: &GraphSnapshot,
     ctx: &ExclusionCtx,
@@ -331,6 +344,7 @@ pub fn evaluate(
     seed_local_roots(snapshot, ctx, &mut taint);
     seed_sticky(
         snapshot,
+        &keys,
         prior,
         &components,
         &mut taint,
@@ -354,17 +368,15 @@ pub fn evaluate(
     }
 
     let decisions = build_decisions(snapshot, ctx, &taint, &sticky_serials);
-    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()
-    };
+    // ⚠️ Readiness gates **retirement only**, never addition (Codex rounds
+    // 1 and 2, which caught the two halves of this in turn). An object
+    // missing from an untrustworthy snapshot has not been observed to
+    // disappear, so retiring on that basis erases history and the next
+    // ready recompute hands back a clean bill of health. But taint
+    // *observed* during a not-ready epoch is real — a reader can consume
+    // and buffer the call and then vanish before readiness — so discarding
+    // additions was the same defect pointing the other way.
+    let next_sticky = build_sticky(snapshot, &keys, &components, &taint, prior, ctx.graph_ready);
     (decisions, next_sticky)
 }
 
@@ -415,8 +427,13 @@ fn local_root_reason(node: &NodeSnapshot, ctx: &ExclusionCtx) -> Option<Reason>
 }
 
 /// Carry taint forward from previous snapshots (v3.4 §6.1.3).
+///
+/// An owner is re-seeded from three kinds of evidence, all lifetime-scoped
+/// to a still-live member: its own surviving nodes, nodes on a surviving
+/// **Client**, and nodes presenting a remembered owner **fingerprint**.
 fn seed_sticky(
     snapshot: &GraphSnapshot,
+    keys: &owner::OwnerKeyIndex,
     prior: &StickyState,
     components: &OwnerComponents,
     taint: &mut BTreeMap<Serial, Reason>,
@@ -431,17 +448,29 @@ fn seed_sticky(
                         live_nodes.push(*serial);
                     }
                 }
-                // ⚠️ A surviving **Client** re-seeds too. An app can close
+                // 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));
+                    live_nodes.extend(nodes_of_client(snapshot, keys, *serial));
                 }
             }
         }
+        if live_nodes.is_empty() && !entry.members.iter().any(|m| is_live(snapshot, *m)) {
+            // Nothing of this owner remains; its fingerprints are just
+            // recyclable strings now and must not be applied to anyone.
+            continue;
+        }
+        // Fingerprints reach a *new connection* of the same still-live
+        // process, which neither of the two paths above can see.
+        for fingerprint in &entry.fingerprints {
+            live_nodes.extend(
+                snapshot
+                    .nodes()
+                    .filter(|node| keys.has_fingerprint(node.serial, fingerprint))
+                    .map(|node| node.serial),
+            );
+        }
         // The owner is sticky, not the individual node: a leg that appears
         // later in the same still-live owner inherits the taint.
         for serial in live_nodes {
@@ -458,7 +487,21 @@ 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> {
+///
+/// Nodes for which `client.id` is not a usable owner key — session-manager
+/// device nodes — are excluded, or the shared `WirePlumber [export]` Client
+/// would drag every sound card on the box into one sticky owner.
+///
+/// The same gate is applied when *recording* clients into a sticky entry
+/// (`owner::client_serials_of`). Either one alone closes the leak; both are
+/// kept because they answer different questions ("may this client be
+/// remembered?" and "may this client speak for that node?"), and the
+/// regression test kills the removal of the pair.
+fn nodes_of_client(
+    snapshot: &GraphSnapshot,
+    keys: &owner::OwnerKeyIndex,
+    client: Serial,
+) -> Vec<Serial> {
     let Some(id) = snapshot
         .clients()
         .find(|c| c.serial == client)
@@ -466,12 +509,10 @@ fn nodes_of_client(snapshot: &GraphSnapshot, client: Serial) -> Vec<Serial> {
     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))
+        .filter(|node| keys.uses_client_key(node.serial))
         .map(|node| node.serial)
         .collect()
 }
@@ -495,6 +536,19 @@ fn downstream_edges(snapshot: &GraphSnapshot, taint: &mut BTreeMap<Serial, Reaso
                 raise(taint, to, Reason::UnresolvedAncestry);
                 receivers.insert(to);
             }
+            (_, Some(IdLookup::Ambiguous)) => {
+                // Several nodes claim the input id and we cannot say which
+                // one this link feeds, so every claimant is a receiver.
+                // They are already tainted as unresolved by their own
+                // ambiguous id — but taint without receiver status cannot
+                // start an owner bridge, so their sibling output legs stayed
+                // Eligible (Codex round 2, finding 3).
+                receivers.extend(
+                    snapshot
+                        .nodes_with_id(link.input_node)
+                        .map(|node| node.serial),
+                );
+            }
             _ => {}
         }
     }
@@ -620,7 +674,7 @@ fn propagate_unresolved_owner(
     taint: &mut BTreeMap<Serial, Reason>,
 ) -> bool {
     let tainted_reader = snapshot.nodes().any(|node| {
-        node.props.device_id.is_none()
+        !node.props.session_device
             && edges.receivers.contains(&node.serial)
             && taint.get(&node.serial).is_some_and(|r| r.propagates())
     });
@@ -708,20 +762,24 @@ fn local_exclusion(snapshot: &GraphSnapshot, node: &NodeSnapshot) -> Option<Reas
 /// all member objects have disappeared" true across churn.
 fn build_sticky(
     snapshot: &GraphSnapshot,
+    keys: &owner::OwnerKeyIndex,
     components: &OwnerComponents,
     taint: &BTreeMap<Serial, Reason>,
     prior: &StickyState,
+    retire_absent: bool,
 ) -> StickyState {
     let mut entries: Vec<StickyOwner> = Vec::new();
 
     // Carry forward prior entries that still have at least one live member.
     // An entry with none is gone for good: serials never recycle, so a
-    // vanished member can never come back.
+    // vanished member can never come back — but only a *trustworthy*
+    // snapshot is allowed to conclude that a member is absent.
     for entry in &prior.owners {
-        if entry
-            .members
-            .iter()
-            .any(|member| is_live(snapshot, *member))
+        if !retire_absent
+            || entry
+                .members
+                .iter()
+                .any(|member| is_live(snapshot, *member))
         {
             entries.push(entry.clone());
         }
@@ -742,12 +800,17 @@ fn build_sticky(
         }
         let mut refs: BTreeSet<ObjectRef> = members.iter().map(|s| ObjectRef::Node(*s)).collect();
         refs.extend(
-            owner::client_serials_of(snapshot, members)
+            owner::client_serials_of(snapshot, keys, members)
                 .into_iter()
                 .map(ObjectRef::Client),
         );
+        let fingerprints = members
+            .iter()
+            .flat_map(|serial| keys.fingerprints(*serial))
+            .collect();
         entries.push(StickyOwner {
             members: refs,
+            fingerprints,
             node_reasons,
         });
     }
@@ -791,6 +854,7 @@ fn merge_overlapping(mut entries: Vec<StickyOwner>) -> Vec<StickyOwner> {
                             .or_insert(reason);
                     }
                     entry.members.extend(other.members);
+                    entry.fingerprints.extend(other.fingerprints);
                     absorbed = true;
                 }
             }