host/observer: address Codex phase-3 review (2 P1 + P3s)

Cross-review round: Codex adversarially reviewed my pure core, found two
merge-blocking P1s and several P3s. Triaged each for reachability; fixes below,
each mutation-verified (revert killed by its intended test).

P1 finding 1 — graph_ready was sticky-once-Complete, so a Link added
post-enumeration whose endpoints are still binding (an INVISIBLE edge, absent
from the snapshot) left graph_ready=true and a candidate could be reported
eligible over unseen tainted ancestry. graph_ready is now dynamic:
Complete AND no outstanding obligations. Readiness::Complete stays sticky as
the epoch marker. New regression test + flipped the old sticky-churn test.

P1 finding 2 — snd_aloop presents with an allowlisted ALSA factory and
device.api=alsa exactly like a real card but forwards audio through a kernel
hop the Link graph cannot see; it was classified session_device=true, dropping
its owner keys + backstop (leak). Added alsa.driver_name to DeviceClaim and a
NON_TERMINAL_ALSA_DRIVERS denylist under the factory allowlist; adapter now
populates it. Negative fixture added.

P3 finding 5 — the BlueZ allowlist entries (api.bluez5.pcm.*) were invented;
removed them (real names are api.bluez5.media.*). A BT sink now over-excludes
(safe) pending a measured fixture. P3 finding 6 — strengthened the timeout
test to assert TimedOut stays sticky through later DeviceAdded/sync/tick.

Findings 3 (dropped-link unrepresented) and 4 (missed-removal generation
ambiguity) documented as accepted low-reachability limitations (links carry
object.serial — confirmed by the live gate; registry does not drop removals).

Codex confirmed the pulse-PID matrix fails safe and the adapter add() FIFO is
lockstep. 120 unit + live gate row 6 green, clippy + fmt clean.

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

src/host/observer/adapter.rs

@@ -39,7 +39,9 @@ impl RegistryObserverHandle {
             .name("pixelpass-pw-observer".to_string())
             .spawn(move || {
                 if let Err(e) = run_observer(latest_for_thread, shutdown_rx) {
-                    tracing::warn!("registry observer: libpipewire thread exited with error: {e:#}");
+                    tracing::warn!(
+                        "registry observer: libpipewire thread exited with error: {e:#}"
+                    );
                 }
             })
             .context("failed to spawn libpipewire registry observer thread")?;
@@ -130,7 +132,10 @@ impl ObserverState {
     /// queues the same length per id — otherwise a phantom slot ahead of a
     /// bound Link would be popped on removal, leaking that Link's proxy.
     fn add(&mut self, id: GlobalId, event: RegEvent) {
-        self.live_globals.entry(id).or_default().push_back(LiveGlobal::default());
+        self.live_globals
+            .entry(id)
+            .or_default()
+            .push_back(LiveGlobal::default());
         self.apply(event);
     }
 
@@ -217,8 +222,7 @@ fn run_observer(
                         );
                         return;
                     };
-                    let Some(serial) =
+                    let Some(serial) = parse_serial(obj.id, "Node", props.get("object.serial"))
-                        parse_serial(obj.id, "Node", props.get("object.serial"))
                     else {
                         return;
                     };
@@ -251,6 +255,7 @@ fn run_observer(
                                 .map(GlobalId),
                             device_api: props.get("device.api").map(str::to_owned),
                             factory_name: props.get("factory.name").map(str::to_owned),
+                            alsa_driver_name: props.get("alsa.driver_name").map(str::to_owned),
                         },
                     };
                     state_for_global
@@ -265,8 +270,7 @@ fn run_observer(
                         );
                         return;
                     };
-                    let Some(serial) =
+                    let Some(serial) = parse_serial(obj.id, "Port", props.get("object.serial"))
-                        parse_serial(obj.id, "Port", props.get("object.serial"))
                     else {
                         return;
                     };
@@ -294,16 +298,17 @@ fn run_observer(
                             return;
                         }
                     };
-                    state_for_global
+                    state_for_global.borrow_mut().add(
-                        .borrow_mut()
+                        id,
-                        .add(id, RegEvent::PortAdded(PortSnapshot {
+                        RegEvent::PortAdded(PortSnapshot {
                             serial,
                             id,
                             node,
                             direction,
                             exclusive: truthy(props.get("port.exclusive")),
                             monitor: truthy(props.get("port.monitor")),
-                        }));
+                        }),
+                    );
                 }
                 ObjectType::Client => {
                     let Some(props) = obj.props.as_ref() else {
@@ -313,20 +318,20 @@ fn run_observer(
                         );
                         return;
                     };
-                    let Some(serial) =
+                    let Some(serial) = parse_serial(obj.id, "Client", props.get("object.serial"))
-                        parse_serial(obj.id, "Client", props.get("object.serial"))
                     else {
                         return;
                     };
-                    state_for_global
+                    state_for_global.borrow_mut().add(
-                        .borrow_mut()
+                        id,
-                        .add(id, RegEvent::ClientAdded(ClientSnapshot {
+                        RegEvent::ClientAdded(ClientSnapshot {
                             serial,
                             id,
                             sec_pid: props
                                 .get("pipewire.sec.pid")
                                 .and_then(|value| value.parse::<u32>().ok()),
-                        }));
+                        }),
+                    );
                 }
                 ObjectType::Device => {
                     state_for_global
@@ -341,8 +346,7 @@ fn run_observer(
                         );
                         return;
                     };
-                    let Some(serial) =
+                    let Some(serial) = parse_serial(obj.id, "Link", props.get("object.serial"))
-                        parse_serial(obj.id, "Link", props.get("object.serial"))
                     else {
                         return;
                     };
@@ -388,19 +392,19 @@ fn run_observer(
                                 input_port: optional_global_id(info.input_port_id()),
                             };
                             if let Some(state) = state_for_info.upgrade() {
-                                state.borrow_mut().apply(RegEvent::LinkEndpointsResolved {
+                                state
-                                    serial,
+                                    .borrow_mut()
-                                    endpoints,
+                                    .apply(RegEvent::LinkEndpointsResolved { serial, endpoints });
-                                });
                             }
                         })
                         .register();
-                    state_for_global
+                    state_for_global.borrow_mut().attach_bound_link(
-                        .borrow_mut()
+                        id,
-                        .attach_bound_link(id, BoundLink {
+                        BoundLink {
                             _proxy: link,
                             _listener: listener,
-                        });
+                        },
+                    );
                 }
                 _ => {}
             }
@@ -566,10 +570,7 @@ mod tests {
         let unique = format!("pixelpass_observer_test_{}", std::process::id());
         let capture_name = format!("{unique}_capture");
         let playback_name = format!("{unique}_playback");
-        let null_sink = PactlModule::load(
+        let null_sink = PactlModule::load("module-null-sink", &[format!("sink_name={unique}")]);
-            "module-null-sink",
-            &[format!("sink_name={unique}")],
-        );
         let with_sink = wait_for(&observer, |projection| has_node(projection, &unique));
         let sink_links = with_sink.snapshot.links().count();
 

src/host/observer/classify.rs

@@ -39,15 +39,33 @@ use crate::host::taint::snapshot::GlobalId;
 /// `support.null-audio-sink`, `*.loopback`, and any filter factory are
 /// intentionally **absent**: those forward audio, which is exactly the shape
 /// this feature must be able to exclude.
+///
+/// ⚠️ **ALSA only, and only these two, because they are the only factories
+/// measured on the target box.** BlueZ was previously listed here as
+/// `api.bluez5.pcm.{sink,source}` — those are invented; the real BlueZ
+/// terminals are `api.bluez5.media.{sink,source}` with profile aliases
+/// (Codex phase-3 review, finding 5). Rather than allowlist an unmeasured
+/// guess, BlueZ is left off entirely: a real Bluetooth sink then keeps its
+/// owner keys (over-exclusion — safe). Add BlueZ back only with a *measured*
+/// factory name and a fixture.
 const HARDWARE_PCM_FACTORIES: &[&str] = &[
     // ALSA — measured on the target box.
     "api.alsa.pcm.sink",
     "api.alsa.pcm.source",
-    // BlueZ — the equivalent real backend for Bluetooth audio terminals.
-    "api.bluez5.pcm.sink",
-    "api.bluez5.pcm.source",
 ];
 
+/// ALSA drivers that expose a hardware-PCM `factory.name` but are **not**
+/// passive terminals — audio written in reappears on their capture side
+/// through a path the PipeWire Link graph cannot see, so classifying them
+/// `session_device` (which drops owner keys and the fail-closed backstop)
+/// would let tainted audio loop back untainted (Codex phase-3 review,
+/// finding 2). `factory.name` alone cannot distinguish these from a real
+/// card — `snd_aloop` presents as `api.alsa.pcm.{sink,source}` exactly like
+/// `snd_hda_intel` — so this is a necessary denylist layered under the
+/// factory allowlist, keyed on the one property that does distinguish them
+/// (`alsa.driver_name`, measured present on real ALSA nodes).
+const NON_TERMINAL_ALSA_DRIVERS: &[&str] = &["snd_aloop"];
+
 /// The three node properties the classifier reads, exactly as the adapter
 /// parsed them off the Node global. Kept separate from
 /// [`super::super::taint::snapshot::NodeProps`] because these feed the
@@ -67,6 +85,12 @@ pub struct DeviceClaim {
     /// `factory.name` — the discriminator. Only an allowlisted hardware-PCM
     /// factory earns `session_device`.
     pub factory_name: Option<String>,
+    /// `alsa.driver_name` — the kernel driver behind an ALSA node (e.g.
+    /// `snd_hda_intel`, `snd_usb_audio`, `snd_aloop`). Needed because the
+    /// factory allowlist cannot tell a real card from a loopback driver that
+    /// shares the same factory; a driver on [`NON_TERMINAL_ALSA_DRIVERS`]
+    /// forfeits `session_device`. Absent on non-ALSA backends.
+    pub alsa_driver_name: Option<String>,
 }
 
 /// The outcome of classifying one node's device claim.
@@ -104,11 +128,16 @@ pub fn classify(claim: &DeviceClaim, device_resolved: bool) -> Classification {
         // forbids.
         return Classification::Withhold { device_id };
     }
-    let is_hardware_pcm = claim.device_api.is_some()
+    let on_factory_allowlist = claim
-        && claim
+        .factory_name
-            .factory_name
+        .as_deref()
-            .as_deref()
+        .is_some_and(|f| HARDWARE_PCM_FACTORIES.contains(&f));
-            .is_some_and(|f| HARDWARE_PCM_FACTORIES.contains(&f));
+    let non_terminal_driver = claim
+        .alsa_driver_name
+        .as_deref()
+        .is_some_and(|d| NON_TERMINAL_ALSA_DRIVERS.contains(&d));
+    let is_hardware_pcm =
+        claim.device_api.is_some() && on_factory_allowlist && !non_terminal_driver;
     if is_hardware_pcm {
         Classification::SessionDevice
     } else {

src/host/observer/mod.rs

@@ -23,6 +23,25 @@
 //!   whose Device we have not observed is held out of the snapshot entirely
 //!   rather than admitted with a provisional `session_device` (see
 //!   [`classify`]).
+//!
+//! **Two accepted limitations (Codex phase-3 review, findings 3 and 4), both
+//! low-reachability, owed to a later hardening round:**
+//!
+//! - *A Link dropped for a missing `object.serial`/props is unrepresented.*
+//!   The adapter drops such a global before it reaches [`RegistryModel`], so
+//!   readiness can reach `Complete` while permanently omitting that Link — an
+//!   invisible edge that could hide tainted ancestry. **Not reachable in
+//!   practice:** every real Link global carries `object.serial` (confirmed by
+//!   the live gate, which only counts links the strict parser admits). A full
+//!   fix needs a pure "required-observation-failed" token that holds readiness
+//!   false; deferred rather than built for a case that does not occur.
+//! - *Removal generation ordering assumes no removal is silently lost.* On a
+//!   recycled id with two live claimants, [`Self::on_removed`] retires the
+//!   oldest generation first; if the *first* generation's removal was never
+//!   delivered, a later removal is misattributed. PipeWire's registry does not
+//!   silently drop `global_remove`, so this needs callback loss to trigger.
+//!   The snapshot treats the two-claimant window as [`IdLookup::Ambiguous`]
+//!   (fail closed) meanwhile.
 
 #![allow(dead_code)] // Wired by the phase-3 adapter (Codex's half) and consumed by later phases.
 
@@ -131,7 +150,12 @@ enum Slot {
 pub enum Readiness {
     /// The initial enumeration is still in flight.
     Waiting,
-    /// Server synced and every obligation resolved. `graph_ready` is true.
+    /// The initial enumeration finished at least once (server synced with no
+    /// obligations then outstanding). **Sticky** — later per-object
+    /// withholding does not revert it. Note this is *not* the same as
+    /// [`RegistryModel::graph_ready`], which additionally requires no *current*
+    /// obligation (Codex finding 1); `Complete` only records that the epoch
+    /// was reached.
     Complete,
     /// The bounded deadline passed with obligations outstanding.
     /// `graph_ready` stays false — fail closed.
@@ -207,8 +231,28 @@ impl RegistryModel {
         self.readiness
     }
 
+    /// Whether the graph is trustworthy enough to make eligibility and sticky
+    /// **retirement** decisions right now.
+    ///
+    /// This is **dynamic**, not the sticky [`Readiness::Complete`] flag: it is
+    /// true only when the initial enumeration has completed **and** there are
+    /// no current obligations outstanding (a node withheld on an unresolved
+    /// Device, or a Link still being bound). The distinction is the fix for
+    /// Codex phase-3 review finding 1: a Link whose endpoints are still
+    /// resolving is an **invisible edge** — it is absent from the snapshot,
+    /// not merely dangling — so a decision made while one exists can miss real
+    /// tainted ancestry and wrongly report a candidate eligible. Unresolved
+    /// ancestry ⇒ fail closed is the governing invariant (v3.4 §6.1), and an
+    /// unresolved Link is unresolved ancestry, so `graph_ready` must drop back
+    /// to false whenever one is pending — even after the initial epoch.
+    ///
+    /// [`Readiness::Complete`] stays sticky (it records that the initial
+    /// enumeration happened, for logging and to distinguish "not started" from
+    /// "momentarily churning"); `graph_ready` layers the dynamic obligation
+    /// check on top. Downstream (phase 6) may debounce the brief blips a
+    /// normal Link bind causes; the observer's job is to report the truth.
     pub fn graph_ready(&self) -> bool {
-        matches!(self.readiness, Readiness::Complete)
+        matches!(self.readiness, Readiness::Complete) && !self.obligations_outstanding()
     }
 
     /// The pulse-PID candidate the adapter should be probing (`None` = no

src/host/observer/tests.rs

@@ -37,6 +37,7 @@ fn hw_claim(device_id: u32, api: &str, factory: &str) -> DeviceClaim {
         device_id: Some(gid(device_id)),
         device_api: Some(api.to_string()),
         factory_name: Some(factory.to_string()),
+        alsa_driver_name: Some("snd_hda_intel".to_string()),
     }
 }
 
@@ -114,25 +115,53 @@ fn classify_unresolved_device_withholds() {
 
 #[test]
 fn classify_resolved_hardware_pcm_is_session_device() {
-    for factory in [
+    // Only the measured ALSA factories are allowlisted (finding 5: the BlueZ
-        "api.alsa.pcm.sink",
+    // entries were invented and were removed).
-        "api.alsa.pcm.source",
+    for factory in ["api.alsa.pcm.sink", "api.alsa.pcm.source"] {
-        "api.bluez5.pcm.sink",
-        "api.bluez5.pcm.source",
-    ] {
-        let api = if factory.contains("bluez5") {
-            "bluez5"
-        } else {
-            "alsa"
-        };
         assert_eq!(
-            classify(&hw_claim(7, api, factory), true),
+            classify(&hw_claim(7, "alsa", factory), true),
             Classification::SessionDevice,
             "factory {factory} should be a session device"
         );
     }
 }
 
+#[test]
+fn classify_invented_bluez_factories_are_not_session_devices() {
+    // Finding 5: `api.bluez5.pcm.*` is not a real factory name; whatever it is,
+    // it is not on the measured allowlist, so it fails closed to false
+    // (over-exclusion, safe) rather than being trusted.
+    for factory in ["api.bluez5.pcm.sink", "api.bluez5.pcm.source"] {
+        let claim = DeviceClaim {
+            device_id: Some(gid(7)),
+            device_api: Some("bluez5".to_string()),
+            factory_name: Some(factory.to_string()),
+            alsa_driver_name: None,
+        };
+        assert_eq!(classify(&claim, true), Classification::NotSessionDevice);
+    }
+}
+
+#[test]
+fn classify_snd_aloop_is_not_a_session_device() {
+    // Finding 2: an ALSA loopback presents with an allowlisted factory and
+    // device.api=alsa exactly like a real card, but forwards audio through a
+    // kernel hop the Link graph cannot see. It must NOT earn session_device.
+    for factory in ["api.alsa.pcm.sink", "api.alsa.pcm.source"] {
+        let claim = DeviceClaim {
+            device_id: Some(gid(7)),
+            device_api: Some("alsa".to_string()),
+            factory_name: Some(factory.to_string()),
+            alsa_driver_name: Some("snd_aloop".to_string()),
+        };
+        assert_eq!(
+            classify(&claim, true),
+            Classification::NotSessionDevice,
+            "snd_aloop {factory} must fail closed"
+        );
+    }
+}
+
 #[test]
 fn classify_resolved_but_not_hardware_pcm_fails_closed() {
     // A null sink, a loopback, and an unknown factory are all forwarders, not
@@ -154,6 +183,7 @@ fn classify_missing_device_api_fails_closed() {
         device_id: Some(gid(7)),
         device_api: None,
         factory_name: Some("api.alsa.pcm.sink".to_string()),
+        alsa_driver_name: Some("snd_hda_intel".to_string()),
     };
     assert_eq!(classify(&claim, true), Classification::NotSessionDevice);
 }
@@ -488,6 +518,15 @@ fn model_readiness_times_out_fail_closed() {
     m.apply(RegEvent::Tick { now: 5000 });
     assert_eq!(m.readiness(), Readiness::TimedOut);
     assert!(!m.graph_ready(), "timeout fails closed");
+
+    // Finding 6: TimedOut must be sticky. Resolving the obligation, syncing
+    // again, and ticking further must NOT flip it to Complete — a timed-out
+    // observer stays fail-closed for its lifetime.
+    m.apply(RegEvent::DeviceAdded { id: gid(42) });
+    m.apply(RegEvent::ServerSynced);
+    m.apply(RegEvent::Tick { now: 6000 });
+    assert_eq!(m.readiness(), Readiness::TimedOut, "timeout is sticky");
+    assert!(!m.graph_ready());
 }
 
 #[test]
@@ -504,23 +543,61 @@ fn model_tick_before_deadline_does_not_time_out() {
 }
 
 #[test]
-fn model_complete_is_sticky_across_later_churn() {
+fn model_complete_epoch_is_sticky_but_graph_ready_is_dynamic() {
     let mut m = model();
     m.apply(RegEvent::ServerSynced);
     assert_eq!(m.readiness(), Readiness::Complete);
-    // A node withheld AFTER completion must not un-complete the epoch — post
+    assert!(m.graph_ready());
-    // enumeration, withholding is per-object (the node is simply absent).
+    // A node withheld AFTER completion does not revert the sticky EPOCH...
     m.apply(device_node(
         100,
         50,
         MediaRole::Sink,
         hw_claim(42, "alsa", "api.alsa.pcm.sink"),
     ));
-    assert_eq!(m.readiness(), Readiness::Complete);
+    assert_eq!(m.readiness(), Readiness::Complete, "epoch stays sticky");
-    assert!(m.graph_ready());
+    // ...but graph_ready DOES drop while the obligation is outstanding
-    // ...and a late timeout Tick is inert once Complete.
+    // (Codex finding 1: unresolved ancestry ⇒ fail closed, even post-epoch).
+    assert!(
+        !m.graph_ready(),
+        "an outstanding obligation makes decisions unsafe"
+    );
+    // A late timeout Tick is inert once Complete.
     m.apply(RegEvent::Tick { now: 100_000 });
     assert_eq!(m.readiness(), Readiness::Complete);
+    // Resolving the obligation restores graph_ready.
+    m.apply(RegEvent::DeviceAdded { id: gid(42) });
+    assert!(m.graph_ready());
+}
+
+#[test]
+fn model_pending_link_drops_graph_ready_after_completion() {
+    // Codex finding 1, the leak that mattered: a real Link added post-epoch
+    // whose endpoints are still binding is an INVISIBLE edge (absent from the
+    // snapshot, not dangling). graph_ready must go false until it resolves,
+    // or a candidate can be reported eligible while tainted ancestry it cannot
+    // see already carries call audio.
+    let mut m = model();
+    m.apply(RegEvent::ServerSynced);
+    assert!(m.graph_ready());
+
+    m.apply(RegEvent::LinkAdded {
+        serial: ser(300),
+        id: gid(90),
+        endpoints: None,
+    });
+    assert!(!m.graph_ready(), "an unresolved link must gate decisions");
+    // The snapshot genuinely omits it, which is exactly why graph_ready must
+    // compensate.
+    assert_eq!(m.project().snapshot.links().count(), 0);
+    assert!(!m.project().graph_ready);
+
+    m.apply(RegEvent::LinkEndpointsResolved {
+        serial: ser(300),
+        endpoints: endpoints(50, 55),
+    });
+    assert!(m.graph_ready(), "resolved ⇒ decisions safe again");
+    assert_eq!(m.project().snapshot.links().count(), 1);
 }
 
 #[test]