host/observer: phase 3 pure core — RegistryModel reducer + classifiers

My half of the phase-3 split (impl plan §4). Pure, no PipeWire: the adapter
(Codex's half) translates live registry callbacks / binds / /proc reads /
core.sync into RegEvents and feeds this reducer.

- RegistryModel::apply folds RegEvents into serial-keyed maps with an
  insertion-ordered id index so global_remove accounts for the oldest
  generation first; recycled ids stay Ambiguous until accounted (v3.4 §6.1.3).
- Readiness epoch: graph_ready false until ServerSynced + no outstanding
  obligations (withheld nodes, pending link binds); bounded timeout fails
  closed. Gates sticky retirement only; sticky once terminal.
- session_device classifier: hardware-PCM factory allowlist, exact match,
  fail closed to false; a node on an unresolved Device is withheld, never
  admitted provisional.
- pulse-PID derivation split into pure candidate (repeated sec_pid) + validate
  (/proc comm), so the 6-case failure matrix is unit-testable; any failure =>
  None (key 4 unusable).

34 tests cover 5 of 6 exit-gate rows (the live topology-diff row is the
adapter's). cargo test --bins 117 green, fmt + clippy clean.

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

src/host/mod.rs

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

src/host/observer/classify.rs

@@ -0,0 +1,119 @@
+//! 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.
+const HARDWARE_PCM_FACTORIES: &[&str] = &[
+    // ALSA — measured on the target box.
+    "api.alsa.pcm.sink",
+    "api.alsa.pcm.source",
+    // BlueZ — the equivalent real backend for Bluetooth audio terminals.
+    "api.bluez5.pcm.sink",
+    "api.bluez5.pcm.source",
+];
+
+/// 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>,
+}
+
+/// 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 is_hardware_pcm = claim.device_api.is_some()
+        && claim
+            .factory_name
+            .as_deref()
+            .is_some_and(|f| HARDWARE_PCM_FACTORIES.contains(&f));
+    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,464 @@
+//! 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`]).
+
+#![allow(dead_code)] // Wired by the phase-3 adapter (Codex's half) and consumed by later phases.
+
+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,
+    /// Server synced and every obligation resolved. `graph_ready` is true.
+    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
+    }
+
+    pub fn graph_ready(&self) -> bool {
+        matches!(self.readiness, Readiness::Complete)
+    }
+
+    /// 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,620 @@
+//! 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()),
+    }
+}
+
+/// 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() {
+    for factory in [
+        "api.alsa.pcm.sink",
+        "api.alsa.pcm.source",
+        "api.bluez5.pcm.sink",
+        "api.bluez5.pcm.source",
+    ] {
+        let api = if factory.contains("bluez5") {
+            "bluez5"
+        } else {
+            "alsa"
+        };
+        assert_eq!(
+            classify(&hw_claim(7, api, factory), true),
+            Classification::SessionDevice,
+            "factory {factory} should be a session device"
+        );
+    }
+}
+
+#[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()),
+    };
+    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");
+}
+
+#[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_is_sticky_across_later_churn() {
+    let mut m = model();
+    m.apply(RegEvent::ServerSynced);
+    assert_eq!(m.readiness(), Readiness::Complete);
+    // A node withheld AFTER completion must not un-complete the epoch — post
+    // enumeration, withholding is per-object (the node is simply absent).
+    m.apply(device_node(
+        100,
+        50,
+        MediaRole::Sink,
+        hw_claim(42, "alsa", "api.alsa.pcm.sink"),
+    ));
+    assert_eq!(m.readiness(), Readiness::Complete);
+    assert!(m.graph_ready());
+    // ...and a late timeout Tick is inert once Complete.
+    m.apply(RegEvent::Tick { now: 100_000 });
+    assert_eq!(m.readiness(), Readiness::Complete);
+}
+
+#[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"
+    );
+}