host/aec: address Codex phase-4 review (no merge-blockers; docs + test holes)

Codex adversarial review found no merge-blocking defects. Five
worth-checking items, all triaged for reachability:

- F1 (spurious pre-ready revoke): unreachable — the AEC's four nodes are
  two Stream/* legs + a null-sink-like virtual sink/source, none claiming
  a device.id, so the phase-3 observer never withholds them; index_present
  goes false only on a genuine full unload. Documented why revoke is NOT
  gated on graph_ready, and why gating it would reopen the reused-index
  alias trap (F4) during a hot-reload-under-churn. Pinned with
  revokes_on_empty_even_while_not_ready (mutation-verified: `&& graph_ready`
  on the revoke guard dies here).
- F4 (test relies on observing the empty gap): documented the phase-5/6
  integration contract it rests on (one observe per graph event, no
  coalescing across a module lifetime boundary) and owed the robust fix
  (serial-continuity / observer-generation) to a later hardening round.
- F2 (late positive evidence beats the deadline): intentional and correct
  — a demonstrably-present identity is ground truth. Documented +
  late_positive_evidence_wins_over_expired_deadline (both arms: node-first
  validates, Tick-first fails closed and stays sticky).
- F3 (real P3 coverage hole): strengthened deadline_is_not_armed_until_
  graph_ready to prove the budget starts at first-ready, not construction
  (mutation-verified: a construction-relative deadline now dies).
- F5 (`+7` grammar mismatch): documented the producer contract — peerspeak
  emits bare decimal (pactl returns unsigned decimal), the narrow parser
  is deliberate. Unreachable on the measured stack.

No core logic change. 24 pure aec tests, cargo test --bins green (145 unit
+ 1 ignored live), clippy -D warnings + fmt clean.

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

src/host/aec/mod.rs

@@ -0,0 +1,311 @@
+//! Phase 4 — the AEC identity validation state machine (impl plan §4, design
+//! v3.4 §5.2/§5.3).
+//!
+//! peerspeak's echo canceller (`module-echo-cancel`) creates four graph nodes
+//! that all carry `pulse.module.id == <the index pactl returned>`, and the
+//! playback leg among them is a `Stream/Output/Audio` node wired straight to
+//! the speakers — a fan-out candidate that would copy the whole remote call
+//! into the share unless it is excluded (v3.4 §5.2, measured ≈desktop level).
+//! The taint engine (phase 2) already excludes it *given* the module index in
+//! [`ExclusionCtx::aec_module_id`](crate::host::taint::ExclusionCtx); this
+//! module is what decides, at runtime and fail-closed, whether that index may
+//! be trusted and handed over.
+//!
+//! **Why a state machine and not a one-shot check (v3.4 §5.3).** The identity
+//! is an *observed correlation on PipeWire 1.6.8*, not a documented contract,
+//! and a start-time enumeration races in both directions: peerspeak's
+//! `enable()` returns before the playback hazard leg is even in the graph, and
+//! pixelpass's capture spawns lazily on the first viewer, at a moment peerspeak
+//! does not control. So validation is a bounded epoch, and the identity can be
+//! *lost* mid-share (the module unloads) as well as *gained*.
+//!
+//! **The two traps this is shaped around:**
+//!
+//! - **Revocation is loss of the whole module identity, not one leg corking**
+//!   (v3.4 §5.3). Each [`AecValidator::observe`] rescans the snapshot for *any*
+//!   node bearing the index; [`AecState::Validated`] drops to
+//!   [`AecState::Revoked`] only when that set becomes **empty**. A single leg
+//!   corking or relinking (still ≥1 present) stays `Validated` — getting this
+//!   wrong turns a normal cork into a spurious share-wide audio stop.
+//! - **Module indices are reused verbatim across unload/reload** (v3.4 §5.2
+//!   correction 3 — both a reload's module index *and* its `node.link-group`
+//!   came back byte-identical, and node ids were recycled *and reassigned
+//!   across legs*). So [`AecState::Failed`] and [`AecState::Revoked`] are
+//!   **sticky terminal**: a later node reappearing with the same index does
+//!   **not** un-revoke and alias onto the new module. A genuine reload gets a
+//!   *fresh* [`AecValidator`] (peerspeak re-tells pixelpass the index on every
+//!   load), never a resurrected one.
+//!
+//! **Scope.** This is the validation state machine + `--aec` parsing only.
+//! Foreign / second-AEC detection (a non-owned `echo-cancel-*` group, v3.4
+//! §5.4 / D3) and the `foreign_aec_warning`/`aec_failed`/`aec_revoked` status
+//! *events* are phase 6's, which reads this machine's [`AecState`]. Wiring the
+//! parsed [`AecConfig`] out of the CLI and calling [`AecValidator::observe`]
+//! in the recompute loop is integration (phases 5/8). The node-side
+//! `pulse.module.id` parse (JSON-number-vs-string, u64-not-u32) is phase 3's
+//! adapter; this module consumes the already-parsed
+//! [`NodeProps::pulse_module_id`](crate::host::taint::snapshot::NodeProps).
+
+#![allow(dead_code)] // Wired into `--aec` parsing + the recompute loop by later phases.
+
+#[cfg(test)]
+mod tests;
+
+use crate::host::observer::Millis;
+use crate::host::taint::snapshot::GraphSnapshot;
+
+/// The parsed `--aec=off|pulse-module:<idx>` argument (decision D5).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AecConfig {
+    /// `--aec=off` — peerspeak's AEC is not in play, so there is nothing to
+    /// exclude and fan-out proceeds with no AEC identity. Not the same as an
+    /// *absent* argument (that default is the caller's; see [`parse_aec_arg`]).
+    Off,
+    /// `--aec=pulse-module:<idx>` — validate this live module index before
+    /// trusting it. The index is compared as `u64`, never `u32` (v3.4 §5.2).
+    PulseModule(u64),
+}
+
+/// Why an `--aec` argument was rejected. Rejection is fatal at the CLI edge —
+/// there is no fail-closed *default* index, because a wrong index would exclude
+/// the wrong node (or nothing), so a malformed value must not silently become
+/// "no AEC".
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AecParseError {
+    /// The value was empty.
+    Empty,
+    /// Not `off` and not `pulse-module:...`.
+    UnknownForm,
+    /// `pulse-module:` with nothing after the colon.
+    MissingIndex,
+    /// The index was not a bare `u64` decimal (sign, whitespace, non-digit, or
+    /// `> u64::MAX`).
+    InvalidIndex,
+}
+
+/// Parse one `--aec` value. `off` and `pulse-module:<idx>` are the only forms.
+///
+/// The index accepts values `> u32::MAX` (v3.4 §5.2: `pulse.module.id` sits
+/// next to the `object.serial` u32-truncation bug, so it is only ever compared
+/// as `u64`) and requires a **bare decimal** — stricter than Rust's [`u64`]
+/// parser, which also accepts a leading `+`. Rejected: any sign, surrounding or
+/// interior whitespace, non-decimal digits, and overflow. Matching is exact and
+/// case-sensitive: the argument is machine-generated by peerspeak from
+/// `EchoCancelGuard::module_index`, not typed by a user.
+///
+/// ⚠️ **Producer contract** (Codex phase-4 review, finding 5): because the
+/// grammar is narrower than Rust's parser, peerspeak must emit a bare decimal.
+/// `pactl load-module` returns an unsigned decimal, so the stored index is
+/// already canonical and no reachable value is rejected; if peerspeak ever
+/// changes how it formats the index it must canonicalize (`value.to_string()`),
+/// not widen this parser — the narrow grammar is the point.
+pub fn parse_aec_arg(value: &str) -> Result<AecConfig, AecParseError> {
+    if value.is_empty() {
+        return Err(AecParseError::Empty);
+    }
+    if value == "off" {
+        return Ok(AecConfig::Off);
+    }
+    if let Some(index) = value.strip_prefix("pulse-module:") {
+        if index.is_empty() {
+            return Err(AecParseError::MissingIndex);
+        }
+        // A bare decimal only: reject a leading sign (Rust's `u64` parser
+        // accepts `+7`), interior/surrounding whitespace, and any non-digit,
+        // before letting the parser catch overflow. Leading zeros are harmless.
+        if !index.bytes().all(|b| b.is_ascii_digit()) {
+            return Err(AecParseError::InvalidIndex);
+        }
+        return index
+            .parse::<u64>()
+            .map(AecConfig::PulseModule)
+            .map_err(|_| AecParseError::InvalidIndex);
+    }
+    Err(AecParseError::UnknownForm)
+}
+
+/// The validation epoch (v3.4 §5.3, verbatim).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AecState {
+    /// `--aec=off` — no AEC identity, fan-out proceeds with no exclusion.
+    /// Terminal.
+    NotConfigured,
+    /// Waiting for the first node bearing the index. **No fan-out occurs here**
+    /// — silence is the safe direction. Ends at `Validated` on first sight, or
+    /// `Failed` once the graph is fully enumerated and the bounded deadline
+    /// passes with the index never seen.
+    Validating,
+    /// The index was observed live. Fan-out is permitted, excluding that
+    /// identity transitively (phase 2 / v3.4 §6.1).
+    Validated,
+    /// The deadline expired with the index never observed. **Fail closed** — no
+    /// fan-out; the caller reports a capability failure rather than sharing.
+    /// Sticky terminal.
+    Failed,
+    /// The whole module identity disappeared mid-share (every node bearing the
+    /// index gone). **Stop fan-out now** and drop the owned link proxies; do
+    /// not keep the numeric index and hope, because it is reused. Sticky
+    /// terminal — see the module header's second trap.
+    Revoked,
+}
+
+/// The bounded, read-only AEC identity validator. Fold the live graph in with
+/// [`AecValidator::observe`] once per recompute; read the result with
+/// [`AecValidator::state`], [`AecValidator::fan_out_permitted`], and
+/// [`AecValidator::validated_module_id`].
+#[derive(Clone, Debug)]
+pub struct AecValidator {
+    /// The index to validate. `None` iff [`AecConfig::Off`] (state stays
+    /// [`AecState::NotConfigured`] forever).
+    target: Option<u64>,
+    state: AecState,
+    /// The `Validating → Failed` budget, applied *after* the deadline is armed.
+    timeout: Millis,
+    /// The absolute `Failed` deadline, armed the first time the graph reports
+    /// ready (the "registry sync barrier" of v3.4 §5.3) and never re-armed —
+    /// `graph_ready` is dynamic and can flap, but the epoch budget must not
+    /// restart. `None` until then: while the initial enumeration is still in
+    /// flight, a not-yet-seen index is *unknown*, not *absent*, so it must not
+    /// time out to `Failed`.
+    deadline: Option<Millis>,
+}
+
+impl AecValidator {
+    /// `timeout` is the `Validating → Failed` budget, counted from the moment
+    /// the graph first becomes ready (not from construction). An `Off` config
+    /// starts (and stays) [`AecState::NotConfigured`].
+    pub fn new(config: AecConfig, timeout: Millis) -> Self {
+        match config {
+            AecConfig::Off => Self {
+                target: None,
+                state: AecState::NotConfigured,
+                timeout,
+                deadline: None,
+            },
+            AecConfig::PulseModule(index) => Self {
+                target: Some(index),
+                state: AecState::Validating,
+                timeout,
+                deadline: None,
+            },
+        }
+    }
+
+    pub fn state(&self) -> AecState {
+        self.state
+    }
+
+    /// The validated index to place in
+    /// [`ExclusionCtx::aec_module_id`](crate::host::taint::ExclusionCtx) —
+    /// `Some` **only** in [`AecState::Validated`]. `None` everywhere else,
+    /// including `NotConfigured` (no AEC ⇒ nothing to exclude) and the
+    /// fail-closed states (whose `None` must be paired with
+    /// [`Self::fan_out_permitted`] `== false`, i.e. no fan-out at all — *not*
+    /// a fan-out that merely skips AEC exclusion).
+    pub fn validated_module_id(&self) -> Option<u64> {
+        match self.state {
+            AecState::Validated => self.target,
+            _ => None,
+        }
+    }
+
+    /// Whether fan-out may proceed at all right now. True only in
+    /// [`AecState::NotConfigured`] (fan out, no exclusion) and
+    /// [`AecState::Validated`] (fan out, excluding the identity). `Validating`,
+    /// `Failed` and `Revoked` all forbid it — silence over echo.
+    pub fn fan_out_permitted(&self) -> bool {
+        matches!(self.state, AecState::NotConfigured | AecState::Validated)
+    }
+
+    /// Fold one recompute's view of the graph into the machine.
+    ///
+    /// `graph_ready` is the observer's dynamic readiness
+    /// ([`Projection::graph_ready`](crate::host::observer::Projection)); `now`
+    /// is a monotonic millisecond clock. Positive evidence (a node bearing the
+    /// index) is authoritative and validates regardless of `graph_ready` —
+    /// seeing the node *is* seeing it — but the `Failed` deadline only begins
+    /// once `graph_ready` has first become true, so a slow initial enumeration
+    /// can never masquerade as a genuinely-absent module.
+    pub fn observe(&mut self, snapshot: &GraphSnapshot, graph_ready: bool, now: Millis) {
+        // `Off` (NotConfigured) and both sticky terminals are no-ops: there is
+        // nothing to look for, and a reappearing reused index must not revive a
+        // Failed/Revoked epoch (v3.4 §5.2 correction 3).
+        let Some(target) = self.target else {
+            return;
+        };
+        match self.state {
+            AecState::Validating => {
+                // Presence is checked *before* the deadline on purpose: a
+                // demonstrably-present identity validates regardless of the
+                // clock, even if the node is first seen just past the deadline
+                // (Codex phase-4 review, finding 2). The deadline only bounds
+                // the wait for an identity that is never seen — seeing it, late
+                // or not, is ground truth that the module exists, and excluding
+                // a real echo leg is always the safe answer. (A `Failed` can
+                // still pre-empt this when a `Tick`-only observation crosses the
+                // deadline first; that only makes the machine *more* fail-closed,
+                // never less.)
+                if self.index_present(snapshot, target) {
+                    self.state = AecState::Validated;
+                    return;
+                }
+                // Arm the deadline once, on the first ready graph.
+                if self.deadline.is_none() && graph_ready {
+                    self.deadline = Some(now.saturating_add(self.timeout));
+                }
+                if self.deadline.is_some_and(|deadline| now >= deadline) {
+                    self.state = AecState::Failed;
+                }
+            }
+            AecState::Validated => {
+                // Revocation is the whole identity gone (no node bears the
+                // index), not one leg corking — see the module header.
+                //
+                // ⚠️ **Deliberately NOT gated on `graph_ready`** (Codex
+                // phase-4 review, findings 1 + 4). Two forces pull opposite
+                // ways and this is the resolution:
+                //
+                // - Gating revoke on readiness would avoid a *spurious* revoke
+                //   from a transient empty snapshot seen while the module is
+                //   still live. But for the AEC that transient does not exist:
+                //   its four nodes are two `Stream/*` legs plus a null-sink-like
+                //   virtual sink/source, none of which claim a `device.id`, so
+                //   the phase-3 observer never *withholds* them
+                //   (`observer::classify` withholds only device-claiming nodes).
+                //   `index_present` therefore goes false only on a genuine
+                //   `global_remove` of every leg — a real unload — and a real
+                //   unload *should* revoke.
+                // - Worse, gating on readiness would REOPEN the reused-index
+                //   alias trap: if an unload+reload (indices recycle, §5.2
+                //   correction 3) both complete inside one not-ready churn
+                //   window, the ready snapshot would already show the *new*
+                //   module's node and we would never observe the empty gap —
+                //   silently aliasing onto an unrelated module. Revoking the
+                //   instant the gap appears, ready or not, is what closes it.
+                //
+                // This correctness rests on the phase-5/6 integration contract:
+                // **one `observe` per graph event, no coalescing across a module
+                // lifetime boundary.** Under coalescing, the empty gap between an
+                // old unload and a reused-index reload can be skipped. The
+                // robust fix that would not depend on that contract is a
+                // serial-continuity / observer-generation signal (the AEC nodes'
+                // `object.serial`s are fresh across a reload even when the index
+                // is not) — owed to a later hardening round, not built here.
+                if !self.index_present(snapshot, target) {
+                    self.state = AecState::Revoked;
+                }
+            }
+            AecState::NotConfigured | AecState::Failed | AecState::Revoked => {}
+        }
+    }
+
+    /// Whether any node in the snapshot bears the target module index. The same
+    /// exact-`u64`-equality predicate the taint engine roots on
+    /// (`taint/mod.rs`), kept here so "is the identity live?" has one
+    /// definition.
+    fn index_present(&self, snapshot: &GraphSnapshot, target: u64) -> bool {
+        snapshot
+            .nodes()
+            .any(|node| node.props.pulse_module_id == Some(target))
+    }
+}

src/host/aec/tests.rs

@@ -0,0 +1,374 @@
+//! Phase 4 exit gate (impl plan §4): a fake-clock / event-sequence transition
+//! matrix, because these are timing semantics a live poke cannot cover.
+
+use super::*;
+use crate::host::taint::snapshot::{
+    GlobalId, GraphSnapshot, MediaRole, NodeProps, NodeSnapshot, Serial,
+};
+
+/// A `Stream/Output/Audio` node carrying `pulse.module.id == module` (or none).
+/// Only the fields the validator reads matter; the rest take their defaults.
+fn node(serial: u64, module: Option<u64>) -> NodeSnapshot {
+    NodeSnapshot {
+        serial: Serial(serial),
+        id: GlobalId(serial as u32),
+        name: None,
+        role: MediaRole::StreamOutput,
+        props: NodeProps {
+            pulse_module_id: module,
+            ..NodeProps::default()
+        },
+    }
+}
+
+/// A snapshot holding exactly the given nodes (no ports/links/clients — the
+/// validator reads only nodes).
+fn snapshot(nodes: Vec<NodeSnapshot>) -> GraphSnapshot {
+    GraphSnapshot::new(nodes, vec![], vec![], vec![])
+}
+
+fn empty() -> GraphSnapshot {
+    snapshot(vec![])
+}
+
+const IDX: u64 = 536_870_919; // 0x20000007 — a real pipewire-pulse module index.
+const TIMEOUT: Millis = 2_000;
+
+// ---------------------------------------------------------------------------
+// Parsing (D5): off / pulse-module:<idx> / > u32::MAX / absent / malformed.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn parses_off() {
+    assert_eq!(parse_aec_arg("off"), Ok(AecConfig::Off));
+}
+
+#[test]
+fn parses_pulse_module_index() {
+    assert_eq!(
+        parse_aec_arg("pulse-module:536870919"),
+        Ok(AecConfig::PulseModule(536_870_919)),
+    );
+}
+
+#[test]
+fn parses_index_beyond_u32() {
+    // v3.4 §5.2: compare as u64, never u32. A value one past u32::MAX must
+    // round-trip, not truncate or reject.
+    let big = u64::from(u32::MAX) + 1;
+    assert_eq!(
+        parse_aec_arg(&format!("pulse-module:{big}")),
+        Ok(AecConfig::PulseModule(big)),
+    );
+    assert_eq!(
+        parse_aec_arg(&format!("pulse-module:{}", u64::MAX)),
+        Ok(AecConfig::PulseModule(u64::MAX)),
+    );
+}
+
+#[test]
+fn rejects_empty() {
+    assert_eq!(parse_aec_arg(""), Err(AecParseError::Empty));
+}
+
+#[test]
+fn rejects_unknown_form() {
+    assert_eq!(parse_aec_arg("on"), Err(AecParseError::UnknownForm));
+    assert_eq!(parse_aec_arg("module:5"), Err(AecParseError::UnknownForm));
+    assert_eq!(parse_aec_arg("536870919"), Err(AecParseError::UnknownForm));
+}
+
+#[test]
+fn rejects_missing_index() {
+    assert_eq!(
+        parse_aec_arg("pulse-module:"),
+        Err(AecParseError::MissingIndex),
+    );
+}
+
+#[test]
+fn rejects_malformed_index() {
+    for bad in [
+        "pulse-module:-1",                   // sign
+        "pulse-module:+7",                   // sign
+        "pulse-module: 7",                   // leading whitespace
+        "pulse-module:7 ",                   // trailing whitespace
+        "pulse-module:0x7",                  // hex
+        "pulse-module:7.0",                  // non-integer
+        "pulse-module:abc",                  // non-numeric
+        "pulse-module:18446744073709551616", // u64::MAX + 1 (overflow)
+    ] {
+        assert_eq!(
+            parse_aec_arg(bad),
+            Err(AecParseError::InvalidIndex),
+            "{bad} should be InvalidIndex",
+        );
+    }
+}
+
+// ---------------------------------------------------------------------------
+// NotConfigured (--aec=off): benign, terminal, fan-out with no exclusion.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn off_is_not_configured_and_permits_fan_out_with_no_identity() {
+    let mut v = AecValidator::new(AecConfig::Off, TIMEOUT);
+    assert_eq!(v.state(), AecState::NotConfigured);
+    assert!(v.fan_out_permitted());
+    assert_eq!(v.validated_module_id(), None);
+
+    // Even a snapshot full of module nodes never moves it off NotConfigured.
+    v.observe(&snapshot(vec![node(1, Some(IDX))]), true, 10_000);
+    assert_eq!(v.state(), AecState::NotConfigured);
+    assert!(v.fan_out_permitted());
+    assert_eq!(v.validated_module_id(), None);
+}
+
+// ---------------------------------------------------------------------------
+// Row: Validating → Validated on first matching node; no fan-out before.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn validating_forbids_fan_out_and_exposes_no_identity() {
+    let v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    assert_eq!(v.state(), AecState::Validating);
+    assert!(!v.fan_out_permitted());
+    assert_eq!(v.validated_module_id(), None);
+}
+
+#[test]
+fn validating_to_validated_on_first_matching_node() {
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    // A node with a *different* index does not validate.
+    v.observe(&snapshot(vec![node(1, Some(IDX + 1))]), true, 0);
+    assert_eq!(v.state(), AecState::Validating);
+
+    v.observe(&snapshot(vec![node(2, Some(IDX))]), true, 100);
+    assert_eq!(v.state(), AecState::Validated);
+    assert!(v.fan_out_permitted());
+    assert_eq!(v.validated_module_id(), Some(IDX));
+}
+
+#[test]
+fn positive_evidence_validates_even_before_graph_ready() {
+    // Seeing the node is authoritative; readiness only gates the Failed clock.
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    v.observe(&snapshot(vec![node(1, Some(IDX))]), false, 0);
+    assert_eq!(v.state(), AecState::Validated);
+    assert_eq!(v.validated_module_id(), Some(IDX));
+}
+
+#[test]
+fn validated_index_is_compared_beyond_u32() {
+    let big = u64::from(u32::MAX) + 7;
+    let mut v = AecValidator::new(AecConfig::PulseModule(big), TIMEOUT);
+    // A node whose id equals `big` only in its low 32 bits must not match.
+    v.observe(
+        &snapshot(vec![node(1, Some(big & u64::from(u32::MAX)))]),
+        true,
+        0,
+    );
+    assert_eq!(v.state(), AecState::Validating);
+
+    v.observe(&snapshot(vec![node(2, Some(big))]), true, 1);
+    assert_eq!(v.state(), AecState::Validated);
+    assert_eq!(v.validated_module_id(), Some(big));
+}
+
+// ---------------------------------------------------------------------------
+// Row: Validating → Failed on deadline expiry; and the deadline is armed only
+// once the graph is ready (the registry sync barrier).
+// ---------------------------------------------------------------------------
+
+#[test]
+fn validating_to_failed_on_deadline_expiry() {
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    v.observe(&empty(), true, 0); // arms deadline at 0 + 2000
+    assert_eq!(v.state(), AecState::Validating);
+
+    v.observe(&empty(), true, TIMEOUT - 1);
+    assert_eq!(v.state(), AecState::Validating);
+
+    v.observe(&empty(), true, TIMEOUT); // now >= deadline
+    assert_eq!(v.state(), AecState::Failed);
+    assert!(!v.fan_out_permitted());
+    assert_eq!(v.validated_module_id(), None);
+}
+
+#[test]
+fn deadline_is_not_armed_until_graph_ready() {
+    // The whole point of arming-on-ready: a slow initial enumeration is
+    // "unknown", not "absent", and must never time out to Failed.
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    // Long past the would-be deadline, but the graph has never been ready.
+    v.observe(&empty(), false, 10 * TIMEOUT);
+    assert_eq!(v.state(), AecState::Validating);
+    // Still no Failed even much later, as long as ready stays false.
+    v.observe(&empty(), false, 100 * TIMEOUT);
+    assert_eq!(v.state(), AecState::Validating);
+
+    // And when readiness finally arrives, the FULL budget starts *there*, not
+    // relative to construction (Codex phase-4 review, finding 3): a mutant that
+    // armed a construction-relative deadline would fail immediately here.
+    let late = 200_000;
+    v.observe(&empty(), true, late); // first ready → arm at `late`
+    assert_eq!(v.state(), AecState::Validating);
+    v.observe(&empty(), true, late + TIMEOUT - 1);
+    assert_eq!(v.state(), AecState::Validating);
+    v.observe(&empty(), true, late + TIMEOUT);
+    assert_eq!(v.state(), AecState::Failed);
+}
+
+#[test]
+fn late_positive_evidence_wins_over_expired_deadline() {
+    // A node first seen just past the deadline still validates: the deadline
+    // only bounds the wait for an identity that is never seen, and a
+    // demonstrably-present module is ground truth (Codex phase-4 review,
+    // finding 2). Reachable only when the first post-deadline observation
+    // carries the node with no intervening Tick-only observation.
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    v.observe(&empty(), true, 0); // arm deadline at 2000
+    v.observe(&snapshot(vec![node(1, Some(IDX))]), true, TIMEOUT + 1);
+    assert_eq!(v.state(), AecState::Validated);
+    assert_eq!(v.validated_module_id(), Some(IDX));
+
+    // Whereas a Tick-only observation that crosses the deadline first pre-empts
+    // it to Failed (stickily), even if the node then shows up — fail-closed.
+    let mut w = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    w.observe(&empty(), true, 0);
+    w.observe(&empty(), true, TIMEOUT); // Tick-only crosses the line first
+    assert_eq!(w.state(), AecState::Failed);
+    w.observe(&snapshot(vec![node(1, Some(IDX))]), true, TIMEOUT + 1);
+    assert_eq!(w.state(), AecState::Failed);
+}
+
+#[test]
+fn revokes_on_empty_even_while_not_ready() {
+    // Revocation is deliberately NOT gated on graph_ready (Codex phase-4 review,
+    // findings 1 + 4): the instant every node bearing the index is gone we
+    // revoke, ready or not, because gating on readiness would let an
+    // unload+reload that reused the index inside one not-ready churn window
+    // silently alias onto the new module. A mutant adding `&& graph_ready` to
+    // the revoke guard survives every other test but dies here.
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    v.observe(&snapshot(vec![node(1, Some(IDX))]), true, 0);
+    assert_eq!(v.state(), AecState::Validated);
+    v.observe(&empty(), false, 10); // identity gone during not-ready churn
+    assert_eq!(v.state(), AecState::Revoked);
+    assert!(!v.fan_out_permitted());
+}
+
+#[test]
+fn deadline_armed_once_survives_ready_flapping() {
+    // graph_ready is dynamic (it drops back to false while a Link is binding).
+    // The epoch budget must be armed on the *first* ready and not restarted.
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    v.observe(&empty(), true, 1_000); // arm at 1000 → deadline 3000
+    v.observe(&empty(), false, 2_000); // ready flaps off; must not disarm
+    assert_eq!(v.state(), AecState::Validating);
+    // At the original deadline it fails, even though ready is false now — the
+    // budget did not restart from the flap.
+    v.observe(&empty(), false, 3_000);
+    assert_eq!(v.state(), AecState::Failed);
+}
+
+#[test]
+fn failed_is_sticky_even_if_the_index_reappears() {
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    v.observe(&empty(), true, 0);
+    v.observe(&empty(), true, TIMEOUT);
+    assert_eq!(v.state(), AecState::Failed);
+
+    // A node bearing the index shows up late — must not resurrect the epoch.
+    v.observe(&snapshot(vec![node(1, Some(IDX))]), true, TIMEOUT + 1);
+    assert_eq!(v.state(), AecState::Failed);
+    assert!(!v.fan_out_permitted());
+    assert_eq!(v.validated_module_id(), None);
+}
+
+// ---------------------------------------------------------------------------
+// Row: partial-node disappearance ⇒ stays Validated; all gone ⇒ Revoked.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn partial_leg_disappearance_stays_validated() {
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    // The module's four nodes all carry the index.
+    let four = snapshot(vec![
+        node(1, Some(IDX)),
+        node(2, Some(IDX)),
+        node(3, Some(IDX)),
+        node(4, Some(IDX)),
+    ]);
+    v.observe(&four, true, 0);
+    assert_eq!(v.state(), AecState::Validated);
+
+    // Three legs cork/relink away; one still bears the index → still Validated.
+    v.observe(&snapshot(vec![node(4, Some(IDX))]), true, 10);
+    assert_eq!(v.state(), AecState::Validated);
+    assert_eq!(v.validated_module_id(), Some(IDX));
+}
+
+#[test]
+fn all_nodes_gone_revokes() {
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    v.observe(&snapshot(vec![node(1, Some(IDX))]), true, 0);
+    assert_eq!(v.state(), AecState::Validated);
+
+    // The whole identity unloads: no node bears the index any more.
+    v.observe(&empty(), true, 10);
+    assert_eq!(v.state(), AecState::Revoked);
+}
+
+#[test]
+fn revoked_stops_fan_out_and_exposes_no_identity() {
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    v.observe(&snapshot(vec![node(1, Some(IDX))]), true, 0);
+    v.observe(&empty(), true, 10);
+    assert_eq!(v.state(), AecState::Revoked);
+    assert!(!v.fan_out_permitted());
+    assert_eq!(v.validated_module_id(), None);
+}
+
+#[test]
+fn a_node_that_merely_changes_index_revokes() {
+    // Not a disappearance in the id sense, but the *identity* is gone: no node
+    // bears our index any more, even though a same-serial node lingers.
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    v.observe(&snapshot(vec![node(1, Some(IDX))]), true, 0);
+    v.observe(&snapshot(vec![node(1, Some(IDX + 1))]), true, 10);
+    assert_eq!(v.state(), AecState::Revoked);
+}
+
+// ---------------------------------------------------------------------------
+// Row: a retained stale index does not alias onto a reloaded module — indices
+// ARE reused (v3.4 §5.2 correction 3). This is the sharpest safety property.
+// ---------------------------------------------------------------------------
+
+#[test]
+fn revoked_index_does_not_alias_onto_a_reloaded_module() {
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    v.observe(&snapshot(vec![node(1, Some(IDX))]), true, 0);
+    v.observe(&empty(), true, 10);
+    assert_eq!(v.state(), AecState::Revoked);
+
+    // A *different* module later reloads and pactl hands it the very same
+    // index (measured: 536870919 came back verbatim). A resurrecting machine
+    // would silently start excluding this unrelated module's node. Ours must
+    // stay Revoked and fail closed; a real reload gets a fresh validator.
+    v.observe(&snapshot(vec![node(99, Some(IDX))]), true, 20);
+    assert_eq!(v.state(), AecState::Revoked);
+    assert!(!v.fan_out_permitted());
+    assert_eq!(v.validated_module_id(), None);
+}
+
+#[test]
+fn a_fresh_validator_re_validates_the_reused_index() {
+    // The counterpart: because peerspeak re-tells pixelpass the index on every
+    // load, the correct response to a reload is a new machine, which validates
+    // the reused index cleanly — proving stickiness costs nothing legitimate.
+    let mut v = AecValidator::new(AecConfig::PulseModule(IDX), TIMEOUT);
+    v.observe(&snapshot(vec![node(1, Some(IDX))]), true, 0);
+    assert_eq!(v.state(), AecState::Validated);
+    assert_eq!(v.validated_module_id(), Some(IDX));
+}

src/host/audio.rs

@@ -615,7 +615,7 @@ fn run_router(
 /// (empty, signed, whitespace-padded, non-numeric, overflowing) is a
 /// property we do not understand and must not guess at. Leading zeroes
 /// are accepted — they are unambiguous and parse to the same value.
-fn parse_object_serial(raw: &str) -> Option<u64> {
+pub(crate) fn parse_object_serial(raw: &str) -> Option<u64> {
     if raw.is_empty() || !raw.bytes().all(|b| b.is_ascii_digit()) {
         return None;
     }

src/host/mod.rs

@@ -1,5 +1,7 @@
+pub mod aec;
 pub mod audio;
 mod capture;
+mod observer;
 mod pipeline;
 mod quality;
 mod serve;

src/host/observer/adapter.rs

@@ -0,0 +1,601 @@
+//! PipeWire I/O adapter for the pure registry observer.
+//!
+//! This module owns a read-only PipeWire main-loop thread, translates registry
+//! callbacks into [`RegEvent`]s, and publishes the latest [`Projection`] for
+//! consumers running outside the PipeWire thread.
+
+use super::classify::DeviceClaim;
+use super::{LinkEndpoints, NodeObservation, Projection, RegEvent, RegistryModel};
+use crate::host::audio::parse_object_serial;
+use crate::host::taint::snapshot::{
+    ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial,
+};
+use anyhow::{Context, Result};
+use pipewire::{self as pw, types::ObjectType};
+use std::cell::{Cell, RefCell};
+use std::collections::{BTreeMap, VecDeque};
+use std::rc::Rc;
+use std::sync::{Arc, Mutex};
+use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
+
+const READINESS_TIMEOUT_MILLIS: u64 = 2_000;
+const TICK_INTERVAL: Duration = Duration::from_millis(250);
+
+/// Tokio-side access to the observer's most recent coherent projection.
+pub struct RegistryObserverHandle {
+    latest: Arc<Mutex<Option<Projection>>>,
+    shutdown_tx: pw::channel::Sender<()>,
+    thread: Option<JoinHandle<()>>,
+}
+
+impl RegistryObserverHandle {
+    /// Spawn the read-only PipeWire registry observer.
+    pub fn spawn() -> Result<Self> {
+        let latest = Arc::new(Mutex::new(None));
+        let latest_for_thread = Arc::clone(&latest);
+        let (shutdown_tx, shutdown_rx) = pw::channel::channel::<()>();
+        let thread = std::thread::Builder::new()
+            .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:#}"
+                    );
+                }
+            })
+            .context("failed to spawn libpipewire registry observer thread")?;
+
+        Ok(Self {
+            latest,
+            shutdown_tx,
+            thread: Some(thread),
+        })
+    }
+
+    /// Return a clone of the latest projection, or `None` before the first
+    /// registry event has been applied.
+    pub fn latest(&self) -> Option<Projection> {
+        self.latest
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .clone()
+    }
+}
+
+impl Drop for RegistryObserverHandle {
+    fn drop(&mut self) {
+        let _ = self.shutdown_tx.send(());
+        if let Some(thread) = self.thread.take()
+            && let Err(e) = thread.join()
+        {
+            tracing::warn!("registry observer: pw thread join failed: {e:?}");
+        }
+    }
+}
+
+struct BoundLink {
+    _proxy: pw::link::Link,
+    _listener: pw::link::LinkListener,
+}
+
+#[derive(Default)]
+struct LiveGlobal {
+    bound_link: Option<BoundLink>,
+}
+
+struct ObserverState {
+    model: RegistryModel,
+    latest: Arc<Mutex<Option<Projection>>>,
+    last_candidate: Option<u32>,
+    live_globals: BTreeMap<GlobalId, VecDeque<LiveGlobal>>,
+}
+
+impl ObserverState {
+    fn new(latest: Arc<Mutex<Option<Projection>>>) -> Self {
+        Self {
+            model: RegistryModel::new(0, READINESS_TIMEOUT_MILLIS),
+            latest,
+            last_candidate: None,
+            live_globals: BTreeMap::new(),
+        }
+    }
+
+    fn apply(&mut self, event: RegEvent) {
+        self.model.apply(event);
+
+        let candidate = self.model.pulse_pid_candidate();
+        if candidate != self.last_candidate {
+            self.last_candidate = candidate;
+            if let Some(pid) = candidate {
+                let comm = std::fs::read_to_string(format!("/proc/{pid}/comm"))
+                    .ok()
+                    .map(|comm| comm.trim_end_matches(['\r', '\n']).to_string());
+                self.model.apply(RegEvent::ProcCommProbed { pid, comm });
+            }
+        }
+
+        self.publish();
+    }
+
+    fn publish(&self) {
+        *self
+            .latest
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(self.model.project());
+    }
+
+    /// Record the global's id and apply its add event as one step, so the
+    /// bound-link FIFO stays provably lockstep with the model's own `live_ids`
+    /// index. Recording only on *applied* adds (never on unknown object types
+    /// or globals dropped for a missing serial) is what keeps the two id
+    /// 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.apply(event);
+    }
+
+    fn attach_bound_link(&mut self, id: GlobalId, bound_link: BoundLink) {
+        let Some(global) = self.live_globals.get_mut(&id).and_then(VecDeque::back_mut) else {
+            tracing::warn!(
+                global_id = id.0,
+                "registry observer: link bind completed without a live global slot"
+            );
+            return;
+        };
+        global.bound_link = Some(bound_link);
+    }
+
+    fn remove_global(&mut self, id: GlobalId) -> Option<BoundLink> {
+        let (bound_link, empty) = {
+            let globals = self.live_globals.get_mut(&id)?;
+            let bound_link = globals.pop_front().and_then(|global| global.bound_link);
+            (bound_link, globals.is_empty())
+        };
+        if empty {
+            self.live_globals.remove(&id);
+        }
+        bound_link
+    }
+}
+
+fn run_observer(
+    latest: Arc<Mutex<Option<Projection>>>,
+    shutdown_rx: pw::channel::Receiver<()>,
+) -> Result<()> {
+    let started_at = Instant::now();
+    let main_loop =
+        pw::main_loop::MainLoopRc::new(None).context("pw main loop construction failed")?;
+    let context =
+        pw::context::ContextRc::new(&main_loop, None).context("pw context construction failed")?;
+    let core = context
+        .connect_rc(None)
+        .context("pw core connect failed (is the daemon running?)")?;
+    let registry = core.get_registry_rc().context("pw get_registry failed")?;
+    let state = Rc::new(RefCell::new(ObserverState::new(latest)));
+
+    let main_loop_for_shutdown = main_loop.clone();
+    let _shutdown_receiver = shutdown_rx.attach(main_loop.loop_(), move |()| {
+        main_loop_for_shutdown.quit();
+    });
+
+    let pending_sync = Rc::new(Cell::new(None));
+    let pending_sync_for_done = Rc::clone(&pending_sync);
+    let state_for_done = Rc::clone(&state);
+    let _core_listener = core
+        .add_listener_local()
+        .done(move |id, seq| {
+            if id == pw::core::PW_ID_CORE && pending_sync_for_done.get() == Some(seq) {
+                pending_sync_for_done.set(None);
+                state_for_done.borrow_mut().apply(RegEvent::ServerSynced);
+            }
+        })
+        .error(|id, seq, res, message| {
+            tracing::warn!(
+                id,
+                seq,
+                result = res,
+                %message,
+                "registry observer: PipeWire core error"
+            );
+        })
+        .register();
+
+    let registry_weak = registry.downgrade();
+    let state_for_global = Rc::clone(&state);
+    let state_for_remove = Rc::clone(&state);
+    let _registry_listener = registry
+        .add_listener_local()
+        .global(move |obj| {
+            let id = GlobalId(obj.id);
+
+            match obj.type_ {
+                ObjectType::Node => {
+                    let Some(props) = obj.props.as_ref() else {
+                        tracing::warn!(
+                            node_id = obj.id,
+                            "registry observer: Node has no properties; dropping"
+                        );
+                        return;
+                    };
+                    let Some(serial) = parse_serial(obj.id, "Node", props.get("object.serial"))
+                    else {
+                        return;
+                    };
+                    let node_props = NodeProps {
+                        peerspeak_owned: truthy(props.get("peerspeak.owned")),
+                        pulse_module_id: props
+                            .get("pulse.module.id")
+                            .and_then(|value| value.parse::<u64>().ok()),
+                        link_group: props.get("node.link-group").map(str::to_owned),
+                        client_id: props
+                            .get("client.id")
+                            .and_then(|value| value.parse::<u32>().ok())
+                            .map(GlobalId),
+                        process_id: props
+                            .get("application.process.id")
+                            .and_then(|value| value.parse::<u32>().ok()),
+                        passthrough: truthy(props.get("node.passthrough")),
+                        session_device: false,
+                    };
+                    let observation = NodeObservation {
+                        serial,
+                        id,
+                        name: props.get("node.name").map(str::to_owned),
+                        role: MediaRole::parse(props.get("media.class")),
+                        props: node_props,
+                        device_claim: DeviceClaim {
+                            device_id: props
+                                .get("device.id")
+                                .and_then(|value| value.parse::<u32>().ok())
+                                .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
+                        .borrow_mut()
+                        .add(id, RegEvent::NodeAdded(observation));
+                }
+                ObjectType::Port => {
+                    let Some(props) = obj.props.as_ref() else {
+                        tracing::warn!(
+                            port_id = obj.id,
+                            "registry observer: Port has no properties; dropping"
+                        );
+                        return;
+                    };
+                    let Some(serial) = parse_serial(obj.id, "Port", props.get("object.serial"))
+                    else {
+                        return;
+                    };
+                    let Some(node) = props
+                        .get("node.id")
+                        .and_then(|value| value.parse::<u32>().ok())
+                        .map(GlobalId)
+                    else {
+                        tracing::warn!(
+                            port_id = obj.id,
+                            node_id = props.get("node.id").unwrap_or("<absent>"),
+                            "registry observer: Port has no usable node.id; dropping"
+                        );
+                        return;
+                    };
+                    let direction = match props.get("port.direction") {
+                        Some("in") => PortDirection::In,
+                        Some("out") => PortDirection::Out,
+                        direction => {
+                            tracing::warn!(
+                                port_id = obj.id,
+                                direction = direction.unwrap_or("<absent>"),
+                                "registry observer: Port has no usable direction; dropping"
+                            );
+                            return;
+                        }
+                    };
+                    state_for_global.borrow_mut().add(
+                        id,
+                        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 {
+                        tracing::warn!(
+                            client_id = obj.id,
+                            "registry observer: Client has no properties; dropping"
+                        );
+                        return;
+                    };
+                    let Some(serial) = parse_serial(obj.id, "Client", props.get("object.serial"))
+                    else {
+                        return;
+                    };
+                    state_for_global.borrow_mut().add(
+                        id,
+                        RegEvent::ClientAdded(ClientSnapshot {
+                            serial,
+                            id,
+                            sec_pid: props
+                                .get("pipewire.sec.pid")
+                                .and_then(|value| value.parse::<u32>().ok()),
+                        }),
+                    );
+                }
+                ObjectType::Device => {
+                    state_for_global
+                        .borrow_mut()
+                        .add(id, RegEvent::DeviceAdded { id });
+                }
+                ObjectType::Link => {
+                    let Some(props) = obj.props.as_ref() else {
+                        tracing::warn!(
+                            link_id = obj.id,
+                            "registry observer: Link has no properties; dropping"
+                        );
+                        return;
+                    };
+                    let Some(serial) = parse_serial(obj.id, "Link", props.get("object.serial"))
+                    else {
+                        return;
+                    };
+                    let endpoints = link_endpoints_from_props(props);
+                    state_for_global.borrow_mut().add(
+                        id,
+                        RegEvent::LinkAdded {
+                            serial,
+                            id,
+                            endpoints,
+                        },
+                    );
+                    if endpoints.is_some() {
+                        return;
+                    }
+
+                    let Some(registry) = registry_weak.upgrade() else {
+                        return;
+                    };
+                    let link: pw::link::Link = match registry.bind(obj) {
+                        Ok(link) => link,
+                        Err(e) => {
+                            tracing::warn!(
+                                link_id = obj.id,
+                                "registry observer: failed to bind Link for endpoints: {e}"
+                            );
+                            return;
+                        }
+                    };
+                    let resolved = Rc::new(Cell::new(false));
+                    let resolved_for_info = Rc::clone(&resolved);
+                    let state_for_info = Rc::downgrade(&state_for_global);
+                    let listener = link
+                        .add_listener_local()
+                        .info(move |info| {
+                            if resolved_for_info.replace(true) {
+                                return;
+                            }
+                            let endpoints = LinkEndpoints {
+                                output_node: GlobalId(info.output_node_id()),
+                                input_node: GlobalId(info.input_node_id()),
+                                output_port: optional_global_id(info.output_port_id()),
+                                input_port: optional_global_id(info.input_port_id()),
+                            };
+                            if let Some(state) = state_for_info.upgrade() {
+                                state
+                                    .borrow_mut()
+                                    .apply(RegEvent::LinkEndpointsResolved { serial, endpoints });
+                            }
+                        })
+                        .register();
+                    state_for_global.borrow_mut().attach_bound_link(
+                        id,
+                        BoundLink {
+                            _proxy: link,
+                            _listener: listener,
+                        },
+                    );
+                }
+                _ => {}
+            }
+        })
+        .global_remove(move |id| {
+            let id = GlobalId(id);
+            let bound_link = state_for_remove.borrow_mut().remove_global(id);
+            state_for_remove
+                .borrow_mut()
+                .apply(RegEvent::Removed { id });
+            drop(bound_link);
+        })
+        .register();
+
+    pending_sync.set(Some(
+        core.sync(0)
+            .context("registry observer: initial core.sync failed")?,
+    ));
+
+    let state_for_tick = Rc::clone(&state);
+    let timer = main_loop.loop_().add_timer(move |_| {
+        let now = u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
+        state_for_tick.borrow_mut().apply(RegEvent::Tick { now });
+    });
+    timer
+        .update_timer(Some(TICK_INTERVAL), Some(TICK_INTERVAL))
+        .into_result()
+        .context("registry observer: failed to arm readiness timer")?;
+
+    tracing::info!("registry observer: pw thread running");
+    main_loop.run();
+    tracing::info!("registry observer: pw thread exiting");
+    Ok(())
+}
+
+fn parse_serial(id: u32, kind: &str, raw: Option<&str>) -> Option<Serial> {
+    match raw.and_then(parse_object_serial) {
+        Some(serial) => Some(Serial(serial)),
+        None => {
+            tracing::warn!(
+                global_id = id,
+                object_type = kind,
+                serial = raw.unwrap_or("<absent>"),
+                "registry observer: global has no usable object.serial; dropping"
+            );
+            None
+        }
+    }
+}
+
+fn truthy(value: Option<&str>) -> bool {
+    value.is_some_and(|value| value != "false" && value != "0")
+}
+
+fn link_endpoints_from_props(props: &pw::spa::utils::dict::DictRef) -> Option<LinkEndpoints> {
+    let output_node = props.get("link.output.node")?.parse::<u32>().ok()?;
+    let input_node = props.get("link.input.node")?.parse::<u32>().ok()?;
+    Some(LinkEndpoints {
+        output_node: GlobalId(output_node),
+        input_node: GlobalId(input_node),
+        output_port: props
+            .get("link.output.port")
+            .and_then(|value| value.parse::<u32>().ok())
+            .map(GlobalId),
+        input_port: props
+            .get("link.input.port")
+            .and_then(|value| value.parse::<u32>().ok())
+            .map(GlobalId),
+    })
+}
+
+fn optional_global_id(id: u32) -> Option<GlobalId> {
+    (id != pw::constants::ID_ANY).then_some(GlobalId(id))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::process::Command;
+
+    struct PactlModule {
+        id: Option<u32>,
+    }
+
+    impl PactlModule {
+        fn load(name: &str, args: &[String]) -> Self {
+            let output = Command::new("pactl")
+                .arg("load-module")
+                .arg(name)
+                .args(args)
+                .output()
+                .expect("pactl must be installed for the live observer test");
+            assert!(
+                output.status.success(),
+                "pactl load-module {name} failed: {}",
+                String::from_utf8_lossy(&output.stderr).trim()
+            );
+            let id = String::from_utf8(output.stdout)
+                .expect("pactl module id must be UTF-8")
+                .trim()
+                .parse::<u32>()
+                .expect("pactl module id must be a u32");
+            Self { id: Some(id) }
+        }
+
+        fn unload(mut self) {
+            let id = self.id.take().expect("module must still be loaded");
+            let output = Command::new("pactl")
+                .arg("unload-module")
+                .arg(id.to_string())
+                .output()
+                .expect("pactl must be installed for the live observer test");
+            assert!(
+                output.status.success(),
+                "pactl unload-module {id} failed: {}",
+                String::from_utf8_lossy(&output.stderr).trim()
+            );
+        }
+    }
+
+    impl Drop for PactlModule {
+        fn drop(&mut self) {
+            if let Some(id) = self.id.take() {
+                let _ = Command::new("pactl")
+                    .arg("unload-module")
+                    .arg(id.to_string())
+                    .output();
+            }
+        }
+    }
+
+    fn wait_for(
+        observer: &RegistryObserverHandle,
+        predicate: impl Fn(&Projection) -> bool,
+    ) -> Projection {
+        let deadline = Instant::now() + Duration::from_secs(10);
+        while Instant::now() < deadline {
+            if let Some(projection) = observer.latest()
+                && predicate(&projection)
+            {
+                return projection;
+            }
+            std::thread::sleep(Duration::from_millis(50));
+        }
+        panic!("timed out waiting for the registry projection");
+    }
+
+    fn has_node(projection: &Projection, name: &str) -> bool {
+        projection
+            .snapshot
+            .nodes()
+            .any(|node| node.name.as_deref() == Some(name))
+    }
+
+    #[test]
+    #[ignore = "needs live pipewire"]
+    fn live_topology_diff_tracks_null_sink_and_loopback() {
+        pw::init();
+        let observer = RegistryObserverHandle::spawn().expect("observer thread must spawn");
+        let baseline = wait_for(&observer, |projection| projection.graph_ready);
+        let baseline_links = baseline.snapshot.links().count();
+
+        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("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();
+
+        let loopback = PactlModule::load(
+            "module-loopback",
+            &[
+                format!("source={unique}.monitor"),
+                format!("sink={unique}"),
+                format!("source_output_properties=node.name={capture_name}"),
+                format!("sink_input_properties=node.name={playback_name}"),
+            ],
+        );
+        wait_for(&observer, |projection| {
+            has_node(projection, &capture_name)
+                && has_node(projection, &playback_name)
+                && projection.snapshot.links().count() > sink_links
+        });
+
+        loopback.unload();
+        null_sink.unload();
+        wait_for(&observer, |projection| {
+            !has_node(projection, &unique)
+                && !has_node(projection, &capture_name)
+                && !has_node(projection, &playback_name)
+                && projection.snapshot.links().count() <= baseline_links
+        });
+    }
+}

src/host/observer/classify.rs

@@ -0,0 +1,161 @@
+//! The `session_device` classifier — pure, no PipeWire.
+//!
+//! `NodeProps::session_device` (see [`super::super::taint::snapshot`]) is a
+//! **positive high-confidence** claim that a node is a passive hardware
+//! terminal: a real sound card's sink or source that terminates audio rather
+//! than forwarding it. Setting it *removes* two protections at once — the
+//! node's coarse owner keys and its ability to trip the fail-closed backstop
+//! — so a false positive is a **leak**, and the whole classifier is shaped so
+//! that anything less than a positive identification resolves to `false`.
+//!
+//! The observer (phase 3) owes this classification; the adapter must never
+//! stuff a raw property through. Two facts from the design (v3.4 §6.1.1,
+//! Codex rounds 2–4) drive the shape here:
+//!
+//! - `device.id` / `device.api` describe *which* Device a node belongs to and
+//!   *how* that Device is reached — **neither promises the node passively
+//!   terminates audio.** A filter chain associated with a card satisfies
+//!   both. So the discriminator is `factory.name` on an **allowlist** of
+//!   real hardware-PCM factories, never a substring or a denylist: an unknown
+//!   factory is not a device.
+//! - The backing Device must actually have been observed. A node that claims
+//!   a `device.id` we have not yet resolved is **withheld**, not admitted with
+//!   a provisional `false` — a provisional `false` during the not-ready
+//!   window fuses sink and mic on the shared session client and that fusion
+//!   can persist as sticky over-exclusion (round-3 finding 3).
+
+use crate::host::taint::snapshot::GlobalId;
+
+/// Factory names that positively identify a passive hardware-PCM terminal.
+///
+/// **An allowlist, deliberately.** Membership *removes* protections, so the
+/// safe error direction is to leave a genuine-but-unlisted device off the
+/// list (it merely keeps its owner keys — over-exclusion, no echo). Adding a
+/// backend here is a security-relevant change and wants the same measurement
+/// the ALSA entries got (snapshot.rs `session_device` contract: the target
+/// box's five ALSA nodes carry `factory.name=api.alsa.pcm.{sink,source}`; the
+/// three `support.null-audio-sink` nodes carry neither).
+///
+/// `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",
+];
+
+/// 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 a real ALSA terminal must present an `alsa.driver_name`
+/// that is **present and not on this denylist**; a missing driver fails closed
+/// (see [`classify`]). `snd_dummy` is intentionally absent: it is virtual but
+/// does not couple playback to capture, so it is not a loopback hazard.
+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
+/// *decision* whose output is the `session_device` field — they are inputs,
+/// not part of the graph the engine reasons over.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct DeviceClaim {
+    /// `device.id` — the Device this node belongs to, if any. Absent on
+    /// `Stream/*` nodes, which is exactly why their absence means "not a
+    /// device", not "unknown".
+    pub device_id: Option<GlobalId>,
+    /// `device.api` — the access API of that Device (e.g. `alsa`, `bluez5`).
+    /// Its mere presence is **not** sufficient (a card-associated filter has
+    /// it too); required only as a corroborating signal alongside the factory
+    /// allowlist.
+    pub device_api: Option<String>,
+    /// `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. `session_device` requires this to be
+    /// **present and not** on [`NON_TERMINAL_ALSA_DRIVERS`]; a driver on the
+    /// denylist, or an absent value, both fail closed (see [`classify`]).
+    /// May be absent on non-ALSA backends or on version pairings that do not
+    /// copy `alsa.*` onto the node.
+    pub alsa_driver_name: Option<String>,
+}
+
+/// The outcome of classifying one node's device claim.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub enum Classification {
+    /// No `device.id` — a `Stream/*` node. Admit with `session_device=false`.
+    NotADevice,
+    /// A `device.id` is claimed but the backing Device has not been resolved
+    /// yet. **Withhold the node and keep the readiness epoch not-ready**;
+    /// re-classify when the Device is observed.
+    Withhold { device_id: GlobalId },
+    /// Positively a passive hardware terminal. Admit with
+    /// `session_device=true`.
+    SessionDevice,
+    /// Backed by a *resolved* Device but not a hardware-PCM terminal — a
+    /// filter or virtual node on a card, an unknown factory, or a Device with
+    /// no `device.api`. Admit with `session_device=false` (fail closed).
+    NotSessionDevice,
+}
+
+/// Classify a node's device claim.
+///
+/// `device_resolved` is whether [`DeviceClaim::device_id`] has been observed
+/// as a Device global; it is only consulted when a `device_id` is present.
+/// Pure: the model supplies `device_resolved` from its resolved-Device set,
+/// and the I/O of *binding* the Device lives in the adapter.
+pub fn classify(claim: &DeviceClaim, device_resolved: bool) -> Classification {
+    let Some(device_id) = claim.device_id else {
+        // No backing Device: a stream. Not withheld, not a device.
+        return Classification::NotADevice;
+    };
+    if !device_resolved {
+        // Backed by a Device we have not seen — the one case that blocks
+        // readiness. A provisional answer here is the leak the contract
+        // forbids.
+        return Classification::Withhold { device_id };
+    }
+    let on_factory_allowlist = claim
+        .factory_name
+        .as_deref()
+        .is_some_and(|f| HARDWARE_PCM_FACTORIES.contains(&f));
+    // A **present, non-denied** ALSA driver is required — absence fails closed
+    // (Codex phase-3 re-review). `alsa.driver_name` is not copied onto the
+    // node on every PipeWire/WirePlumber version pairing (PipeWire ≥1.2.6
+    // stopped overwriting node props with card props; WirePlumber only began
+    // copying `alsa.*` onto nodes in 0.5.13), so a *missing* value must not be
+    // read as "not a loopback" — that is exactly the hole an `snd_aloop` node
+    // without the property would slip through. A real card whose node lacks
+    // the driver is instead over-excluded (keeps its owner keys — safe);
+    // recovering `session_device` for it needs reading the driver from the
+    // backing Device global, which is owed to a later round.
+    let driver_ok = 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 && driver_ok;
+    if is_hardware_pcm {
+        Classification::SessionDevice
+    } else {
+        // Resolved, but not positively a terminal: fail closed to false so
+        // the node keeps its owner keys and its backstop.
+        Classification::NotSessionDevice
+    }
+}

src/host/observer/mod.rs

@@ -0,0 +1,512 @@
+//! The registry observer's **pure core** (impl plan §4, phase 3).
+//!
+//! This is my half of the phase-3 split: a reducer that folds a stream of
+//! typed [`RegEvent`]s into a live model of the PipeWire graph and projects
+//! the [`GraphSnapshot`] + context the taint engine (phase 2) consumes. **No
+//! PipeWire types appear here** — the I/O adapter (Codex's half) translates
+//! live registry callbacks, Link/Device binds, `/proc` reads, and the
+//! `core.sync`/`done` round-trip into these events and feeds them in. Every
+//! test in this module builds the event stream by hand.
+//!
+//! Three things this core is shaped to get right, each an exit-gate row:
+//!
+//! - **Removal by recycled id.** `global_remove` names only a 32-bit global
+//!   id, and those recycle. The model keeps an insertion-ordered index per id
+//!   so a removal accounts for the *oldest* generation first, and the
+//!   snapshot projection treats any id still claimed by two live objects as
+//!   [`IdLookup::Ambiguous`] — fail closed (v3.4 §6.1.3).
+//! - **The readiness epoch.** `graph_ready` is false until the initial graph
+//!   is fully observed: the server has synced **and** no binds/withheld nodes
+//!   remain outstanding. A bounded timeout makes it fail closed. It gates
+//!   sticky *retirement* only; withholding after completion is per-object.
+//! - **Withholding on unresolved devices.** A node claiming a `device.id`
+//!   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:** PipeWire's native protocol defines `object.serial` as the
+//!   unique identity every global carries, so a Link without one requires a
+//!   protocol/server failure, not ordinary churn. (The live gate is
+//!   consistent with this but does not *prove* it — it only counts Links the
+//!   strict parser already admitted.) 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.
+
+pub mod adapter;
+pub mod classify;
+pub mod pulse_pid;
+
+#[cfg(test)]
+mod tests;
+
+use crate::host::taint::snapshot::{
+    ClientSnapshot, GlobalId, GraphSnapshot, LinkSnapshot, MediaRole, NodeProps, NodeSnapshot,
+    PortSnapshot, Serial,
+};
+use classify::{Classification, DeviceClaim};
+use std::collections::{BTreeMap, VecDeque};
+
+/// A monotonic millisecond clock value, supplied by the adapter via
+/// [`RegEvent::Tick`]. Kept as a bare integer rather than
+/// [`std::time::Instant`] so the readiness timeout is deterministic in tests.
+pub type Millis = u64;
+
+/// A Node as observed off the registry, before `session_device` has been
+/// decided. The adapter fills [`NodeProps`] with everything it can parse and
+/// leaves `session_device` at its `false` default; the model overwrites it
+/// from the [`classify`] result once the backing Device (if any) is resolved.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct NodeObservation {
+    pub serial: Serial,
+    pub id: GlobalId,
+    pub name: Option<String>,
+    pub role: MediaRole,
+    pub props: NodeProps,
+    pub device_claim: DeviceClaim,
+}
+
+/// The four endpoint references a Link carries. Node endpoints are required —
+/// a Link with unknown nodes is useless — so this whole struct is what the
+/// adapter must resolve (from the global's props if present, else by binding
+/// `LinkInfoRef`, the correctness path) before a Link enters the snapshot.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct LinkEndpoints {
+    pub output_node: GlobalId,
+    pub input_node: GlobalId,
+    pub output_port: Option<GlobalId>,
+    pub input_port: Option<GlobalId>,
+}
+
+/// A typed observation of the live graph. The adapter produces these; the
+/// model consumes them in [`RegistryModel::apply`].
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum RegEvent {
+    /// A Node global appeared. Admitted immediately unless it claims an
+    /// unresolved Device (then withheld — see [`classify`]).
+    NodeAdded(NodeObservation),
+    /// A Port global appeared.
+    PortAdded(PortSnapshot),
+    /// A Client global appeared. Feeds pulse-PID derivation via `sec_pid`.
+    ClientAdded(ClientSnapshot),
+    /// A Device global appeared. Resolves any nodes withheld on its id.
+    DeviceAdded { id: GlobalId },
+    /// A Link global appeared. `endpoints` is `Some` when the global carried
+    /// them (the optimisation) and `None` when the adapter must bind to learn
+    /// them (the correctness path) — the latter is an outstanding obligation
+    /// until a matching [`RegEvent::LinkEndpointsResolved`] arrives.
+    LinkAdded {
+        serial: Serial,
+        id: GlobalId,
+        endpoints: Option<LinkEndpoints>,
+    },
+    /// The bind-`LinkInfoRef` fallback resolved a Link's endpoints.
+    LinkEndpointsResolved {
+        serial: Serial,
+        endpoints: LinkEndpoints,
+    },
+    /// The adapter read `/proc/<pid>/comm` (`None` = the read failed / the
+    /// process is gone). Validates the pulse-PID candidate.
+    ProcCommProbed { pid: u32, comm: Option<String> },
+    /// Any global was removed. Only its 32-bit id is known.
+    Removed { id: GlobalId },
+    /// A `core.sync()` issued after the initial enumeration completed its
+    /// round-trip (`done`). One half of readiness; the other is that no
+    /// binds/withheld nodes are still outstanding.
+    ServerSynced,
+    /// A monotonic clock sample. Drives the readiness timeout only.
+    Tick { now: Millis },
+}
+
+/// Which slot in the id index a live object occupies. `global_remove` gives
+/// only the id, so the index remembers what each id currently holds. A Node
+/// slot's serial may live in either the admitted or the withheld map.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum Slot {
+    Node(Serial),
+    Port(Serial),
+    Link(Serial),
+    Client(Serial),
+    Device,
+}
+
+/// The readiness epoch. A one-time transition out of [`Readiness::Waiting`];
+/// both terminal states are sticky (a completed graph is not un-completed by
+/// later per-object withholding, and a timed-out observer stays fail-closed
+/// for its lifetime).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Readiness {
+    /// The initial enumeration is still in flight.
+    Waiting,
+    /// 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.
+    TimedOut,
+}
+
+/// The pure handoff to the taint engine: a coherent [`GraphSnapshot`] plus the
+/// two context fields phase 3 owns. The caller merges these into
+/// [`crate::host::taint::ExclusionCtx`] alongside `aec_module_id` (phase 4)
+/// and `pixelpass_owned` (pixelpass's own tracking).
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Projection {
+    pub snapshot: GraphSnapshot,
+    pub pipewire_pulse_pid: Option<u32>,
+    pub graph_ready: bool,
+}
+
+/// The live model. Folds [`RegEvent`]s; project with [`RegistryModel::project`].
+#[derive(Clone, Debug)]
+pub struct RegistryModel {
+    // Admitted objects, keyed by their never-recycled serial.
+    nodes: BTreeMap<Serial, NodeSnapshot>,
+    ports: BTreeMap<Serial, PortSnapshot>,
+    links: BTreeMap<Serial, LinkSnapshot>,
+    clients: BTreeMap<Serial, ClientSnapshot>,
+
+    /// Nodes held out of the snapshot pending their Device's resolution.
+    withheld: BTreeMap<Serial, NodeObservation>,
+    /// Links whose endpoints the adapter is still binding; the id is kept so
+    /// removal and resolution can find them.
+    pending_links: BTreeMap<Serial, GlobalId>,
+
+    /// Live Device global ids, ref-counted so a recycled id is only
+    /// considered resolved while a Device actually holds it.
+    resolved_devices: BTreeMap<GlobalId, usize>,
+
+    /// Insertion-ordered holders of each live global id. `global_remove`
+    /// accounts for the oldest generation first (v3.4 §6.1.3).
+    live_ids: BTreeMap<GlobalId, VecDeque<Slot>>,
+
+    /// `/proc/<pid>/comm` reads keyed by pid, for pulse-PID validation.
+    probed_comm: BTreeMap<u32, Option<String>>,
+
+    server_synced: bool,
+    readiness: Readiness,
+    deadline: Millis,
+    last_now: Millis,
+}
+
+impl RegistryModel {
+    /// `now` seeds the clock; `timeout` is the readiness budget. The deadline
+    /// is `now + timeout`; a [`RegEvent::Tick`] at or past it while still
+    /// [`Readiness::Waiting`] fails the epoch closed.
+    pub fn new(now: Millis, timeout: Millis) -> Self {
+        Self {
+            nodes: BTreeMap::new(),
+            ports: BTreeMap::new(),
+            links: BTreeMap::new(),
+            clients: BTreeMap::new(),
+            withheld: BTreeMap::new(),
+            pending_links: BTreeMap::new(),
+            resolved_devices: BTreeMap::new(),
+            live_ids: BTreeMap::new(),
+            probed_comm: BTreeMap::new(),
+            server_synced: false,
+            readiness: Readiness::Waiting,
+            deadline: now.saturating_add(timeout),
+            last_now: now,
+        }
+    }
+
+    pub fn readiness(&self) -> Readiness {
+        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) && !self.obligations_outstanding()
+    }
+
+    /// The pulse-PID candidate the adapter should be probing (`None` = no
+    /// repeated `sec_pid`, nothing to probe). Exposed so the adapter re-probes
+    /// only when the candidate changes.
+    pub fn pulse_pid_candidate(&self) -> Option<u32> {
+        let clients: Vec<ClientSnapshot> = self.clients.values().cloned().collect();
+        pulse_pid::candidate(&clients)
+    }
+
+    /// Fold one observation into the model.
+    pub fn apply(&mut self, event: RegEvent) {
+        match event {
+            RegEvent::NodeAdded(obs) => self.on_node_added(obs),
+            RegEvent::PortAdded(port) => {
+                self.push_id(port.id, Slot::Port(port.serial));
+                self.ports.insert(port.serial, port);
+            }
+            RegEvent::ClientAdded(client) => {
+                self.push_id(client.id, Slot::Client(client.serial));
+                self.clients.insert(client.serial, client);
+                // A new client can change the pulse candidate; the adapter
+                // learns that via `pulse_pid_candidate`. No readiness effect.
+            }
+            RegEvent::DeviceAdded { id } => self.on_device_added(id),
+            RegEvent::LinkAdded {
+                serial,
+                id,
+                endpoints,
+            } => self.on_link_added(serial, id, endpoints),
+            RegEvent::LinkEndpointsResolved { serial, endpoints } => {
+                self.on_link_resolved(serial, endpoints)
+            }
+            RegEvent::ProcCommProbed { pid, comm } => {
+                self.probed_comm.insert(pid, comm);
+            }
+            RegEvent::Removed { id } => self.on_removed(id),
+            RegEvent::ServerSynced => {
+                self.server_synced = true;
+                self.maybe_complete();
+            }
+            RegEvent::Tick { now } => {
+                self.last_now = now;
+                self.maybe_timeout(now);
+            }
+        }
+    }
+
+    fn on_node_added(&mut self, obs: NodeObservation) {
+        self.push_id(obs.id, Slot::Node(obs.serial));
+        let resolved = obs
+            .device_claim
+            .device_id
+            .is_some_and(|id| self.device_resolved(id));
+        match classify::classify(&obs.device_claim, resolved) {
+            Classification::Withhold { .. } => {
+                self.withheld.insert(obs.serial, obs);
+            }
+            Classification::SessionDevice => self.admit_node(obs, true),
+            Classification::NotADevice | Classification::NotSessionDevice => {
+                self.admit_node(obs, false)
+            }
+        }
+        // Withholding a node adds an obligation; admitting one can never
+        // complete readiness on its own, but re-check is cheap and keeps the
+        // invariant local.
+        self.maybe_complete();
+    }
+
+    fn admit_node(&mut self, obs: NodeObservation, session_device: bool) {
+        let mut props = obs.props;
+        props.session_device = session_device;
+        self.nodes.insert(
+            obs.serial,
+            NodeSnapshot {
+                serial: obs.serial,
+                id: obs.id,
+                name: obs.name,
+                role: obs.role,
+                props,
+            },
+        );
+    }
+
+    fn on_device_added(&mut self, id: GlobalId) {
+        self.push_id(id, Slot::Device);
+        *self.resolved_devices.entry(id).or_insert(0) += 1;
+        // Admit every node that was withheld waiting on exactly this Device.
+        let ready: Vec<Serial> = self
+            .withheld
+            .iter()
+            .filter(|(_, obs)| obs.device_claim.device_id == Some(id))
+            .map(|(&serial, _)| serial)
+            .collect();
+        for serial in ready {
+            if let Some(obs) = self.withheld.remove(&serial) {
+                // Resolved now, so classify yields a terminal answer, never
+                // Withhold again.
+                let session_device = matches!(
+                    classify::classify(&obs.device_claim, true),
+                    Classification::SessionDevice
+                );
+                self.admit_node(obs, session_device);
+            }
+        }
+        self.maybe_complete();
+    }
+
+    fn on_link_added(&mut self, serial: Serial, id: GlobalId, endpoints: Option<LinkEndpoints>) {
+        self.push_id(id, Slot::Link(serial));
+        match endpoints {
+            Some(e) => {
+                self.links.insert(serial, link_snapshot(serial, id, e));
+            }
+            None => {
+                // Correctness path: withhold the Link until the bind fallback
+                // resolves it. Counts as an outstanding obligation.
+                self.pending_links.insert(serial, id);
+            }
+        }
+        self.maybe_complete();
+    }
+
+    fn on_link_resolved(&mut self, serial: Serial, endpoints: LinkEndpoints) {
+        // `remove` also guards against a stale resolution for a Link already
+        // gone: unknown serial ⇒ ignore.
+        if let Some(id) = self.pending_links.remove(&serial) {
+            self.links
+                .insert(serial, link_snapshot(serial, id, endpoints));
+            self.maybe_complete();
+        }
+    }
+
+    fn on_removed(&mut self, id: GlobalId) {
+        let Some(queue) = self.live_ids.get_mut(&id) else {
+            tracing::warn!(global_id = id.0, "observer: remove for an id we never saw");
+            return;
+        };
+        // Oldest generation first — the id may be shared during a
+        // missed-removal window.
+        let slot = queue.pop_front();
+        if queue.is_empty() {
+            self.live_ids.remove(&id);
+        }
+        match slot {
+            Some(Slot::Node(serial)) => {
+                if self.nodes.remove(&serial).is_none() {
+                    // Was still withheld — drop the obligation.
+                    self.withheld.remove(&serial);
+                }
+            }
+            Some(Slot::Port(serial)) => {
+                self.ports.remove(&serial);
+            }
+            Some(Slot::Link(serial)) => {
+                self.links.remove(&serial);
+                self.pending_links.remove(&serial);
+            }
+            Some(Slot::Client(serial)) => {
+                self.clients.remove(&serial);
+            }
+            Some(Slot::Device) => {
+                if let Some(count) = self.resolved_devices.get_mut(&id) {
+                    *count -= 1;
+                    if *count == 0 {
+                        self.resolved_devices.remove(&id);
+                    }
+                }
+            }
+            None => {
+                tracing::warn!(global_id = id.0, "observer: empty id slot on remove");
+            }
+        }
+        // A removal can drain the last obligation (a withheld node or pending
+        // link vanished before it resolved).
+        self.maybe_complete();
+    }
+
+    fn push_id(&mut self, id: GlobalId, slot: Slot) {
+        self.live_ids.entry(id).or_default().push_back(slot);
+    }
+
+    fn device_resolved(&self, id: GlobalId) -> bool {
+        self.resolved_devices.get(&id).is_some_and(|&n| n > 0)
+    }
+
+    /// Every obligation that must clear before the initial graph is trusted:
+    /// no node withheld on an unresolved Device, no Link awaiting its bind.
+    fn obligations_outstanding(&self) -> bool {
+        !self.withheld.is_empty() || !self.pending_links.is_empty()
+    }
+
+    /// Completion needs no clock — only the sync flag and an empty obligation
+    /// set — so it may fire on any mutating event. Sticky once reached.
+    fn maybe_complete(&mut self) {
+        if self.readiness != Readiness::Waiting {
+            return;
+        }
+        if self.server_synced && !self.obligations_outstanding() {
+            self.readiness = Readiness::Complete;
+            tracing::info!("observer: readiness epoch reached (synced + no obligations)");
+        }
+    }
+
+    /// Only the timeout consults the clock.
+    fn maybe_timeout(&mut self, now: Millis) {
+        if self.readiness != Readiness::Waiting {
+            return;
+        }
+        if now >= self.deadline {
+            self.readiness = Readiness::TimedOut;
+            tracing::warn!(
+                withheld = self.withheld.len(),
+                pending_links = self.pending_links.len(),
+                "observer: readiness epoch timed out with obligations outstanding — fail closed"
+            );
+        }
+    }
+
+    /// pipewire-pulse's PID from the current clients, validated against the
+    /// 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()?;
+        let comm = self.probed_comm.get(&candidate).and_then(|c| c.as_deref());
+        pulse_pid::validate(candidate, comm)
+    }
+
+    /// Project the current state into the taint engine's inputs.
+    pub fn project(&self) -> Projection {
+        let snapshot = GraphSnapshot::new(
+            self.nodes.values().cloned().collect(),
+            self.ports.values().cloned().collect(),
+            self.links.values().cloned().collect(),
+            self.clients.values().cloned().collect(),
+        );
+        Projection {
+            snapshot,
+            pipewire_pulse_pid: self.pulse_pid(),
+            graph_ready: self.graph_ready(),
+        }
+    }
+}
+
+fn link_snapshot(serial: Serial, id: GlobalId, e: LinkEndpoints) -> LinkSnapshot {
+    LinkSnapshot {
+        serial,
+        id,
+        output_node: e.output_node,
+        input_node: e.input_node,
+        output_port: e.output_port,
+        input_port: e.input_port,
+    }
+}

src/host/observer/pulse_pid.rs

@@ -0,0 +1,89 @@
+//! Deriving pipewire-pulse's own PID — pure, no PipeWire and no `/proc` I/O.
+//!
+//! The owner bridge's key 4 is `application.process.id`. For a stream created
+//! by a **Pulse-emulated** client that PID is *pipewire-pulse's own*, shared
+//! verbatim across every unrelated Pulse app, so bridging on it would fuse
+//! every Pulse module into one tainted owner (design v3.4 §5.2 correction 5,
+//! §6.1.2). The engine therefore needs to know that one PID so it can refuse
+//! to bridge on it — and **every** way of deriving it can fail, in which case
+//! the safe answer is `None`: key 4 becomes unusable (coarser, never wrong).
+//!
+//! 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
+//!    alone: the `pipewire.sec.pid` value shared across multiple Clients.
+//!    Native PipeWire clients carry their own distinct PID; only the
+//!    Pulse shim repeats one value, so a repeated value is the signal.
+//! 2. [`validate`] confirms that candidate against the `comm` the adapter
+//!    read from `/proc`. This is what closes **PID reuse**: a recycled PID
+//!    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,
+//! the property missing, `/proc` gone, a `comm` mismatch — yields `None`.
+
+use crate::host::taint::snapshot::ClientSnapshot;
+use std::collections::BTreeMap;
+
+/// 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 —
+/// and exact is the only safe match, since a prefix match would accept a
+/// 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.
+///
+/// Returns `Some(pid)` only when **exactly one** `pipewire.sec.pid` value is
+/// shared by two or more clients. Rationale, matched to the failure matrix:
+///
+/// - **consistent** — one value repeats, the rest (native clients) are
+///   distinct ⇒ that value.
+/// - **inconsistent** — two or more values each repeat ⇒ we cannot tell which
+///   is pulse ⇒ `None`.
+/// - **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
+/// `/proc/<candidate>/comm`.
+///
+/// `comm` is `None` when the adapter's read failed — the `/proc` entry is
+/// gone (the process exited between derivation and probe) — which is itself a
+/// reason to fail closed. A present-but-different `comm` is the **PID reuse**
+/// guard: the number is live but now belongs to someone else.
+pub fn validate(candidate: u32, comm: Option<&str>) -> Option<u32> {
+    match comm {
+        Some(PULSE_COMM) => Some(candidate),
+        _ => None,
+    }
+}
+
+/// The two stages composed, for callers that already hold the probed `comm`.
+/// The model keeps them separate (it recomputes the candidate as clients
+/// churn, and only re-probes when the candidate *changes*), 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> {
+    let candidate = candidate(clients)?;
+    validate(candidate, comm_of(candidate).as_deref())
+}

src/host/observer/tests.rs

@@ -0,0 +1,717 @@
+//! Pure exit-gate coverage for the phase-3 observer core.
+//!
+//! Five of the six exit-gate rows live here (the sixth — a live create/destroy
+//! topology diff — needs the daemon and belongs to the adapter). Each test
+//! builds the [`RegEvent`] stream by hand; nothing links PipeWire.
+//!
+//! Carrying the phase-0a lesson: the id/pid/serial tests use **interior**
+//! values, not just 1 and a huge number, so a middle-of-range mistake cannot
+//! hide.
+
+use super::classify::{Classification, DeviceClaim, classify};
+use super::pulse_pid;
+use super::*;
+use crate::host::taint::snapshot::{
+    ClientSnapshot, GlobalId, IdLookup, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial,
+};
+
+// ---- builders -------------------------------------------------------------
+
+fn ser(n: u64) -> Serial {
+    Serial(n)
+}
+fn gid(n: u32) -> GlobalId {
+    GlobalId(n)
+}
+fn model() -> RegistryModel {
+    // now=0, a 5 s readiness budget.
+    RegistryModel::new(0, 5000)
+}
+
+fn no_device() -> DeviceClaim {
+    DeviceClaim::default()
+}
+
+fn hw_claim(device_id: u32, api: &str, factory: &str) -> DeviceClaim {
+    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()),
+    }
+}
+
+/// A `Stream/Output/Audio` node with no backing Device — admitted at once.
+fn stream_out(serial: u64, id: u32) -> RegEvent {
+    RegEvent::NodeAdded(NodeObservation {
+        serial: ser(serial),
+        id: gid(id),
+        name: Some(format!("stream-{id}")),
+        role: MediaRole::StreamOutput,
+        props: NodeProps::default(),
+        device_claim: no_device(),
+    })
+}
+
+/// A node backed by a Device (withheld until that Device resolves).
+fn device_node(serial: u64, id: u32, role: MediaRole, claim: DeviceClaim) -> RegEvent {
+    RegEvent::NodeAdded(NodeObservation {
+        serial: ser(serial),
+        id: gid(id),
+        name: Some(format!("dev-node-{id}")),
+        role,
+        props: NodeProps::default(),
+        device_claim: claim,
+    })
+}
+
+fn client(serial: u64, id: u32, sec_pid: Option<u32>) -> RegEvent {
+    RegEvent::ClientAdded(ClientSnapshot {
+        serial: ser(serial),
+        id: gid(id),
+        sec_pid,
+    })
+}
+
+fn port(serial: u64, id: u32, node_id: u32, dir: PortDirection) -> RegEvent {
+    RegEvent::PortAdded(PortSnapshot {
+        serial: ser(serial),
+        id: gid(id),
+        node: gid(node_id),
+        direction: dir,
+        exclusive: false,
+        monitor: false,
+    })
+}
+
+fn endpoints(out_node: u32, in_node: u32) -> LinkEndpoints {
+    LinkEndpoints {
+        output_node: gid(out_node),
+        input_node: gid(in_node),
+        output_port: None,
+        input_port: None,
+    }
+}
+
+// ==========================================================================
+// classify() — session_device
+// ==========================================================================
+
+#[test]
+fn classify_no_device_is_not_a_device() {
+    assert_eq!(classify(&no_device(), false), Classification::NotADevice);
+    // `device_resolved` is irrelevant with no device_id.
+    assert_eq!(classify(&no_device(), true), Classification::NotADevice);
+}
+
+#[test]
+fn classify_unresolved_device_withholds() {
+    let claim = hw_claim(42, "alsa", "api.alsa.pcm.sink");
+    assert_eq!(
+        classify(&claim, false),
+        Classification::Withhold { device_id: gid(42) }
+    );
+}
+
+#[test]
+fn classify_resolved_hardware_pcm_is_session_device() {
+    // Only the measured ALSA factories are allowlisted (finding 5: the BlueZ
+    // entries were invented and were removed).
+    for factory in ["api.alsa.pcm.sink", "api.alsa.pcm.source"] {
+        assert_eq!(
+            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_alsa_without_driver_name_fails_closed() {
+    // Codex re-review: a missing `alsa.driver_name` must NOT grant
+    // session_device — an snd_aloop node whose driver prop was not copied onto
+    // the node would otherwise slip through. Absence fails closed.
+    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: None,
+        };
+        assert_eq!(
+            classify(&claim, true),
+            Classification::NotSessionDevice,
+            "absent driver on {factory} must fail closed"
+        );
+    }
+}
+
+#[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
+    // terminals: resolved, but session_device stays false.
+    for factory in ["support.null-audio-sink", "api.alsa.pcm.loopback", "wat"] {
+        assert_eq!(
+            classify(&hw_claim(7, "alsa", factory), true),
+            Classification::NotSessionDevice,
+            "factory {factory} must not be a session device"
+        );
+    }
+}
+
+#[test]
+fn classify_missing_device_api_fails_closed() {
+    // Even with an allowlisted factory, no device.api ⇒ not positively a
+    // real-backend terminal.
+    let claim = DeviceClaim {
+        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);
+}
+
+#[test]
+fn classify_allowlist_is_exact_not_substring() {
+    // A factory that merely *contains* an allowlisted name must not pass.
+    let claim = hw_claim(7, "alsa", "api.alsa.pcm.sink.evil");
+    assert_eq!(classify(&claim, true), Classification::NotSessionDevice);
+}
+
+// ==========================================================================
+// pulse_pid — the six-case derivation matrix
+// ==========================================================================
+
+fn clients_with(pids: &[Option<u32>]) -> Vec<ClientSnapshot> {
+    pids.iter()
+        .enumerate()
+        .map(|(i, &sec_pid)| ClientSnapshot {
+            serial: ser(1000 + i as u64),
+            id: gid(200 + i as u32),
+            sec_pid,
+        })
+        .collect()
+}
+
+#[test]
+fn pid_candidate_consistent_repeated_value() {
+    // interior pid values, not 1 / u32::MAX.
+    let cs = clients_with(&[Some(4137), Some(4137), Some(9001), Some(12034)]);
+    assert_eq!(pulse_pid::candidate(&cs), Some(4137));
+}
+
+#[test]
+fn pid_candidate_inconsistent_two_repeats_is_none() {
+    let cs = clients_with(&[Some(4137), Some(4137), Some(9001), Some(9001)]);
+    assert_eq!(pulse_pid::candidate(&cs), None);
+}
+
+#[test]
+fn pid_candidate_missing_property_is_none() {
+    let cs = clients_with(&[None, None, None]);
+    assert_eq!(pulse_pid::candidate(&cs), None);
+}
+
+#[test]
+fn pid_candidate_single_occurrence_is_none() {
+    // A lone native client carrying its own pid is indistinguishable from a
+    // one-client pulse; the >=2 threshold rejects it.
+    let cs = clients_with(&[Some(4137), Some(9001), Some(12034)]);
+    assert_eq!(pulse_pid::candidate(&cs), None);
+}
+
+#[test]
+fn pid_validate_matches_pulse_comm() {
+    assert_eq!(
+        pulse_pid::validate(4137, Some("pipewire-pulse")),
+        Some(4137)
+    );
+}
+
+#[test]
+fn pid_validate_proc_missing_is_none() {
+    // case 4: /proc entry gone.
+    assert_eq!(pulse_pid::validate(4137, None), None);
+}
+
+#[test]
+fn pid_validate_comm_mismatch_is_none() {
+    // case 5: a different process holds the number.
+    assert_eq!(pulse_pid::validate(4137, Some("firefox")), None);
+}
+
+#[test]
+fn pid_validate_reuse_named_other_process_is_none() {
+    // case 6: PID reuse — the number is live but /proc names someone else.
+    assert_eq!(pulse_pid::validate(4137, Some("Xwayland")), None);
+    // and a truncation-adjacent near-miss must not pass an exact match.
+    assert_eq!(pulse_pid::validate(4137, Some("pipewire-pulseX")), None);
+}
+
+#[test]
+fn pid_derive_end_to_end_valid() {
+    let cs = clients_with(&[Some(4137), Some(4137), Some(9001)]);
+    let got = pulse_pid::derive(&cs, |pid| {
+        (pid == 4137).then(|| "pipewire-pulse".to_string())
+    });
+    assert_eq!(got, Some(4137));
+}
+
+// ==========================================================================
+// model — pulse pid through project()
+// ==========================================================================
+
+/// Drive the model to Complete so `project` reflects a trusted graph, without
+/// caring about the specific objects.
+fn drive_ready(m: &mut RegistryModel) {
+    m.apply(RegEvent::ServerSynced);
+}
+
+#[test]
+fn model_pulse_pid_valid_through_projection() {
+    let mut m = model();
+    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));
+    m.apply(RegEvent::ProcCommProbed {
+        pid: 4137,
+        comm: Some("pipewire-pulse".to_string()),
+    });
+    assert_eq!(m.project().pipewire_pulse_pid, Some(4137));
+}
+
+#[test]
+fn model_pulse_pid_none_until_probed() {
+    let mut m = model();
+    m.apply(client(1, 200, Some(4137)));
+    m.apply(client(2, 201, Some(4137)));
+    // candidate exists, but no /proc confirmation yet ⇒ fail closed.
+    assert_eq!(m.project().pipewire_pulse_pid, None);
+}
+
+#[test]
+fn model_pulse_pid_none_on_comm_mismatch() {
+    let mut m = model();
+    m.apply(client(1, 200, Some(4137)));
+    m.apply(client(2, 201, Some(4137)));
+    m.apply(RegEvent::ProcCommProbed {
+        pid: 4137,
+        comm: Some("firefox".to_string()),
+    });
+    assert_eq!(m.project().pipewire_pulse_pid, None);
+}
+
+// ==========================================================================
+// model — add / remove of all four object types
+// ==========================================================================
+
+#[test]
+fn model_adds_all_four_object_types() {
+    let mut m = model();
+    m.apply(stream_out(100, 50));
+    m.apply(port(101, 60, 50, PortDirection::Out));
+    m.apply(client(102, 70, Some(4137)));
+    m.apply(RegEvent::LinkAdded {
+        serial: ser(103),
+        id: gid(80),
+        endpoints: Some(endpoints(50, 55)),
+    });
+    let snap = m.project().snapshot;
+    assert_eq!(snap.nodes().count(), 1);
+    assert_eq!(snap.ports().count(), 1);
+    assert_eq!(snap.clients().count(), 1);
+    assert_eq!(snap.links().count(), 1);
+}
+
+#[test]
+fn model_removes_all_four_object_types() {
+    let mut m = model();
+    m.apply(stream_out(100, 50));
+    m.apply(port(101, 60, 50, PortDirection::Out));
+    m.apply(client(102, 70, Some(4137)));
+    m.apply(RegEvent::LinkAdded {
+        serial: ser(103),
+        id: gid(80),
+        endpoints: Some(endpoints(50, 55)),
+    });
+
+    m.apply(RegEvent::Removed { id: gid(50) });
+    m.apply(RegEvent::Removed { id: gid(60) });
+    m.apply(RegEvent::Removed { id: gid(70) });
+    m.apply(RegEvent::Removed { id: gid(80) });
+
+    let snap = m.project().snapshot;
+    assert_eq!(snap.nodes().count(), 0);
+    assert_eq!(snap.ports().count(), 0);
+    assert_eq!(snap.clients().count(), 0);
+    assert_eq!(snap.links().count(), 0);
+}
+
+#[test]
+fn model_remove_of_unknown_id_is_harmless() {
+    let mut m = model();
+    m.apply(stream_out(100, 50));
+    m.apply(RegEvent::Removed { id: gid(999) });
+    assert_eq!(m.project().snapshot.nodes().count(), 1);
+}
+
+// ==========================================================================
+// model — recycled global id, oldest generation first (fail closed)
+// ==========================================================================
+
+#[test]
+fn model_recycled_id_is_ambiguous_until_removal_accounted() {
+    let mut m = model();
+    // A missed removal: two live nodes claim id 50 (serials 100 then 200).
+    m.apply(stream_out(100, 50));
+    m.apply(stream_out(200, 50));
+
+    // The snapshot fails closed: id 50 is ambiguous.
+    let snap = m.project().snapshot;
+    assert_eq!(snap.node_by_id(gid(50)), Some(IdLookup::Ambiguous));
+    assert_eq!(snap.nodes().count(), 2);
+
+    // One removal accounts for the OLDEST generation (serial 100); the newer
+    // node survives and the id is unambiguous again.
+    m.apply(RegEvent::Removed { id: gid(50) });
+    let snap = m.project().snapshot;
+    assert_eq!(snap.node_by_id(gid(50)), Some(IdLookup::Unique(ser(200))));
+    assert!(snap.node(ser(200)).is_some());
+    assert!(snap.node(ser(100)).is_none());
+}
+
+// ==========================================================================
+// model — Link endpoint resolution (bind fallback path)
+// ==========================================================================
+
+#[test]
+fn model_link_with_endpoints_appears_immediately() {
+    let mut m = model();
+    m.apply(RegEvent::LinkAdded {
+        serial: ser(103),
+        id: gid(80),
+        endpoints: Some(endpoints(50, 55)),
+    });
+    assert_eq!(m.project().snapshot.links().count(), 1);
+}
+
+#[test]
+fn model_link_without_endpoints_is_withheld_until_resolved() {
+    let mut m = model();
+    // The correctness path: the global carried no endpoint props.
+    m.apply(RegEvent::LinkAdded {
+        serial: ser(103),
+        id: gid(80),
+        endpoints: None,
+    });
+    // Not in the snapshot yet, and it blocks readiness.
+    assert_eq!(m.project().snapshot.links().count(), 0);
+    m.apply(RegEvent::ServerSynced);
+    assert!(!m.graph_ready(), "pending link must hold readiness");
+
+    // The bind fallback resolves it.
+    m.apply(RegEvent::LinkEndpointsResolved {
+        serial: ser(103),
+        endpoints: endpoints(50, 55),
+    });
+    let snap = m.project().snapshot;
+    assert_eq!(snap.links().count(), 1);
+    let link = snap.links().next().unwrap();
+    assert_eq!(link.output_node, gid(50));
+    assert_eq!(link.input_node, gid(55));
+    assert!(
+        m.graph_ready(),
+        "resolving the last obligation completes readiness"
+    );
+}
+
+#[test]
+fn model_stale_link_resolution_is_ignored() {
+    let mut m = model();
+    m.apply(RegEvent::LinkAdded {
+        serial: ser(103),
+        id: gid(80),
+        endpoints: None,
+    });
+    // Link removed before the bind returned.
+    m.apply(RegEvent::Removed { id: gid(80) });
+    // A late resolution for the gone link must not resurrect it.
+    m.apply(RegEvent::LinkEndpointsResolved {
+        serial: ser(103),
+        endpoints: endpoints(50, 55),
+    });
+    assert_eq!(m.project().snapshot.links().count(), 0);
+    m.apply(RegEvent::ServerSynced);
+    assert!(
+        m.graph_ready(),
+        "the obligation cleared when the link was removed"
+    );
+}
+
+// ==========================================================================
+// model — readiness epoch
+// ==========================================================================
+
+#[test]
+fn model_readiness_waits_for_sync() {
+    let mut m = model();
+    m.apply(stream_out(100, 50));
+    assert_eq!(m.readiness(), Readiness::Waiting);
+    assert!(!m.project().graph_ready);
+    m.apply(RegEvent::ServerSynced);
+    assert_eq!(m.readiness(), Readiness::Complete);
+    assert!(m.project().graph_ready);
+}
+
+#[test]
+fn model_readiness_does_not_release_with_obligation_outstanding() {
+    let mut m = model();
+    // A node withheld on an unresolved device is an outstanding obligation.
+    m.apply(device_node(
+        100,
+        50,
+        MediaRole::Sink,
+        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
+    ));
+    m.apply(RegEvent::ServerSynced);
+    // Synced, but the withheld node keeps the epoch shut.
+    assert_eq!(m.readiness(), Readiness::Waiting);
+    assert!(!m.graph_ready());
+
+    // Resolving the device admits the node and completes readiness.
+    m.apply(RegEvent::DeviceAdded { id: gid(42) });
+    assert_eq!(m.readiness(), Readiness::Complete);
+    assert!(m.graph_ready());
+}
+
+#[test]
+fn model_readiness_times_out_fail_closed() {
+    let mut m = model();
+    m.apply(device_node(
+        100,
+        50,
+        MediaRole::Sink,
+        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
+    ));
+    m.apply(RegEvent::ServerSynced);
+    assert_eq!(m.readiness(), Readiness::Waiting);
+
+    // The device never resolves; the deadline passes.
+    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]
+fn model_tick_before_deadline_does_not_time_out() {
+    let mut m = model();
+    m.apply(device_node(
+        100,
+        50,
+        MediaRole::Sink,
+        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
+    ));
+    m.apply(RegEvent::Tick { now: 4999 });
+    assert_eq!(m.readiness(), Readiness::Waiting);
+}
+
+#[test]
+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);
+    assert!(m.graph_ready());
+    // 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, "epoch stays sticky");
+    // ...but graph_ready DOES drop while the obligation is outstanding
+    // (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]
+fn model_withheld_node_removed_clears_obligation() {
+    let mut m = model();
+    m.apply(device_node(
+        100,
+        50,
+        MediaRole::Sink,
+        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
+    ));
+    m.apply(RegEvent::ServerSynced);
+    assert_eq!(m.readiness(), Readiness::Waiting);
+    // The withheld node disappears before its device ever showed up.
+    m.apply(RegEvent::Removed { id: gid(50) });
+    assert_eq!(m.readiness(), Readiness::Complete);
+}
+
+// ==========================================================================
+// model — device withholding & session_device flag
+// ==========================================================================
+
+#[test]
+fn model_device_first_admits_node_immediately() {
+    let mut m = model();
+    // Device enumerated before the node that references it.
+    m.apply(RegEvent::DeviceAdded { id: gid(42) });
+    m.apply(device_node(
+        100,
+        50,
+        MediaRole::Sink,
+        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
+    ));
+    let snap = m.project().snapshot;
+    let node = snap.node(ser(100)).expect("node admitted immediately");
+    assert!(
+        node.props.session_device,
+        "hardware sink is a session device"
+    );
+    // No obligation ⇒ a sync completes readiness.
+    m.apply(RegEvent::ServerSynced);
+    assert!(m.graph_ready());
+}
+
+#[test]
+fn model_withheld_node_admitted_with_correct_session_device() {
+    let mut m = model();
+    // A real hardware sink and a card-associated filter share client/device
+    // ancestry but classify differently once the device resolves.
+    m.apply(device_node(
+        100,
+        50,
+        MediaRole::Sink,
+        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
+    ));
+    m.apply(device_node(
+        200,
+        51,
+        MediaRole::Sink,
+        hw_claim(42, "alsa", "support.null-audio-sink"),
+    ));
+    // Both withheld until the device resolves.
+    assert_eq!(m.project().snapshot.nodes().count(), 0);
+    m.apply(RegEvent::DeviceAdded { id: gid(42) });
+
+    let snap = m.project().snapshot;
+    assert_eq!(
+        snap.nodes().count(),
+        2,
+        "both admitted once the device resolved"
+    );
+    assert!(
+        snap.node(ser(100)).unwrap().props.session_device,
+        "the real hardware sink is a session device"
+    );
+    assert!(
+        !snap.node(ser(200)).unwrap().props.session_device,
+        "the null sink sharing the same device is not"
+    );
+}
+
+#[test]
+fn model_withheld_filter_admitted_as_not_session_device() {
+    let mut m = model();
+    m.apply(device_node(
+        200,
+        51,
+        MediaRole::Sink,
+        hw_claim(42, "alsa", "support.null-audio-sink"),
+    ));
+    m.apply(RegEvent::DeviceAdded { id: gid(42) });
+    let snap = m.project().snapshot;
+    assert!(
+        !snap.node(ser(200)).unwrap().props.session_device,
+        "a null sink on a card is not a session device"
+    );
+}