host/observer: derive pipewire-pulse's PID from comm, not from repetition

MEASURED DEFECT, found by the §5.1 row-1 matrix run: the daemon PID was
unresolvable on this host, permanently, which switched key 4's suppression off
and fused every Pulse-emulated node into a single owner. Row 1's clean control
forwarder was excluded, Firefox was excluded across an application.process.id
bridge, and the taint reached the sunshine sinks and both sound cards -- the
same machine-wide over-exclusion cascade as phase 5's F2, from a new cause.

Stage 1 returned the one sec_pid shared by 2+ Clients, reasoning that only the
Pulse shim repeats a value. WirePlumber repeats one too: it holds Clients
'WirePlumber' and 'WirePlumber [export]', both sec_pid 1747. Two values
repeated, the rule called that ambiguous and returned None, and
owner::keys_of's documented fail-closed asymmetry did the rest.

The rule was wrong in both directions, so the prefilter is gone rather than
patched: a second process holding two Clients defeats it (permanent, not a
corner case), and a session where pipewire-pulse holds exactly one Client never
repeats anything so the candidate is missed. comm was always the authoritative
check; repetition stood in front of it and was a guess about other processes'
Client counts.

Now: candidates() lists every distinct sec_pid, resolve() picks the unique one
whose /proc comm is exactly pipewire-pulse, and several matches still fail
closed (a single Option<u32> cannot suppress two daemons -- recorded, not
approximated). The adapter probes only PIDs entering the set, and
retain_probed_comms bounds the map to live PIDs so a PID that leaves and
returns is re-probed instead of answered from a stale comm.

Row 1 now passes its exact partition, key named: tainted forwarder leg excluded
on node.link-group, clean forwarder leg and Firefox eligible, taint confined to
the tainted half. 225 tests green (WirePlumber-pair and single-Client
regressions covered), clippy clean.

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() {
-    let clients = clients_with(&[Some(4137), Some(4137), Some(9001), Some(9001)]);
-    assert_eq!(pulse_pid::candidate(&clients), None);
+fn pid_candidates_are_every_distinct_sec_pid() {
+    let clients = clients_with(&[Some(4137), Some(4137), Some(9001), 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() {
-    // One client per pid: nothing repeats, so nothing is pipewire-pulse.
+fn pid_resolve_no_match_is_none() {
     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();