From b3d71724ae9dc0547fa0d48d701070ea1b5c1ef6 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sat, 25 Jul 2026 18:17:48 -0400 Subject: [PATCH 1/3] =?UTF-8?q?host/observer:=20phase=203r=20pure=20core?= =?UTF-8?q?=20=E2=80=94=20node/device=20props=20come=20from=20a=20bind?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit v3.5 §6.7. The registry `global` event announces only a fixed 13-key subset of a Node's properties, and eight the engine depends on are never among them (phase-5 gate failure F1/F2). The core now treats the global as an index and takes every property from the object's bound `info`. - `RegEvent::NodeAdded { serial, id }` is identity only; `RegEvent::NodeInfo` carries the properties and is both the first resolution and every later PROPS change for the node's lifetime (decision 2). Same split for Device (`DeviceAdded` / `DeviceInfo`). - A node with no `info` is withheld from the snapshot and is a readiness obligation; an unresolvable bind ends in sticky `TimedOut`, fail closed (decision 3). Devices are keyed by serial too, so a recycled device id with two live claimants is ambiguous ⇒ withheld rather than guessed. - One live-node map replaces the admitted/withheld pair; classification is recomputed at projection time from current inputs, since both sides of it now change over an object's lifetime. - `classify` takes the bound Device's props: presence is a union with the Device winning (this recovers a real card whose node was never given `alsa.driver_name` — the phase-3 review's owed fix), while the non-terminal-driver denylist is a union in the safe direction. - `apply` returns `Outcome`, the only sound place to enforce the suppression rule: a property update is dropped only when model state provably did not change, i.e. the resulting projection is identical. Adapter: stops reading properties off Node/Device globals and emits the new index events. Binding every Node and Device — the I/O half — is the next commit (Codex's), so until then every node is withheld and readiness times out by design. Tests: 55 observer (was 38) — the prop-update matrix, readiness with node binds, and recycled-Node-id churn (phase 3r gate rows 2–4). 195 green, clippy -D warnings and fmt clean. --- src/host/observer/adapter.rs | 73 ++- src/host/observer/classify.rs | 124 +++-- src/host/observer/mod.rs | 447 +++++++++++++----- src/host/observer/tests.rs | 842 +++++++++++++++++++++++++++++----- 4 files changed, 1171 insertions(+), 315 deletions(-) diff --git a/src/host/observer/adapter.rs b/src/host/observer/adapter.rs index 25289cf..fe02da4 100644 --- a/src/host/observer/adapter.rs +++ b/src/host/observer/adapter.rs @@ -4,12 +4,9 @@ //! callbacks into [`RegEvent`]s, and publishes the latest [`Projection`] for //! consumers running outside the PipeWire thread. -use super::classify::DeviceClaim; -use super::{EventKind, LinkEndpoints, NodeObservation, Projection, RegEvent, RegistryModel}; +use super::{EventKind, LinkEndpoints, Outcome, Projection, RegEvent, RegistryModel}; use crate::host::audio::parse_object_serial; -use crate::host::taint::snapshot::{ - ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial, -}; +use crate::host::taint::snapshot::{ClientSnapshot, GlobalId, PortDirection, PortSnapshot, Serial}; use anyhow::{Context, Result}; use pipewire::{self as pw, types::ObjectType}; use std::cell::{Cell, RefCell}; @@ -147,7 +144,7 @@ impl ObserverState { // of observation produced the projection, and deriving that from the // event itself is what stops the two from ever disagreeing. let kind = event.kind(); - self.model.apply(event); + let mut outcome = self.model.apply(event); let candidate = self.model.pulse_pid_candidate(); if candidate != self.last_candidate { @@ -160,11 +157,18 @@ impl ObserverState { // one registry event still yields exactly one sink call — the // no-coalescing contract cuts both ways, and a *duplicated* // observation would make the O5 event rate a fiction. - self.model.apply(RegEvent::ProcCommProbed { pid, comm }); + if self.model.apply(RegEvent::ProcCommProbed { pid, comm }) == Outcome::Applied { + outcome = Outcome::Applied; + } } } - self.publish(kind); + // v3.5 §6.7 decision 2: a projection the model proved identical is not + // published. Only the model can make that claim soundly, which is why + // it is [`Outcome`] and not a diff of two snapshots here. + if outcome == Outcome::Applied { + self.publish(kind); + } } fn publish(&mut self, kind: EventKind) { @@ -273,6 +277,9 @@ fn run_observer( match obj.type_ { ObjectType::Node => { + // ⚠️ v3.5 §6.7: the global is an INDEX. Only `object.serial` + // is read here; every property the engine reasons about + // comes from the bind's `info` (phase 3r). let Some(props) = obj.props.as_ref() else { tracing::warn!( node_id = obj.id, @@ -284,41 +291,9 @@ fn run_observer( 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::().ok()), - link_group: props.get("node.link-group").map(str::to_owned), - client_id: props - .get("client.id") - .and_then(|value| value.parse::().ok()) - .map(GlobalId), - process_id: props - .get("application.process.id") - .and_then(|value| value.parse::().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::().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)); + .add(id, RegEvent::NodeAdded { serial, id }); } ObjectType::Port => { let Some(props) = obj.props.as_ref() else { @@ -392,9 +367,23 @@ fn run_observer( ); } ObjectType::Device => { + // Index only, exactly as for a Node: `device.api` and + // `alsa.driver_name` live on the bind's `info` (v3.5 §6.7 + // decision 4), not here. + let Some(props) = obj.props.as_ref() else { + tracing::warn!( + device_id = obj.id, + "registry observer: Device has no properties; dropping" + ); + return; + }; + let Some(serial) = parse_serial(obj.id, "Device", props.get("object.serial")) + else { + return; + }; state_for_global .borrow_mut() - .add(id, RegEvent::DeviceAdded { id }); + .add(id, RegEvent::DeviceAdded { serial, id }); } ObjectType::Link => { let Some(props) = obj.props.as_ref() else { diff --git a/src/host/observer/classify.rs b/src/host/observer/classify.rs index 37080f6..299d2d2 100644 --- a/src/host/observer/classify.rs +++ b/src/host/observer/classify.rs @@ -18,11 +18,21 @@ //! 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). +//! - The backing Device must actually have been **bound and resolved**. A node +//! that claims a `device.id` whose Device's properties we do not hold 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). +//! +//! **Round 8 (v3.5 §6.7 decision 4): the Device is the authority on +//! `device.api` and `alsa.driver_name`.** Both are absent from the Node +//! *global* and both are present on the **bound Device**'s `info` props +//! (measured 2026-07-25). Reading them from the Device closes the phase-3 +//! review's owed fix: on PipeWire ≥ 1.2.6 with WirePlumber < 0.5.13 the driver +//! name is not copied onto the node, and the fail-closed "absent driver ⇒ not +//! a session device" rule would over-exclude real sound cards. `factory.name` +//! exists only on the node, which is why the node bind is required regardless. use crate::host::taint::snapshot::GlobalId; @@ -67,8 +77,9 @@ const HARDWARE_PCM_FACTORIES: &[&str] = &[ /// 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 +/// The node-side properties the classifier reads, exactly as the adapter +/// parsed them off the **bound Node's `info`** (never off the registry +/// global — v3.5 §6.7). 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. @@ -78,10 +89,12 @@ pub struct DeviceClaim { /// `Stream/*` nodes, which is exactly why their absence means "not a /// device", not "unknown". pub device_id: Option, - /// `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. + /// `device.api` **as copied onto the node**, when it is — 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. The + /// authoritative copy is [`DeviceProps::device_api`]; this is the + /// fallback. pub device_api: Option, /// `factory.name` — the discriminator. Only an allowlisted hardware-PCM /// factory earns `session_device`. @@ -92,8 +105,27 @@ pub struct DeviceClaim { /// 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. + /// Frequently absent here — PipeWire ≥ 1.2.6 with WirePlumber < 0.5.13 + /// does not copy `alsa.*` onto the node — which is why the authoritative + /// copy is [`DeviceProps::alsa_driver_name`] and this is only the + /// fallback. + pub alsa_driver_name: Option, +} + +/// The **bound Device's** `info` properties — the authoritative half of the +/// `session_device` decision (v3.5 §6.7 decision 4). +/// +/// Absent from the Device *registry global* exactly as the node's properties +/// are absent from the Node global; both are recovered by binding. A node +/// claiming a `device.id` is withheld until this struct exists for that +/// Device (see [`Classification::Withhold`]). +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct DeviceProps { + /// `device.api` on the Device — `alsa`, `bluez5`, `v4l2`, … + pub device_api: Option, + /// `alsa.driver_name` on the Device — the kernel driver behind the card, + /// authoritative regardless of whether the session manager copied it onto + /// the node. pub alsa_driver_name: Option, } @@ -102,9 +134,10 @@ pub struct DeviceClaim { 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. + /// A `device.id` is claimed but the backing Device's properties are not + /// held: never observed, its bind still outstanding, or its global id + /// ambiguously shared by two live Devices. **Withhold the node and keep + /// the readiness epoch not-ready**; re-classify when the Device resolves. Withhold { device_id: GlobalId }, /// Positively a passive hardware terminal. Admit with /// `session_device=true`. @@ -115,42 +148,59 @@ pub enum Classification { NotSessionDevice, } -/// Classify a node's device claim. +/// Classify a node's device claim against its backing Device. /// -/// `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 { +/// `device` is the bound Device's properties, and `None` means the claim is +/// **unresolved** — never observed, bind outstanding, or an ambiguous +/// recycled id. It is only consulted when a `device_id` is present. Pure: the +/// model looks the Device up, and the I/O of *binding* it lives in the +/// adapter. +/// +/// Where the two sides disagree the rule is deliberately asymmetric, and +/// safety picks the direction (v3.5 §6.7 decision 4): +/// +/// - **Presence: the Device wins, the node is the fallback.** That is what +/// recovers a real card whose node was never given `alsa.driver_name`. +/// - **The denylist is a union.** If *either* side names a non-terminal +/// driver the node is not a session device. A disagreement here is not +/// expected on any measured configuration, and treating it as "the Device +/// says it is fine" would be the one reading that can leak. +pub fn classify(claim: &DeviceClaim, device: Option<&DeviceProps>) -> 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 + let Some(device) = device else { + // Backed by a Device we have not resolved — 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 + // (Codex phase-3 re-review). The factory allowlist cannot tell a real card + // from `snd_aloop`, which presents the same `api.alsa.pcm.*` factory, so a + // *missing* value must not be read as "not a loopback". Round 8 makes the + // bound Device the primary source, so a real card is no longer + // over-excluded merely because the session manager did not copy `alsa.*` + // onto its node. + let driver = device .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; + .or(claim.alsa_driver_name.as_deref()); + let driver_denied = [ + device.alsa_driver_name.as_deref(), + claim.alsa_driver_name.as_deref(), + ] + .into_iter() + .flatten() + .any(|d| NON_TERMINAL_ALSA_DRIVERS.contains(&d)); + let driver_ok = driver.is_some() && !driver_denied; + let api_present = device.device_api.is_some() || claim.device_api.is_some(); + let is_hardware_pcm = api_present && on_factory_allowlist && driver_ok; if is_hardware_pcm { Classification::SessionDevice } else { diff --git a/src/host/observer/mod.rs b/src/host/observer/mod.rs index e340de6..108ba50 100644 --- a/src/host/observer/mod.rs +++ b/src/host/observer/mod.rs @@ -1,12 +1,34 @@ -//! The registry observer's **pure core** (impl plan §4, phase 3). +//! The registry observer's **pure core** (impl plan §4, phases 3 and 3r). //! //! 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. +//! live registry callbacks, 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. +//! +//! ## 🔴 Round 8 (v3.5 §6.7): the global is an INDEX, not a source of truth +//! +//! Phase 3 shipped reading node properties off the registry `global` event. +//! The registry announces only a fixed 13-key subset for a Node, and **eight +//! properties this feature depends on are never among them** — they read as +//! absent rather than failing, so the engine was silently, permanently +//! starved of both its primary taint root and every strong owner key (the +//! phase-5 gate failure, F1/F2). The rule that replaces it: +//! +//! > A node's properties come from a **bind**, never from the global. The +//! > global tells us an object exists, its id and its serial. Everything +//! > else — including `node.name` and `media.class`, so there is exactly one +//! > source — arrives on [`RegEvent::NodeInfo`]. Same for `Device` +//! > ([`RegEvent::DeviceInfo`]). +//! +//! Consequences visible in this file: a Node is admitted to the snapshot +//! **only** once its `info` has arrived (until then it is withheld and is a +//! readiness obligation); a Device resolves a node's claim only once *its* +//! `info` has arrived; and `info` may fire again for the lifetime of the +//! object, so [`RegEvent::NodeInfo`] is both the first resolution and every +//! later property change (v3.5 §6.7 decisions 1–4). //! //! Three things this core is shaped to get right, each an exit-gate row: //! @@ -14,18 +36,20 @@ //! 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). +//! [`IdLookup::Ambiguous`] — fail closed (v3.4 §6.1.3). Everything the +//! model *owns* is keyed by never-recycled `object.serial`; ids are only +//! ever a lookup. //! - **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`]). +//! - **Withholding on unresolved input.** A node with no `info` yet, or one +//! claiming a `device.id` whose Device we have not resolved, is held out of +//! the snapshot entirely rather than admitted with provisional ownership +//! (see [`classify`]). //! -//! **Two accepted limitations (Codex phase-3 review, findings 3 and 4), both -//! low-reachability, owed to a later hardening round:** +//! **Three accepted limitations, all 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 @@ -45,6 +69,13 @@ //! silently drop `global_remove`, so this needs callback loss to trigger. //! The snapshot treats the two-claimant window as [`IdLookup::Ambiguous`] //! (fail closed) meanwhile. +//! - *An unresolvable bind takes the whole graph down, not just its node* +//! (v3.5 §6.7 decision 3). A node whose `info` never arrives keeps +//! readiness false until the deadline, then sticky-[`Readiness::TimedOut`] +//! — no fan-out at all, identical to a never-resolving Link bind. Per-node +//! quarantine (that node ineligible **and** taint-bearing, the rest of the +//! graph still working) is strictly better and is deferred because it is a +//! new concept in the *pure engine*, not a fix to the observer. #![allow(dead_code)] // Wired by the phase-3 adapter (Codex's half) and consumed by later phases. @@ -59,7 +90,7 @@ use crate::host::taint::snapshot::{ ClientSnapshot, GlobalId, GraphSnapshot, LinkSnapshot, MediaRole, NodeProps, NodeSnapshot, PortSnapshot, Serial, }; -use classify::{Classification, DeviceClaim}; +use classify::{Classification, DeviceClaim, DeviceProps}; use std::collections::{BTreeMap, VecDeque}; /// A monotonic millisecond clock value, supplied by the adapter via @@ -67,14 +98,17 @@ use std::collections::{BTreeMap, VecDeque}; /// [`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. +/// A Node's **bound `info` properties** — the sole source of node properties +/// (v3.5 §6.7), delivered by [`RegEvent::NodeInfo`]. +/// +/// This carries no identity: the serial names the node on the event and the +/// global id was recorded by [`RegEvent::NodeAdded`], so the adapter cannot +/// contradict the index it already published. `session_device` inside +/// [`NodeObservation::props`] is left at its `false` default; the model +/// overwrites it from the [`classify`] result at projection time, 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, pub role: MediaRole, pub props: NodeProps, @@ -97,19 +131,39 @@ pub struct LinkEndpoints { /// 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 Node global appeared. **Index only** — the global's properties are a + /// filtered subset and are not read (v3.5 §6.7). The node is withheld + /// from the snapshot and is a readiness obligation until its + /// [`RegEvent::NodeInfo`] arrives. + NodeAdded { serial: Serial, id: GlobalId }, + /// A bound Node's `info` properties. **Both** the first resolution and + /// every later `PROPS` change for the node's lifetime — the model tells + /// them apart, so the adapter holds no per-node "have I seen info yet?" + /// state to get wrong. An `info` for a serial we do not hold (a node + /// already removed) is ignored. + NodeInfo { + serial: Serial, + observation: 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 Device global appeared. Index only, exactly as for a Node: it does + /// not resolve anything until [`RegEvent::DeviceInfo`] arrives. + DeviceAdded { serial: Serial, id: GlobalId }, + /// A bound Device's `info` properties — the **authoritative** source of + /// `device.api` and `alsa.driver_name` (v3.5 §6.7 decision 4). Resolves + /// every node withheld on this Device's id. + DeviceInfo { serial: Serial, props: DeviceProps }, /// 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. + /// + /// Unlike Nodes and Devices, Link endpoint props **are** announced on the + /// global (measured, phase-5 results F1), so this asymmetry is real and + /// deliberate. LinkAdded { serial: Serial, id: GlobalId, @@ -143,7 +197,7 @@ pub enum RegEvent { /// one is worth suppressing on a tick but never on a graph event. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum EventKind { - /// A registry observation: an add, a removal, a link resolution, a `/proc` + /// A registry observation: an add, a removal, a bind resolution, a `/proc` /// probe, or the server sync. Graph, /// The periodic clock sample. Carries no graph information; it exists so the @@ -169,16 +223,39 @@ impl RegEvent { } } +/// Whether an applied event could have changed the projection. +/// +/// The suppression rule of v3.5 §6.7 decision 2, in the one place that can +/// enforce it: **a property update may be dropped only when the resulting +/// [`Projection`] is identical to the current one.** The projection is a pure +/// function of model state, so "state provably unchanged" *is* "projection +/// identical" — which is what [`Outcome::Suppressed`] means and why the check +/// is a cheap field comparison rather than building and diffing two snapshots. +/// +/// Anything looser (dropping updates that do change state) breaks phase 4's +/// no-coalescing contract, which needs to see the empty gap between an AEC +/// module unload and a reload that reuses the index. Anything stricter +/// (publishing on every `info`, including the state-only changes PipeWire +/// emits constantly) inflates the O5 event rate with non-events. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Outcome { + /// Model state may have changed; the caller must publish the projection. + Applied, + /// Model state provably did not change; publishing is optional and the + /// adapter skips it. + Suppressed, +} + /// 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. +/// only the id, so the index remembers what each id currently holds. Every +/// slot names its object by never-recycled serial. #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum Slot { Node(Serial), Port(Serial), Link(Serial), Client(Serial), - Device, + Device(Serial), } /// The readiness epoch. A one-time transition out of [`Readiness::Waiting`]; @@ -220,25 +297,45 @@ pub struct Projection { pub readiness: Readiness, } +/// A live Node: its global id (for link endpoint lookup) plus its bound +/// properties once they arrive. +#[derive(Clone, Debug, PartialEq, Eq)] +struct NodeEntry { + id: GlobalId, + /// `None` while the bind is outstanding — withheld from the snapshot and + /// an outstanding readiness obligation (v3.5 §6.7 decision 3). + obs: Option, +} + +/// A live Device: its global id plus its bound properties once they arrive. +#[derive(Clone, Debug, PartialEq, Eq)] +struct DeviceEntry { + id: GlobalId, + /// `None` while the bind is outstanding. A node claiming this Device + /// stays withheld until it is `Some` — the Device's `device.api` and + /// `alsa.driver_name` are the authoritative inputs to `session_device` + /// (v3.5 §6.7 decision 4), so classifying without them would be the same + /// provisional answer the contract forbids. + props: Option, +} + /// 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, + /// **Every** live Node, keyed by serial — admitted or withheld. Admission + /// is decided at projection time from the entry's own state, so there is + /// no admitted/withheld pair of maps to drift apart. + nodes: BTreeMap, + /// Every live Device, keyed by serial. + devices: BTreeMap, ports: BTreeMap, links: BTreeMap, clients: BTreeMap, - /// Nodes held out of the snapshot pending their Device's resolution. - withheld: BTreeMap, /// Links whose endpoints the adapter is still binding; the id is kept so /// removal and resolution can find them. pending_links: BTreeMap, - /// Live Device global ids, ref-counted so a recycled id is only - /// considered resolved while a Device actually holds it. - resolved_devices: BTreeMap, - /// Insertion-ordered holders of each live global id. `global_remove` /// accounts for the oldest generation first (v3.4 §6.1.3). live_ids: BTreeMap>, @@ -259,12 +356,11 @@ impl RegistryModel { pub fn new(now: Millis, timeout: Millis) -> Self { Self { nodes: BTreeMap::new(), + devices: 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, @@ -283,21 +379,22 @@ impl RegistryModel { /// /// 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. + /// no current obligations outstanding (a node whose bind is 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 round 8 adds the far more common case: an + /// unbound node is an invisible *vertex*, which hides everything the edge + /// case hides and its ownership besides. /// /// [`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. + /// normal 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() } @@ -310,102 +407,120 @@ impl RegistryModel { pulse_pid::candidate(&clients) } - /// Fold one observation into the model. - pub fn apply(&mut self, event: RegEvent) { + /// Fold one observation into the model. The returned [`Outcome`] tells the + /// caller whether the projection can have changed; see [`Outcome`] for why + /// that is the only sound place to enforce the suppression rule. + pub fn apply(&mut self, event: RegEvent) -> Outcome { match event { - RegEvent::NodeAdded(obs) => self.on_node_added(obs), + RegEvent::NodeAdded { serial, id } => { + self.push_id(id, Slot::Node(serial)); + self.nodes.insert(serial, NodeEntry { id, obs: None }); + // A node awaiting its bind is a fresh obligation, so this can + // only ever *hold* readiness, never complete it — but the + // re-check is cheap and keeps the invariant local. + self.maybe_complete(); + Outcome::Applied + } + RegEvent::NodeInfo { + serial, + observation, + } => self.on_node_info(serial, observation), RegEvent::PortAdded(port) => { self.push_id(port.id, Slot::Port(port.serial)); self.ports.insert(port.serial, port); + Outcome::Applied } 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. + Outcome::Applied } - RegEvent::DeviceAdded { id } => self.on_device_added(id), + RegEvent::DeviceAdded { serial, id } => { + self.push_id(id, Slot::Device(serial)); + self.devices.insert(serial, DeviceEntry { id, props: None }); + self.maybe_complete(); + Outcome::Applied + } + RegEvent::DeviceInfo { serial, props } => self.on_device_info(serial, props), RegEvent::LinkAdded { serial, id, endpoints, - } => self.on_link_added(serial, id, endpoints), + } => { + self.on_link_added(serial, id, endpoints); + Outcome::Applied + } RegEvent::LinkEndpointsResolved { serial, endpoints } => { self.on_link_resolved(serial, endpoints) } RegEvent::ProcCommProbed { pid, comm } => { - self.probed_comm.insert(pid, comm); + let previous = self.probed_comm.insert(pid, comm.clone()); + if previous.as_ref() == Some(&comm) { + Outcome::Suppressed + } else { + Outcome::Applied + } } RegEvent::Removed { id } => self.on_removed(id), RegEvent::ServerSynced => { + let already = self.server_synced; self.server_synced = true; self.maybe_complete(); + if already { + Outcome::Suppressed + } else { + Outcome::Applied + } } RegEvent::Tick { now } => { self.last_now = now; self.maybe_timeout(now); + Outcome::Applied } } } - 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) - } + /// First resolution *and* every later property change (v3.5 §6.7 + /// decision 2). The model distinguishes them by what it already holds, so + /// the adapter can forward every `info` callback unconditionally. + fn on_node_info(&mut self, serial: Serial, observation: NodeObservation) -> Outcome { + let Some(entry) = self.nodes.get_mut(&serial) else { + // A late `info` for a node already removed. Re-inserting it here + // would resurrect a dead node with no id index behind it. + tracing::debug!(serial = serial.0, "observer: node info for an unknown node"); + return Outcome::Suppressed; + }; + if entry.obs.as_ref() == Some(&observation) { + // The state-only `info` callbacks PipeWire emits constantly: same + // properties, so the projection is provably identical. + return Outcome::Suppressed; } - // 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. + entry.obs = Some(observation); + // The first `info` retires this node's obligation, which can be the + // last one outstanding. self.maybe_complete(); + Outcome::Applied } - 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 = 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); - } + fn on_device_info(&mut self, serial: Serial, props: DeviceProps) -> Outcome { + let Some(entry) = self.devices.get_mut(&serial) else { + tracing::debug!( + serial = serial.0, + "observer: device info for an unknown device" + ); + return Outcome::Suppressed; + }; + if entry.props.as_ref() == Some(&props) { + return Outcome::Suppressed; } + entry.props = Some(props); + // Resolving a Device admits every node that was withheld on it — + // which happens at projection time; here it can only retire + // obligations. self.maybe_complete(); + Outcome::Applied } fn on_link_added(&mut self, serial: Serial, id: GlobalId, endpoints: Option) { @@ -423,20 +538,23 @@ impl RegistryModel { self.maybe_complete(); } - fn on_link_resolved(&mut self, serial: Serial, endpoints: LinkEndpoints) { + fn on_link_resolved(&mut self, serial: Serial, endpoints: LinkEndpoints) -> Outcome { // `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(); + Outcome::Applied + } else { + Outcome::Suppressed } } - fn on_removed(&mut self, id: GlobalId) { + fn on_removed(&mut self, id: GlobalId) -> Outcome { 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; + return Outcome::Suppressed; }; // Oldest generation first — the id may be shared during a // missed-removal window. @@ -446,10 +564,7 @@ impl RegistryModel { } match slot { Some(Slot::Node(serial)) => { - if self.nodes.remove(&serial).is_none() { - // Was still withheld — drop the obligation. - self.withheld.remove(&serial); - } + self.nodes.remove(&serial); } Some(Slot::Port(serial)) => { self.ports.remove(&serial); @@ -461,35 +576,67 @@ impl RegistryModel { 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); - } - } + Some(Slot::Device(serial)) => { + self.devices.remove(&serial); } None => { tracing::warn!(global_id = id.0, "observer: empty id slot on remove"); + return Outcome::Suppressed; } } - // A removal can drain the last obligation (a withheld node or pending - // link vanished before it resolved). + // A removal can drain the last obligation (an unbound node, a node + // withheld on a Device, or a pending link vanished before it + // resolved). self.maybe_complete(); + Outcome::Applied } 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) + /// The bound properties of the Device a node claims by global id, or + /// `None` when that claim is unresolved — which covers all three + /// fail-closed cases at once: no such Device observed, its bind still + /// outstanding, or **two live Devices sharing the recycled id**, where + /// there is no way to tell whose properties these are (v3.4 §6.1.3). + fn device_props(&self, id: GlobalId) -> Option<&DeviceProps> { + let mut found: Option = None; + for slot in self.live_ids.get(&id)? { + if let Slot::Device(serial) = slot { + if found.is_some() { + return None; // Ambiguous ⇒ unresolved ⇒ withheld. + } + found = Some(*serial); + } + } + self.devices.get(&found?)?.props.as_ref() + } + + /// Classify one node's device claim against the currently resolved + /// Devices. Recomputed per projection rather than cached at admission: + /// the inputs (this node's props, its Device's props) both change over an + /// object's lifetime now, and a cached classification is exactly the kind + /// of stale provisional answer §6.1.3 forbids. + fn classification(&self, obs: &NodeObservation) -> Classification { + let device = obs + .device_claim + .device_id + .and_then(|id| self.device_props(id)); + classify::classify(&obs.device_claim, device) } /// Every obligation that must clear before the initial graph is trusted: - /// no node withheld on an unresolved Device, no Link awaiting its bind. + /// no node awaiting its bind, 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() + if !self.pending_links.is_empty() { + return true; + } + self.nodes.values().any(|entry| match &entry.obs { + None => true, + Some(obs) => matches!(self.classification(obs), Classification::Withhold { .. }), + }) } /// Completion needs no clock — only the sync flag and an empty obligation @@ -512,13 +659,34 @@ impl RegistryModel { if now >= self.deadline { self.readiness = Readiness::TimedOut; tracing::warn!( - withheld = self.withheld.len(), + unbound_nodes = self.unbound_node_count(), + withheld = self.withheld_node_count(), pending_links = self.pending_links.len(), "observer: readiness epoch timed out with obligations outstanding — fail closed" ); } } + /// Nodes whose bind has not delivered `info` yet — diagnostics only. + fn unbound_node_count(&self) -> usize { + self.nodes + .values() + .filter(|entry| entry.obs.is_none()) + .count() + } + + /// Nodes held out on an unresolved Device — diagnostics only. + fn withheld_node_count(&self) -> usize { + self.nodes + .values() + .filter(|entry| { + entry.obs.as_ref().is_some_and(|obs| { + matches!(self.classification(obs), Classification::Withhold { .. }) + }) + }) + .count() + } + /// 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). @@ -529,9 +697,34 @@ impl RegistryModel { } /// Project the current state into the taint engine's inputs. + /// + /// A node enters the snapshot only if its bind has delivered `info` + /// **and** its device claim classifies terminally; anything else is + /// withheld (and is already holding `graph_ready` false). pub fn project(&self) -> Projection { + let nodes: Vec = self + .nodes + .iter() + .filter_map(|(&serial, entry)| { + let obs = entry.obs.as_ref()?; + let session_device = match self.classification(obs) { + Classification::Withhold { .. } => return None, + Classification::SessionDevice => true, + Classification::NotADevice | Classification::NotSessionDevice => false, + }; + let mut props = obs.props.clone(); + props.session_device = session_device; + Some(NodeSnapshot { + serial, + id: entry.id, + name: obs.name.clone(), + role: obs.role, + props, + }) + }) + .collect(); let snapshot = GraphSnapshot::new( - self.nodes.values().cloned().collect(), + nodes, self.ports.values().cloned().collect(), self.links.values().cloned().collect(), self.clients.values().cloned().collect(), diff --git a/src/host/observer/tests.rs b/src/host/observer/tests.rs index 61db94d..87b4087 100644 --- a/src/host/observer/tests.rs +++ b/src/host/observer/tests.rs @@ -1,14 +1,16 @@ -//! Pure exit-gate coverage for the phase-3 observer core. +//! Pure exit-gate coverage for the observer core (phases 3 and 3r). //! -//! 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. +//! Five of phase 3's six exit-gate rows live here (the sixth — a live +//! create/destroy topology diff — needs the daemon and belongs to the +//! adapter), plus three of phase 3r's four (the fourth is the live +//! prop-recovery gate, likewise the adapter's). 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::classify::{Classification, DeviceClaim, DeviceProps, classify}; use super::pulse_pid; use super::*; use crate::host::taint::snapshot::{ @@ -32,6 +34,9 @@ fn no_device() -> DeviceClaim { DeviceClaim::default() } +/// A node-side device claim as the *node's* bound `info` reports it: the +/// factory name (node-only) plus the `device.api`/`alsa.driver_name` copies +/// that some PipeWire/WirePlumber pairings make and some do not. fn hw_claim(device_id: u32, api: &str, factory: &str) -> DeviceClaim { DeviceClaim { device_id: Some(gid(device_id)), @@ -41,28 +46,92 @@ fn hw_claim(device_id: u32, api: &str, factory: &str) -> DeviceClaim { } } -/// 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(), - }) +/// The same claim with **neither** value copied onto the node — the common +/// case on PipeWire ≥ 1.2.6 with WirePlumber < 0.5.13, where only the bound +/// Device knows. +fn node_only_claim(device_id: u32, factory: &str) -> DeviceClaim { + DeviceClaim { + device_id: Some(gid(device_id)), + device_api: None, + factory_name: Some(factory.to_string()), + alsa_driver_name: None, + } } -/// 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}")), +/// A real sound card's bound Device props. +fn alsa_device() -> DeviceProps { + DeviceProps { + device_api: Some("alsa".to_string()), + alsa_driver_name: Some("snd_hda_intel".to_string()), + } +} + +fn device_with(api: Option<&str>, driver: Option<&str>) -> DeviceProps { + DeviceProps { + device_api: api.map(str::to_owned), + alsa_driver_name: driver.map(str::to_owned), + } +} + +fn obs(name: &str, role: MediaRole, claim: DeviceClaim) -> NodeObservation { + NodeObservation { + name: Some(name.to_string()), role, props: NodeProps::default(), device_claim: claim, - }) + } +} + +/// Announce a node and deliver its bind `info` — the settled state most tests +/// want. The two steps are driven separately only where the gap itself is +/// under test. +fn add_node(m: &mut RegistryModel, serial: u64, id: u32, observation: NodeObservation) { + m.apply(RegEvent::NodeAdded { + serial: ser(serial), + id: gid(id), + }); + m.apply(RegEvent::NodeInfo { + serial: ser(serial), + observation, + }); +} + +/// A `Stream/Output/Audio` node with no backing Device — admitted as soon as +/// its `info` lands. +fn add_stream_out(m: &mut RegistryModel, serial: u64, id: u32) { + add_node( + m, + serial, + id, + obs( + &format!("stream-{id}"), + MediaRole::StreamOutput, + no_device(), + ), + ); +} + +/// A node backed by a Device (withheld until that Device's `info` lands). +fn add_device_node( + m: &mut RegistryModel, + serial: u64, + id: u32, + role: MediaRole, + claim: DeviceClaim, +) { + add_node(m, serial, id, obs(&format!("dev-node-{id}"), role, claim)); +} + +/// Announce a Device and deliver its bind `info`. +fn add_device(m: &mut RegistryModel, serial: u64, id: u32, props: DeviceProps) { + m.apply(RegEvent::DeviceAdded { + serial: ser(serial), + id: gid(id), + }); + m.apply(RegEvent::DeviceInfo { + serial: ser(serial), + props, + }); } fn client(serial: u64, id: u32, sec_pid: Option) -> RegEvent { @@ -99,16 +168,19 @@ fn endpoints(out_node: u32, in_node: u32) -> LinkEndpoints { #[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); + assert_eq!(classify(&no_device(), None), Classification::NotADevice); + // A resolved Device is irrelevant with no device_id. + assert_eq!( + classify(&no_device(), Some(&alsa_device())), + Classification::NotADevice + ); } #[test] fn classify_unresolved_device_withholds() { let claim = hw_claim(42, "alsa", "api.alsa.pcm.sink"); assert_eq!( - classify(&claim, false), + classify(&claim, None), Classification::Withhold { device_id: gid(42) } ); } @@ -119,7 +191,7 @@ fn classify_resolved_hardware_pcm_is_session_device() { // 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), + classify(&hw_claim(7, "alsa", factory), Some(&alsa_device())), Classification::SessionDevice, "factory {factory} should be a session device" ); @@ -138,15 +210,19 @@ fn classify_invented_bluez_factories_are_not_session_devices() { factory_name: Some(factory.to_string()), alsa_driver_name: None, }; - assert_eq!(classify(&claim, true), Classification::NotSessionDevice); + let device = device_with(Some("bluez5"), None); + assert_eq!( + classify(&claim, Some(&device)), + Classification::NotSessionDevice + ); } } #[test] -fn classify_alsa_without_driver_name_fails_closed() { +fn classify_alsa_without_any_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. + // the node would otherwise slip through. Absence on BOTH sides fails closed. for factory in ["api.alsa.pcm.sink", "api.alsa.pcm.source"] { let claim = DeviceClaim { device_id: Some(gid(7)), @@ -154,8 +230,9 @@ fn classify_alsa_without_driver_name_fails_closed() { factory_name: Some(factory.to_string()), alsa_driver_name: None, }; + let device = device_with(Some("alsa"), None); assert_eq!( - classify(&claim, true), + classify(&claim, Some(&device)), Classification::NotSessionDevice, "absent driver on {factory} must fail closed" ); @@ -163,21 +240,55 @@ fn classify_alsa_without_driver_name_fails_closed() { } #[test] -fn classify_snd_aloop_is_not_a_session_device() { +fn classify_driver_from_the_device_recovers_a_real_card() { + // Round 8 / v3.5 §6.7 decision 4 — the phase-3 review's owed fix. The node + // carries neither `device.api` nor `alsa.driver_name` (PipeWire ≥ 1.2.6 + + // WirePlumber < 0.5.13); the bound Device carries both. Before this, such a + // card was over-excluded on every one of those installs. + let claim = node_only_claim(7, "api.alsa.pcm.sink"); + assert_eq!( + classify(&claim, Some(&alsa_device())), + Classification::SessionDevice, + "the Device is authoritative for device.api and alsa.driver_name" + ); +} + +#[test] +fn classify_snd_aloop_is_not_a_session_device_from_either_side() { // 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 { + // kernel hop the Link graph cannot see. It must NOT earn session_device — + // and the denylist is a UNION, so one side naming it is enough even when + // the other side disagrees (v3.5 §6.7 decision 4, safety-picked direction). + let factory = "api.alsa.pcm.sink"; + let aloop_both = ( + 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()), - }; + }, + device_with(Some("alsa"), Some("snd_aloop")), + ); + let aloop_node_only = ( + 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()), + }, + // The Device disagrees — the union still denies. + alsa_device(), + ); + let aloop_device_only = ( + node_only_claim(7, factory), + device_with(Some("alsa"), Some("snd_aloop")), + ); + for (claim, device) in [aloop_both, aloop_node_only, aloop_device_only] { assert_eq!( - classify(&claim, true), + classify(&claim, Some(&device)), Classification::NotSessionDevice, - "snd_aloop {factory} must fail closed" + "snd_aloop must fail closed whichever side names it" ); } } @@ -188,7 +299,7 @@ fn classify_resolved_but_not_hardware_pcm_fails_closed() { // 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), + classify(&hw_claim(7, "alsa", factory), Some(&alsa_device())), Classification::NotSessionDevice, "factory {factory} must not be a session device" ); @@ -196,23 +307,53 @@ fn classify_resolved_but_not_hardware_pcm_fails_closed() { } #[test] -fn classify_missing_device_api_fails_closed() { - // Even with an allowlisted factory, no device.api ⇒ not positively a - // real-backend terminal. +fn classify_missing_device_api_everywhere_fails_closed() { + // Even with an allowlisted factory and a good driver, no `device.api` on + // either side ⇒ 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); + let device = device_with(None, Some("snd_hda_intel")); + assert_eq!( + classify(&claim, Some(&device)), + Classification::NotSessionDevice + ); +} + +#[test] +fn classify_device_api_from_either_side_corroborates() { + // Presence is a union: whichever side has it, the corroborating signal is + // satisfied. + let node_side = DeviceClaim { + device_id: Some(gid(7)), + device_api: Some("alsa".to_string()), + factory_name: Some("api.alsa.pcm.sink".to_string()), + alsa_driver_name: Some("snd_hda_intel".to_string()), + }; + assert_eq!( + classify(&node_side, Some(&device_with(None, None))), + Classification::SessionDevice + ); + assert_eq!( + classify( + &node_only_claim(7, "api.alsa.pcm.sink"), + Some(&alsa_device()) + ), + Classification::SessionDevice + ); } #[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); + assert_eq!( + classify(&claim, Some(&alsa_device())), + Classification::NotSessionDevice + ); } // ========================================================================== @@ -232,29 +373,27 @@ fn clients_with(pids: &[Option]) -> Vec { #[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)); + let clients = clients_with(&[Some(4137), Some(4137), Some(9001)]); + assert_eq!(pulse_pid::candidate(&clients), 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); + let clients = clients_with(&[Some(4137), Some(4137), Some(9001), Some(9001)]); + assert_eq!(pulse_pid::candidate(&clients), None); } #[test] fn pid_candidate_missing_property_is_none() { - let cs = clients_with(&[None, None, None]); - assert_eq!(pulse_pid::candidate(&cs), None); + let clients = clients_with(&[None, None, None]); + assert_eq!(pulse_pid::candidate(&clients), 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); + // One client per pid: nothing repeats, so nothing is pipewire-pulse. + let clients = clients_with(&[Some(4137), Some(9001)]); + assert_eq!(pulse_pid::candidate(&clients), None); } #[test] @@ -267,43 +406,37 @@ fn pid_validate_matches_pulse_comm() { #[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); + // PID reuse: the repeated sec_pid is now some other process entirely. + assert_eq!(pulse_pid::validate(4137, Some("systemd")), None); + // A prefix match must not count either. + assert_eq!(pulse_pid::validate(4137, Some("pipewire-pulse-x")), None); + assert_eq!(pulse_pid::validate(4137, Some("pipewire")), 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)); + let clients = clients_with(&[Some(4137), Some(4137)]); + let candidate = pulse_pid::candidate(&clients).expect("candidate"); + assert_eq!( + pulse_pid::validate(candidate, Some("pipewire-pulse")), + 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(); @@ -346,7 +479,7 @@ fn model_pulse_pid_none_on_comm_mismatch() { #[test] fn model_adds_all_four_object_types() { let mut m = model(); - m.apply(stream_out(100, 50)); + add_stream_out(&mut m, 100, 50); m.apply(port(101, 60, 50, PortDirection::Out)); m.apply(client(102, 70, Some(4137))); m.apply(RegEvent::LinkAdded { @@ -364,7 +497,7 @@ fn model_adds_all_four_object_types() { #[test] fn model_removes_all_four_object_types() { let mut m = model(); - m.apply(stream_out(100, 50)); + add_stream_out(&mut m, 100, 50); m.apply(port(101, 60, 50, PortDirection::Out)); m.apply(client(102, 70, Some(4137))); m.apply(RegEvent::LinkAdded { @@ -388,8 +521,12 @@ fn model_removes_all_four_object_types() { #[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) }); + add_stream_out(&mut m, 100, 50); + assert_eq!( + m.apply(RegEvent::Removed { id: gid(999) }), + Outcome::Suppressed, + "a removal that changes nothing is not a projection event" + ); assert_eq!(m.project().snapshot.nodes().count(), 1); } @@ -401,8 +538,8 @@ fn model_remove_of_unknown_id_is_harmless() { 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)); + add_stream_out(&mut m, 100, 50); + add_stream_out(&mut m, 200, 50); // The snapshot fails closed: id 50 is ambiguous. let snap = m.project().snapshot; @@ -418,6 +555,85 @@ fn model_recycled_id_is_ambiguous_until_removal_accounted() { assert!(snap.node(ser(100)).is_none()); } +// ========================================================================== +// phase 3r gate row 4 — recycled Node id under churn (the bind is now on the +// hot path, so a generation mix-up costs a whole node's ownership) +// ========================================================================== + +#[test] +fn model_recycled_node_id_keeps_generations_distinct() { + let mut m = model(); + // Generation 1 on id 50, fully bound and tagged. + let owned = NodeProps { + peerspeak_owned: true, + ..Default::default() + }; + add_node( + &mut m, + 100, + 50, + NodeObservation { + name: Some("gen-1".to_string()), + role: MediaRole::Sink, + props: owned, + device_claim: no_device(), + }, + ); + // Generation 2 recycles the id before generation 1's removal is accounted + // (the missed-removal window), and is NOT peerspeak's. + add_node( + &mut m, + 200, + 50, + obs("gen-2", MediaRole::StreamOutput, no_device()), + ); + + let snap = m.project().snapshot; + assert!( + snap.node(ser(100)).unwrap().props.peerspeak_owned, + "generation 1 keeps its own props" + ); + assert!( + !snap.node(ser(200)).unwrap().props.peerspeak_owned, + "generation 2 must not inherit generation 1's ownership" + ); + + // The removal retires the oldest generation only. + m.apply(RegEvent::Removed { id: gid(50) }); + let snap = m.project().snapshot; + assert!(snap.node(ser(100)).is_none()); + let survivor = snap.node(ser(200)).expect("generation 2 survives"); + assert_eq!(survivor.name.as_deref(), Some("gen-2")); + assert!(!survivor.props.peerspeak_owned); +} + +#[test] +fn model_repeated_node_churn_on_one_id_leaves_no_residue() { + let mut m = model(); + m.apply(RegEvent::ServerSynced); + for generation in 0..8u64 { + let serial = 500 + generation; + m.apply(RegEvent::NodeAdded { + serial: ser(serial), + id: gid(50), + }); + // graph_ready drops while the bind is outstanding, every cycle. + assert!(!m.graph_ready(), "unbound node holds readiness"); + m.apply(RegEvent::NodeInfo { + serial: ser(serial), + observation: obs("churn", MediaRole::StreamOutput, no_device()), + }); + assert!(m.graph_ready(), "bound node releases readiness"); + assert_eq!(m.project().snapshot.nodes().count(), 1); + m.apply(RegEvent::Removed { id: gid(50) }); + assert_eq!(m.project().snapshot.nodes().count(), 0); + } + // Nothing accumulated: a stale generation would show up as a phantom node + // or a stuck obligation. + assert!(m.graph_ready()); + assert_eq!(m.project().snapshot.nodes().count(), 0); +} + // ========================================================================== // model — Link endpoint resolution (bind fallback path) // ========================================================================== @@ -474,10 +690,13 @@ fn model_stale_link_resolution_is_ignored() { // 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.apply(RegEvent::LinkEndpointsResolved { + serial: ser(103), + endpoints: endpoints(50, 55), + }), + Outcome::Suppressed + ); assert_eq!(m.project().snapshot.links().count(), 0); m.apply(RegEvent::ServerSynced); assert!( @@ -493,7 +712,7 @@ fn model_stale_link_resolution_is_ignored() { #[test] fn model_readiness_waits_for_sync() { let mut m = model(); - m.apply(stream_out(100, 50)); + add_stream_out(&mut m, 100, 50); assert_eq!(m.readiness(), Readiness::Waiting); assert!(!m.project().graph_ready); m.apply(RegEvent::ServerSynced); @@ -505,19 +724,33 @@ fn model_readiness_waits_for_sync() { 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( + add_device_node( + &mut m, 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) }); + // Announcing the Device is NOT enough — its own bind must land first + // (v3.5 §6.7 decision 4). + m.apply(RegEvent::DeviceAdded { + serial: ser(4200), + id: gid(42), + }); + assert_eq!( + m.readiness(), + Readiness::Waiting, + "an unbound Device resolves nothing" + ); + m.apply(RegEvent::DeviceInfo { + serial: ser(4200), + props: alsa_device(), + }); assert_eq!(m.readiness(), Readiness::Complete); assert!(m.graph_ready()); } @@ -525,12 +758,13 @@ fn model_readiness_does_not_release_with_obligation_outstanding() { #[test] fn model_readiness_times_out_fail_closed() { let mut m = model(); - m.apply(device_node( + add_device_node( + &mut m, 100, 50, MediaRole::Sink, hw_claim(42, "alsa", "api.alsa.pcm.sink"), - )); + ); m.apply(RegEvent::ServerSynced); assert_eq!(m.readiness(), Readiness::Waiting); @@ -542,7 +776,7 @@ fn model_readiness_times_out_fail_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) }); + add_device(&mut m, 4200, 42, alsa_device()); m.apply(RegEvent::ServerSynced); m.apply(RegEvent::Tick { now: 6000 }); assert_eq!(m.readiness(), Readiness::TimedOut, "timeout is sticky"); @@ -552,12 +786,13 @@ fn model_readiness_times_out_fail_closed() { #[test] fn model_tick_before_deadline_does_not_time_out() { let mut m = model(); - m.apply(device_node( + add_device_node( + &mut m, 100, 50, MediaRole::Sink, hw_claim(42, "alsa", "api.alsa.pcm.sink"), - )); + ); m.apply(RegEvent::Tick { now: 4999 }); assert_eq!(m.readiness(), Readiness::Waiting); } @@ -569,12 +804,13 @@ fn model_complete_epoch_is_sticky_but_graph_ready_is_dynamic() { 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( + add_device_node( + &mut m, 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). @@ -586,7 +822,7 @@ fn model_complete_epoch_is_sticky_but_graph_ready_is_dynamic() { 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) }); + add_device(&mut m, 4200, 42, alsa_device()); assert!(m.graph_ready()); } @@ -623,12 +859,13 @@ fn model_pending_link_drops_graph_ready_after_completion() { #[test] fn model_withheld_node_removed_clears_obligation() { let mut m = model(); - m.apply(device_node( + add_device_node( + &mut m, 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. @@ -636,6 +873,327 @@ fn model_withheld_node_removed_clears_obligation() { assert_eq!(m.readiness(), Readiness::Complete); } +// ========================================================================== +// phase 3r gate row 3 — readiness with node binds +// ========================================================================== + +#[test] +fn model_unbound_node_is_withheld_and_holds_readiness() { + // The direct consequence of v3.5 §6.7: a node whose properties have not + // arrived is an invisible VERTEX. It must not appear in the snapshot with + // default (untainted, unowned) properties, and no projection may report + // graph_ready while it is outstanding. + let mut m = model(); + m.apply(RegEvent::NodeAdded { + serial: ser(100), + id: gid(50), + }); + m.apply(RegEvent::ServerSynced); + assert_eq!(m.project().snapshot.nodes().count(), 0, "withheld entirely"); + assert_eq!(m.readiness(), Readiness::Waiting); + assert!(!m.project().graph_ready); + + m.apply(RegEvent::NodeInfo { + serial: ser(100), + observation: obs("late", MediaRole::StreamOutput, no_device()), + }); + assert_eq!(m.project().snapshot.nodes().count(), 1); + assert!(m.project().graph_ready); +} + +#[test] +fn model_unbound_node_after_completion_drops_graph_ready() { + let mut m = model(); + m.apply(RegEvent::ServerSynced); + assert!(m.graph_ready()); + m.apply(RegEvent::NodeAdded { + serial: ser(100), + id: gid(50), + }); + assert_eq!(m.readiness(), Readiness::Complete, "epoch stays sticky"); + assert!(!m.graph_ready(), "an unbound node is unresolved ancestry"); + m.apply(RegEvent::NodeInfo { + serial: ser(100), + observation: obs("bound", MediaRole::StreamOutput, no_device()), + }); + assert!(m.graph_ready()); +} + +#[test] +fn model_node_bind_that_never_resolves_times_out_sticky() { + // The accepted limitation, pinned: an unresolvable bind takes the whole + // graph down fail-closed rather than quarantining one node. + let mut m = model(); + m.apply(RegEvent::NodeAdded { + serial: ser(100), + id: gid(50), + }); + m.apply(RegEvent::ServerSynced); + m.apply(RegEvent::Tick { now: 5000 }); + assert_eq!(m.readiness(), Readiness::TimedOut); + assert!(!m.graph_ready()); + // Even a late arrival does not un-stick it. + m.apply(RegEvent::NodeInfo { + serial: ser(100), + observation: obs("very-late", MediaRole::StreamOutput, no_device()), + }); + assert_eq!(m.readiness(), Readiness::TimedOut, "timeout is sticky"); + assert!(!m.graph_ready()); +} + +#[test] +fn model_unbound_node_removed_clears_its_obligation() { + let mut m = model(); + m.apply(RegEvent::NodeAdded { + serial: ser(100), + id: gid(50), + }); + m.apply(RegEvent::ServerSynced); + assert_eq!(m.readiness(), Readiness::Waiting); + // A node that vanishes before its bind returns owes nothing. + m.apply(RegEvent::Removed { id: gid(50) }); + assert_eq!(m.readiness(), Readiness::Complete); + assert!(m.graph_ready()); +} + +#[test] +fn model_node_info_for_an_unknown_node_is_ignored() { + // A bind callback that lands after the node's removal must not resurrect + // it — there is no id index behind it any more. + let mut m = model(); + m.apply(RegEvent::NodeAdded { + serial: ser(100), + id: gid(50), + }); + m.apply(RegEvent::Removed { id: gid(50) }); + assert_eq!( + m.apply(RegEvent::NodeInfo { + serial: ser(100), + observation: obs("ghost", MediaRole::StreamOutput, no_device()), + }), + Outcome::Suppressed + ); + assert_eq!(m.project().snapshot.nodes().count(), 0); + m.apply(RegEvent::ServerSynced); + assert!(m.graph_ready()); +} + +// ========================================================================== +// phase 3r gate row 2 — lifetime property tracking and the suppression rule +// ========================================================================== + +#[test] +fn model_node_props_update_is_applied() { + // `node.link-group` set after node creation — the case a one-shot read + // would miss forever (v3.5 §6.7 decision 2). + let mut m = model(); + add_stream_out(&mut m, 100, 50); + assert_eq!( + m.project() + .snapshot + .node(ser(100)) + .unwrap() + .props + .link_group, + None + ); + + let grouped = NodeProps { + link_group: Some("loopback-2528-13".to_string()), + ..Default::default() + }; + assert_eq!( + m.apply(RegEvent::NodeInfo { + serial: ser(100), + observation: NodeObservation { + name: Some("stream-50".to_string()), + role: MediaRole::StreamOutput, + props: grouped, + device_claim: no_device(), + }, + }), + Outcome::Applied + ); + assert_eq!( + m.project() + .snapshot + .node(ser(100)) + .unwrap() + .props + .link_group + .as_deref(), + Some("loopback-2528-13") + ); +} + +#[test] +fn model_identical_node_info_is_suppressed() { + // PipeWire re-emits `info` for state changes constantly. Same props ⇒ the + // projection is provably identical ⇒ no event. Anything looser here would + // break phase 4's no-coalescing contract; anything stricter inflates the + // O5 event rate with non-events. + let mut m = model(); + let observation = obs("stream-50", MediaRole::StreamOutput, no_device()); + m.apply(RegEvent::NodeAdded { + serial: ser(100), + id: gid(50), + }); + assert_eq!( + m.apply(RegEvent::NodeInfo { + serial: ser(100), + observation: observation.clone(), + }), + Outcome::Applied, + "the first info resolves the node" + ); + let before = m.project(); + assert_eq!( + m.apply(RegEvent::NodeInfo { + serial: ser(100), + observation, + }), + Outcome::Suppressed + ); + assert_eq!(m.project(), before, "suppression means literally identical"); +} + +#[test] +fn model_node_props_update_can_flip_session_device() { + // The classification is recomputed from current inputs, not cached at + // admission: a node that starts as a null sink on a card and is later + // re-reported with a hardware-PCM factory flips. + let mut m = model(); + add_device(&mut m, 4200, 42, alsa_device()); + add_device_node( + &mut m, + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "support.null-audio-sink"), + ); + assert!( + !m.project() + .snapshot + .node(ser(100)) + .unwrap() + .props + .session_device + ); + + m.apply(RegEvent::NodeInfo { + serial: ser(100), + observation: obs( + "dev-node-50", + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + ), + }); + assert!( + m.project() + .snapshot + .node(ser(100)) + .unwrap() + .props + .session_device, + "re-classified from the updated props" + ); +} + +#[test] +fn model_device_props_update_reclassifies_its_nodes() { + // The Device side of the same rule, and the phase-3 owed fix in motion: + // the Device's first `info` carries no driver name (fails closed), a later + // one does (the card is recognised). + let mut m = model(); + m.apply(RegEvent::DeviceAdded { + serial: ser(4200), + id: gid(42), + }); + m.apply(RegEvent::DeviceInfo { + serial: ser(4200), + props: device_with(Some("alsa"), None), + }); + add_device_node( + &mut m, + 100, + 50, + MediaRole::Sink, + node_only_claim(42, "api.alsa.pcm.sink"), + ); + assert!( + !m.project() + .snapshot + .node(ser(100)) + .unwrap() + .props + .session_device, + "no driver anywhere ⇒ fail closed" + ); + + assert_eq!( + m.apply(RegEvent::DeviceInfo { + serial: ser(4200), + props: alsa_device(), + }), + Outcome::Applied + ); + assert!( + m.project() + .snapshot + .node(ser(100)) + .unwrap() + .props + .session_device, + "the Device's driver name is authoritative" + ); +} + +#[test] +fn model_identical_device_info_is_suppressed() { + let mut m = model(); + m.apply(RegEvent::DeviceAdded { + serial: ser(4200), + id: gid(42), + }); + assert_eq!( + m.apply(RegEvent::DeviceInfo { + serial: ser(4200), + props: alsa_device(), + }), + Outcome::Applied + ); + let before = m.project(); + assert_eq!( + m.apply(RegEvent::DeviceInfo { + serial: ser(4200), + props: alsa_device(), + }), + Outcome::Suppressed + ); + assert_eq!(m.project(), before); +} + +#[test] +fn model_device_info_for_an_unknown_device_is_ignored() { + let mut m = model(); + assert_eq!( + m.apply(RegEvent::DeviceInfo { + serial: ser(4200), + props: alsa_device(), + }), + Outcome::Suppressed + ); + // And it did not resolve anything: a node claiming that id stays withheld. + add_device_node( + &mut m, + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + ); + assert_eq!(m.project().snapshot.nodes().count(), 0); +} + // ========================================================================== // model — device withholding & session_device flag // ========================================================================== @@ -643,14 +1201,15 @@ fn model_withheld_node_removed_clears_obligation() { #[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( + // Device enumerated and bound before the node that references it. + add_device(&mut m, 4200, 42, alsa_device()); + add_device_node( + &mut m, 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!( @@ -667,21 +1226,23 @@ 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( + add_device_node( + &mut m, 100, 50, MediaRole::Sink, hw_claim(42, "alsa", "api.alsa.pcm.sink"), - )); - m.apply(device_node( + ); + add_device_node( + &mut m, 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) }); + add_device(&mut m, 4200, 42, alsa_device()); let snap = m.project().snapshot; assert_eq!( @@ -702,16 +1263,79 @@ fn model_withheld_node_admitted_with_correct_session_device() { #[test] fn model_withheld_filter_admitted_as_not_session_device() { let mut m = model(); - m.apply(device_node( + add_device_node( + &mut m, 200, 51, MediaRole::Sink, hw_claim(42, "alsa", "support.null-audio-sink"), - )); - m.apply(RegEvent::DeviceAdded { id: gid(42) }); + ); + add_device(&mut m, 4200, 42, alsa_device()); 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" ); } + +#[test] +fn model_removed_device_withholds_its_nodes_again() { + // A Device that goes away takes its resolution with it: the node reverts + // to withheld (fail closed) rather than keeping a stale classification. + let mut m = model(); + add_device(&mut m, 4200, 42, alsa_device()); + add_device_node( + &mut m, + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + ); + m.apply(RegEvent::ServerSynced); + assert!(m.graph_ready()); + assert_eq!(m.project().snapshot.nodes().count(), 1); + + m.apply(RegEvent::Removed { id: gid(42) }); + assert_eq!(m.project().snapshot.nodes().count(), 0, "withheld again"); + assert!(!m.graph_ready(), "and it is an obligation again"); +} + +#[test] +fn model_ambiguous_device_id_withholds_its_nodes() { + // Two live Devices on one recycled id: there is no way to know whose + // properties a claiming node should be classified against, so it is + // withheld (v3.4 §6.1.3, fail closed) rather than guessing a generation. + let mut m = model(); + add_device(&mut m, 4200, 42, alsa_device()); + add_device_node( + &mut m, + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + ); + assert_eq!(m.project().snapshot.nodes().count(), 1); + + // A second Device recycles id 42 before the first removal is accounted. + add_device( + &mut m, + 4201, + 42, + device_with(Some("alsa"), Some("snd_aloop")), + ); + assert_eq!( + m.project().snapshot.nodes().count(), + 0, + "ambiguous device id ⇒ withheld" + ); + + // Accounting for the older generation makes the claim unambiguous again — + // and it now resolves against the *surviving* Device, which is a loopback. + m.apply(RegEvent::Removed { id: gid(42) }); + let snap = m.project().snapshot; + assert_eq!(snap.nodes().count(), 1); + assert!( + !snap.node(ser(100)).unwrap().props.session_device, + "resolved against the surviving generation, not the dead one" + ); +} From 306b6014902b45f0dae576de021df77fbc6b53cb Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sat, 25 Jul 2026 18:32:27 -0400 Subject: [PATCH 2/3] =?UTF-8?q?host/observer:=20phase=203r=20adapter=20?= =?UTF-8?q?=E2=80=94=20bind=20every=20Node=20and=20Device?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The I/O half of round 8 (Codex, gpt-5.6-sol xhigh; reviewed, formatted and extended here). The adapter now reads `object.serial` and nothing else off a Node or Device global, binds the object, and takes every property the engine reasons about from its `info` props. - `BoundProxy` generalises `BoundLink` to Node/Device/Link, each holding its listener *before* its proxy so the listener is dropped first — the original Link variant had that order inverted. - Bind attachment now finds its slot by never-recycled serial rather than taking the queue's back, so nested callback activity during a bind cannot attach one generation's proxy to another's slot on a recycled id. A proxy that finds no slot is returned to the caller and dropped after the borrow ends. Removal still pops oldest-first, matching the model's `live_ids`. - An `info` is parsed and emitted on the first callback carrying props and thereafter only when `change_mask` contains PROPS. I considered emitting unconditionally and leaning on the model's suppression rule, and rejected it: if a state-only `info` ever delivered a partial props dict, that would overwrite a complete observation with an incomplete one — a worse failure than the one it guards against, and the same class as F1. - Ports stay unbound (v3.5 §6.7 / impl plan §4 item 6). Gates: exit-gate row 1 (live prop recovery) passes on this host — the tagged null sink projects `peerspeak.owned`, `pulse.module.id`, `node.passthrough`, the loopback legs share a `node.link-group`, and a real ALSA node classifies `session_device`. Added a second live test for the Device half. Row 1's `session_device` assertion is satisfied by a *union*: WirePlumber 0.5.15 copies `device.api` and `alsa.driver_name` onto ALSA nodes here, so it passes through the node fallback and would keep passing if the Device bind delivered nothing — leaving §6.7 decision 4 ungated on the development machine. The new test binds every Device and requires an ALSA card to announce both keys. Mutation-verified: breaking the Device-side driver read fails the new test while row 1 still passes, which is the gap as claimed. 195 unit + 3 live green, clippy -D warnings and fmt clean. --- src/host/observer/adapter.rs | 448 ++++++++++++++++++++++++++++++++--- 1 file changed, 409 insertions(+), 39 deletions(-) diff --git a/src/host/observer/adapter.rs b/src/host/observer/adapter.rs index fe02da4..f0e31ee 100644 --- a/src/host/observer/adapter.rs +++ b/src/host/observer/adapter.rs @@ -4,9 +4,14 @@ //! callbacks into [`RegEvent`]s, and publishes the latest [`Projection`] for //! consumers running outside the PipeWire thread. -use super::{EventKind, LinkEndpoints, Outcome, Projection, RegEvent, RegistryModel}; +use super::classify::{DeviceClaim, DeviceProps}; +use super::{ + EventKind, LinkEndpoints, NodeObservation, Outcome, Projection, RegEvent, RegistryModel, +}; use crate::host::audio::parse_object_serial; -use crate::host::taint::snapshot::{ClientSnapshot, GlobalId, PortDirection, PortSnapshot, 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}; @@ -101,14 +106,24 @@ impl Drop for RegistryObserverHandle { } } -struct BoundLink { - _proxy: pw::link::Link, - _listener: pw::link::LinkListener, +enum BoundProxy { + Node { + _listener: pw::node::NodeListener, + _proxy: pw::node::Node, + }, + Device { + _listener: pw::device::DeviceListener, + _proxy: pw::device::Device, + }, + Link { + _listener: pw::link::LinkListener, + _proxy: pw::link::Link, + }, } -#[derive(Default)] struct LiveGlobal { - bound_link: Option, + serial: Serial, + bound_proxy: Option, } struct ObserverState { @@ -139,12 +154,13 @@ impl ObserverState { } } - fn apply(&mut self, event: RegEvent) { + fn apply(&mut self, event: RegEvent) -> Outcome { // Taken before the model consumes the event: the sink is told what kind // of observation produced the projection, and deriving that from the // event itself is what stops the two from ever disagreeing. let kind = event.kind(); - let mut outcome = self.model.apply(event); + let event_outcome = self.model.apply(event); + let mut outcome = event_outcome; let candidate = self.model.pulse_pid_candidate(); if candidate != self.last_candidate { @@ -169,6 +185,7 @@ impl ObserverState { if outcome == Outcome::Applied { self.publish(kind); } + event_outcome } fn publish(&mut self, kind: EventKind) { @@ -187,40 +204,65 @@ impl ObserverState { } /// 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` + /// bound-proxy 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); + /// queues the same length per id — otherwise a phantom slot could pop + /// another generation's proxy after an id is recycled. + fn add(&mut self, serial: Serial, id: GlobalId, event: RegEvent) { + if self.apply(event) == Outcome::Applied { + self.live_globals + .entry(id) + .or_default() + .push_back(LiveGlobal { + serial, + bound_proxy: None, + }); + } } - 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 { + /// Return a proxy that could not be attached so its listener is dropped + /// after the caller releases the `RefCell` borrow. + fn attach_bound_proxy( + &mut self, + id: GlobalId, + serial: Serial, + bound_proxy: BoundProxy, + ) -> Option { + let Some(global) = self + .live_globals + .get_mut(&id) + .and_then(|globals| globals.iter_mut().find(|global| global.serial == serial)) + else { tracing::warn!( global_id = id.0, - "registry observer: link bind completed without a live global slot" + serial = serial.0, + "registry observer: bind completed without a live global slot" ); - return; + return Some(bound_proxy); }; - global.bound_link = Some(bound_link); + if global.bound_proxy.is_some() { + tracing::warn!( + global_id = id.0, + serial = serial.0, + "registry observer: live global slot already has a bound proxy" + ); + return Some(bound_proxy); + } + global.bound_proxy = Some(bound_proxy); + None } - fn remove_global(&mut self, id: GlobalId) -> Option { - let (bound_link, empty) = { + fn remove_global(&mut self, id: GlobalId) -> Option { + let (bound_proxy, 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()) + let bound_proxy = globals.pop_front().and_then(|global| global.bound_proxy); + (bound_proxy, globals.is_empty()) }; if empty { self.live_globals.remove(&id); } - bound_link + bound_proxy } } @@ -291,9 +333,59 @@ fn run_observer( else { return; }; - state_for_global - .borrow_mut() - .add(id, RegEvent::NodeAdded { serial, id }); + state_for_global.borrow_mut().add( + serial, + id, + RegEvent::NodeAdded { serial, id }, + ); + + let Some(registry) = registry_weak.upgrade() else { + return; + }; + let node: pw::node::Node = match registry.bind(obj) { + Ok(node) => node, + Err(e) => { + tracing::warn!( + node_id = obj.id, + "registry observer: failed to bind Node for properties: {e}" + ); + return; + } + }; + // This bit only recognizes the initial callback for the + // change-mask fast path. Admission vs update remains + // entirely the model's decision. + let first_info = Cell::new(true); + let state_for_info = Rc::downgrade(&state_for_global); + let listener = node + .add_listener_local() + .info(move |info| { + let Some(props) = info.props() else { + return; + }; + let first = first_info.replace(false); + if !first + && !info.change_mask().contains(pw::node::NodeChangeMask::PROPS) + { + return; + } + if let Some(state) = state_for_info.upgrade() { + state.borrow_mut().apply(RegEvent::NodeInfo { + serial, + observation: node_observation_from_props(props), + }); + } + }) + .register(); + let unattached = state_for_global.borrow_mut().attach_bound_proxy( + id, + serial, + BoundProxy::Node { + _listener: listener, + _proxy: node, + }, + ); + drop(unattached); } ObjectType::Port => { let Some(props) = obj.props.as_ref() else { @@ -332,6 +424,7 @@ fn run_observer( } }; state_for_global.borrow_mut().add( + serial, id, RegEvent::PortAdded(PortSnapshot { serial, @@ -356,6 +449,7 @@ fn run_observer( return; }; state_for_global.borrow_mut().add( + serial, id, RegEvent::ClientAdded(ClientSnapshot { serial, @@ -381,9 +475,58 @@ fn run_observer( else { return; }; - state_for_global - .borrow_mut() - .add(id, RegEvent::DeviceAdded { serial, id }); + state_for_global.borrow_mut().add( + serial, + id, + RegEvent::DeviceAdded { serial, id }, + ); + + let Some(registry) = registry_weak.upgrade() else { + return; + }; + let device: pw::device::Device = match registry.bind(obj) { + Ok(device) => device, + Err(e) => { + tracing::warn!( + device_id = obj.id, + "registry observer: failed to bind Device for properties: {e}" + ); + return; + } + }; + let first_info = Cell::new(true); + let state_for_info = Rc::downgrade(&state_for_global); + let listener = device + .add_listener_local() + .info(move |info| { + let Some(props) = info.props() else { + return; + }; + let first = first_info.replace(false); + if !first + && !info + .change_mask() + .contains(pw::device::DeviceChangeMask::PROPS) + { + return; + } + if let Some(state) = state_for_info.upgrade() { + state.borrow_mut().apply(RegEvent::DeviceInfo { + serial, + props: device_props_from_props(props), + }); + } + }) + .register(); + let unattached = state_for_global.borrow_mut().attach_bound_proxy( + id, + serial, + BoundProxy::Device { + _listener: listener, + _proxy: device, + }, + ); + drop(unattached); } ObjectType::Link => { let Some(props) = obj.props.as_ref() else { @@ -399,6 +542,7 @@ fn run_observer( }; let endpoints = link_endpoints_from_props(props); state_for_global.borrow_mut().add( + serial, id, RegEvent::LinkAdded { serial, @@ -445,24 +589,26 @@ fn run_observer( } }) .register(); - state_for_global.borrow_mut().attach_bound_link( + let unattached = state_for_global.borrow_mut().attach_bound_proxy( id, - BoundLink { - _proxy: link, + serial, + BoundProxy::Link { _listener: listener, + _proxy: link, }, ); + drop(unattached); } _ => {} } }) .global_remove(move |id| { let id = GlobalId(id); - let bound_link = state_for_remove.borrow_mut().remove_global(id); + let bound_proxy = state_for_remove.borrow_mut().remove_global(id); state_for_remove .borrow_mut() .apply(RegEvent::Removed { id }); - drop(bound_link); + drop(bound_proxy); }) .register(); @@ -506,6 +652,45 @@ fn truthy(value: Option<&str>) -> bool { value.is_some_and(|value| value != "false" && value != "0") } +fn node_observation_from_props(props: &pw::spa::utils::dict::DictRef) -> NodeObservation { + NodeObservation { + name: props.get("node.name").map(str::to_string), + role: MediaRole::parse(props.get("media.class")), + props: NodeProps { + peerspeak_owned: truthy(props.get("peerspeak.owned")), + pulse_module_id: props + .get("pulse.module.id") + .and_then(|value| value.parse::().ok()), + link_group: props.get("node.link-group").map(str::to_string), + client_id: props + .get("client.id") + .and_then(|value| value.parse::().ok()) + .map(GlobalId), + process_id: props + .get("application.process.id") + .and_then(|value| value.parse::().ok()), + passthrough: truthy(props.get("node.passthrough")), + session_device: false, + }, + device_claim: DeviceClaim { + device_id: props + .get("device.id") + .and_then(|value| value.parse::().ok()) + .map(GlobalId), + device_api: props.get("device.api").map(str::to_string), + factory_name: props.get("factory.name").map(str::to_string), + alsa_driver_name: props.get("alsa.driver_name").map(str::to_string), + }, + } +} + +fn device_props_from_props(props: &pw::spa::utils::dict::DictRef) -> DeviceProps { + DeviceProps { + device_api: props.get("device.api").map(str::to_string), + alsa_driver_name: props.get("alsa.driver_name").map(str::to_string), + } +} + fn link_endpoints_from_props(props: &pw::spa::utils::dict::DictRef) -> Option { let output_node = props.get("link.output.node")?.parse::().ok()?; let input_node = props.get("link.input.node")?.parse::().ok()?; @@ -606,6 +791,191 @@ mod tests { .any(|node| node.name.as_deref() == Some(name)) } + /// Phase 3r exit-gate row 1, the Device half — and the reason it needs its + /// own test. + /// + /// `live_bound_properties_recover_node_and_device_inputs` asserts + /// `session_device`, which the classifier grants on a **union**: + /// `device.api` and `alsa.driver_name` may come from the bound Device *or* + /// from the node's own copies. On this host (WirePlumber 0.5.15 ≥ 0.5.13) + /// the session manager *does* copy both onto ALSA nodes, so that assertion + /// passes through the node fallback and would keep passing if the Device + /// bind delivered nothing at all — leaving v3.5 §6.7 decision 4, the whole + /// authoritative path, ungated on the machine we develop on. + /// + /// So assert the Device side directly: bind every Device global and require + /// that at least one ALSA card announces **both** keys on its `info` props. + /// A failure here means the fix for the phase-3 review's owed finding (a + /// real card over-excluded on installs that do not copy `alsa.*` onto the + /// node) rests on nothing. + #[test] + #[ignore = "needs live pipewire"] + fn live_device_bind_carries_api_and_driver_name() { + pw::init(); + let main_loop = pw::main_loop::MainLoopRc::new(None).expect("pw main loop"); + let context = pw::context::ContextRc::new(&main_loop, None).expect("pw context"); + let core = context.connect_rc(None).expect("pw core connect"); + let registry = core.get_registry_rc().expect("pw registry"); + + // Devices bound off the registry, each holding its proxy + listener so + // the callback lives long enough to fire, exactly as the adapter does. + let bound: Rc>> = + Rc::new(RefCell::new(Vec::new())); + let observed: Rc>> = Rc::new(RefCell::new(Vec::new())); + + let bound_for_global = Rc::clone(&bound); + let observed_for_global = Rc::clone(&observed); + let registry_weak = registry.downgrade(); + let _listener = registry + .add_listener_local() + .global(move |obj| { + if obj.type_ != ObjectType::Device { + return; + } + let Some(registry) = registry_weak.upgrade() else { + return; + }; + let Ok(device) = registry.bind::(obj) else { + return; + }; + let observed_for_info = Rc::clone(&observed_for_global); + let listener = device + .add_listener_local() + .info(move |info| { + if let Some(props) = info.props() { + observed_for_info + .borrow_mut() + .push(device_props_from_props(props)); + } + }) + .register(); + bound_for_global.borrow_mut().push((device, listener)); + }) + .register(); + + // Two seconds is the same budget the observer gives its own binds. + let main_loop_for_timer = main_loop.clone(); + let timer = main_loop + .loop_() + .add_timer(move |_| main_loop_for_timer.quit()); + timer + .update_timer(Some(Duration::from_secs(2)), None) + .into_result() + .expect("arm the test deadline"); + main_loop.run(); + + let observed = observed.borrow(); + assert!( + !observed.is_empty(), + "no Device delivered info props at all — the Device bind path is dead" + ); + assert!( + observed.iter().any(|props| { + props.device_api.as_deref() == Some("alsa") && props.alsa_driver_name.is_some() + }), + "no bound Device carried both device.api=alsa and alsa.driver_name; \ + observed: {observed:?}" + ); + } + + // Phase 3r exit-gate row 1: failure means the observation boundary regressed. + #[test] + #[ignore = "needs live pipewire"] + fn live_bound_properties_recover_node_and_device_inputs() { + pw::init(); + let observer = RegistryObserverHandle::spawn().expect("observer thread must spawn"); + wait_for(&observer, |projection| projection.graph_ready); + + let unique = format!("pixelpass_observer_props_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}"), + "sink_properties=peerspeak.owned=true node.passthrough=true".to_string(), + ], + ); + let null_sink_id = null_sink.id.expect("null-sink module must have an id"); + 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}"), + ], + ); + + let projection = wait_for(&observer, |projection| { + projection.graph_ready + && has_node(projection, &unique) + && has_node(projection, &capture_name) + && has_node(projection, &playback_name) + }); + let tagged_sink = projection + .snapshot + .nodes() + .find(|node| node.name.as_deref() == Some(&unique)) + .expect("tagged null sink must be projected"); + assert!(tagged_sink.props.peerspeak_owned); + assert!(tagged_sink.props.passthrough); + assert_eq!( + tagged_sink.props.pulse_module_id, + Some(u64::from(null_sink_id)) + ); + + let capture = projection + .snapshot + .nodes() + .find(|node| node.name.as_deref() == Some(&capture_name)) + .expect("loopback capture leg must be projected"); + let playback = projection + .snapshot + .nodes() + .find(|node| node.name.as_deref() == Some(&playback_name)) + .expect("loopback playback leg must be projected"); + let capture_group = capture + .props + .link_group + .as_ref() + .expect("loopback capture leg must carry node.link-group"); + let playback_group = playback + .props + .link_group + .as_ref() + .expect("loopback playback leg must carry node.link-group"); + assert_eq!(capture_group, playback_group); + assert!( + projection + .snapshot + .nodes() + .any(|node| node.props.process_id.is_some()), + "at least one projected node must carry application.process.id" + ); + let session_device = projection.snapshot.nodes().find(|node| { + node.props.session_device + && (node + .name + .as_deref() + .is_some_and(|name| name.contains("alsa")) + || matches!(node.role, MediaRole::Sink | MediaRole::Source)) + }); + assert!( + session_device.is_some(), + "a named ALSA or Audio/Sink/Audio/Source node must classify as a session device" + ); + assert!(projection.graph_ready); + + loopback.unload(); + null_sink.unload(); + wait_for(&observer, |projection| { + !has_node(projection, &unique) + && !has_node(projection, &capture_name) + && !has_node(projection, &playback_name) + }); + } + #[test] #[ignore = "needs live pipewire"] fn live_topology_diff_tracks_null_sink_and_loopback() { From 471b8221ff0e477a8b21451a2bec8ae4248a803e Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sat, 25 Jul 2026 18:48:53 -0400 Subject: [PATCH 3/3] host/observer: address the Codex phase-3r review (2 fixes, both verified) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex's adversarial review of the pure core found no *certain* P1. Two findings taken, both mutation-verified (the fix reverted, the intended test dies, nothing else moves): **F2, certain, P2 — `device_props` tested the wrong kind of ambiguity.** It required exactly one live *Device* on the claimed id rather than exactly one live *global*. With `[Device, Port]` on one id — a missed removal, the same precondition as every other recycled-id hazard — it kept answering from the older Device, so a node claiming that id held a stale `session_device = true`. That flag strips the node's owner keys and its fail-closed backstop, so a forwarder wearing it can put its output leg back on the eligible side. Now: one slot total, and it must be the Device. **F3, worth checking, P3 — `device.api` was corroborating by presence.** `device.api=v4l2` under `factory.name=api.alsa.pcm.sink` satisfied the positive classifier. No truthful configuration produces that pair, which is the argument for reading it as an observation gone wrong rather than as corroboration. The API must now equal the one the factory allowlist is written for, an empty value is not a value, and the two sides disagreeing fails closed. Tied to the allowlist being ALSA-only via a named constant. Two findings NOT fixed here, both pre-existing and neither introduced by round 8 — raised to the design doc instead: - **P1, worth checking: hardware playback-to-capture paths** (Stereo Mix, Digital Loopback) on a card whose driver is an ordinary `snd_hda_intel`. Both its sink and source classify `session_device`, taint cannot cross the hardware hop, and a capture app reading that source can re-emit the call. This is `snd_aloop` again in a form the driver name cannot detect; distinguishing it needs ALSA control inspection, which is a design change and a new I/O surface, not a local fix. - **P3: the 2 s readiness budget** can in principle never see an obligation-free instant under sustained startup churn, and `TimedOut` is sticky by design, so the process would be silent for its lifetime. Measured here: readiness at ~3 ms with 19 binds, so the margin is three orders of magnitude — but it wants a calibration argument, not a guess. 197 unit + 3 live green, clippy -D warnings and fmt clean. --- src/host/observer/classify.rs | 21 ++++++++- src/host/observer/mod.rs | 36 ++++++++------ src/host/observer/tests.rs | 88 +++++++++++++++++++++++++++++++++++ 3 files changed, 130 insertions(+), 15 deletions(-) diff --git a/src/host/observer/classify.rs b/src/host/observer/classify.rs index 299d2d2..62e3fd2 100644 --- a/src/host/observer/classify.rs +++ b/src/host/observer/classify.rs @@ -64,6 +64,11 @@ const HARDWARE_PCM_FACTORIES: &[&str] = &[ "api.alsa.pcm.source", ]; +/// The `device.api` every entry in [`HARDWARE_PCM_FACTORIES`] belongs to. +/// A single value rather than a list, because the allowlist is ALSA-only; +/// this constant is the thing to change when that stops being true. +const HARDWARE_PCM_API: &str = "alsa"; + /// 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 @@ -199,8 +204,20 @@ pub fn classify(claim: &DeviceClaim, device: Option<&DeviceProps>) -> Classifica .flatten() .any(|d| NON_TERMINAL_ALSA_DRIVERS.contains(&d)); let driver_ok = driver.is_some() && !driver_denied; - let api_present = device.device_api.is_some() || claim.device_api.is_some(); - let is_hardware_pcm = api_present && on_factory_allowlist && driver_ok; + // The API must positively be the one the factory allowlist is written + // for, not merely present (Codex phase-3r review, finding 3). "Present" + // admitted `device.api=v4l2` alongside `factory.name=api.alsa.pcm.sink` + // — a contradiction no truthful configuration produces, which is exactly + // why it should be read as an observation gone wrong rather than as + // corroboration. Disagreement between the two sides fails closed for the + // same reason. ⚠️ Tied to [`HARDWARE_PCM_FACTORIES`] being ALSA-only: + // adding a BlueZ factory means allowing `bluez5` here too. + let api_ok = match (device.device_api.as_deref(), claim.device_api.as_deref()) { + (Some(from_device), Some(from_node)) if from_device != from_node => false, + (Some(api), _) | (None, Some(api)) => api == HARDWARE_PCM_API, + (None, None) => false, + }; + let is_hardware_pcm = api_ok && on_factory_allowlist && driver_ok; if is_hardware_pcm { Classification::SessionDevice } else { diff --git a/src/host/observer/mod.rs b/src/host/observer/mod.rs index 108ba50..87f5a27 100644 --- a/src/host/observer/mod.rs +++ b/src/host/observer/mod.rs @@ -596,21 +596,31 @@ impl RegistryModel { } /// The bound properties of the Device a node claims by global id, or - /// `None` when that claim is unresolved — which covers all three - /// fail-closed cases at once: no such Device observed, its bind still - /// outstanding, or **two live Devices sharing the recycled id**, where - /// there is no way to tell whose properties these are (v3.4 §6.1.3). + /// `None` when that claim is unresolved — which covers every fail-closed + /// case at once: no such Device observed, its bind still outstanding, or + /// **the id claimed by more than one live global**, where there is no way + /// to tell whose properties these are (v3.4 §6.1.3). + /// + /// ⚠️ The ambiguity test is "**exactly one** live global holds this id", + /// not "exactly one live *Device*" (Codex phase-3r review, finding 2). + /// The weaker test looks equivalent and is not: with `[Device, Port]` on + /// one id — a missed removal, the same precondition as every other + /// recycled-id hazard — it keeps answering with the older Device's + /// properties, so a node claiming that id holds a stale + /// `session_device = true`. That flag *removes* the node's owner keys and + /// its fail-closed backstop, so a forwarder wearing it can put its output + /// leg back on the eligible side: echo, from a lookup that was merely + /// looking at the wrong object type. fn device_props(&self, id: GlobalId) -> Option<&DeviceProps> { - let mut found: Option = None; - for slot in self.live_ids.get(&id)? { - if let Slot::Device(serial) = slot { - if found.is_some() { - return None; // Ambiguous ⇒ unresolved ⇒ withheld. - } - found = Some(*serial); - } + let slots = self.live_ids.get(&id)?; + if slots.len() != 1 { + return None; // Ambiguous ⇒ unresolved ⇒ withheld. } - self.devices.get(&found?)?.props.as_ref() + let Slot::Device(serial) = slots.front()? else { + // The id is live, but it is not a Device any more. + return None; + }; + self.devices.get(serial)?.props.as_ref() } /// Classify one node's device claim against the currently resolved diff --git a/src/host/observer/tests.rs b/src/host/observer/tests.rs index 87b4087..3b38e74 100644 --- a/src/host/observer/tests.rs +++ b/src/host/observer/tests.rs @@ -346,6 +346,53 @@ fn classify_device_api_from_either_side_corroborates() { ); } +#[test] +fn classify_contradictory_api_fails_closed() { + // Codex phase-3r review, finding 3. A `device.api` that is merely + // *present* is not corroboration: `v4l2` under an ALSA PCM factory is a + // contradiction, and the safe reading of a contradiction is "an + // observation went wrong", not "close enough". + 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, + Some(&device_with(Some("v4l2"), Some("snd_hda_intel"))) + ), + Classification::NotSessionDevice, + "a non-ALSA api under an ALSA factory must not corroborate" + ); + + // The two sides disagreeing fails closed for the same reason. + let disagreeing = DeviceClaim { + device_id: Some(gid(7)), + device_api: Some("bluez5".to_string()), + factory_name: Some("api.alsa.pcm.sink".to_string()), + alsa_driver_name: Some("snd_hda_intel".to_string()), + }; + assert_eq!( + classify(&disagreeing, Some(&alsa_device())), + Classification::NotSessionDevice, + "node and Device naming different APIs must fail closed" + ); + + // An empty value is not a value. + let empty = DeviceClaim { + device_id: Some(gid(7)), + device_api: Some(String::new()), + factory_name: Some("api.alsa.pcm.sink".to_string()), + alsa_driver_name: Some("snd_hda_intel".to_string()), + }; + assert_eq!( + classify(&empty, Some(&device_with(None, Some("snd_hda_intel")))), + Classification::NotSessionDevice + ); +} + #[test] fn classify_allowlist_is_exact_not_substring() { // A factory that merely *contains* an allowlisted name must not pass. @@ -1300,6 +1347,47 @@ fn model_removed_device_withholds_its_nodes_again() { assert!(!m.graph_ready(), "and it is an obligation again"); } +#[test] +fn model_device_id_shared_with_another_object_type_withholds() { + // Codex phase-3r review, finding 2 (certain). The ambiguity test has to + // be "exactly one live global holds this id", not "exactly one live + // *Device*": a Port recycling the id is the same missed-removal + // condition, and answering from the older Device leaves the claiming + // node wearing a stale `session_device = true` — which strips its owner + // keys and backstop, the difference between over-exclusion and echo. + let mut m = model(); + add_device(&mut m, 4200, 42, alsa_device()); + add_device_node( + &mut m, + 100, + 50, + MediaRole::Sink, + hw_claim(42, "alsa", "api.alsa.pcm.sink"), + ); + assert!( + m.project() + .snapshot + .node(ser(100)) + .unwrap() + .props + .session_device + ); + + // A Port appears on the recycled id 42. + m.apply(port(4300, 42, 50, PortDirection::In)); + assert_eq!( + m.project().snapshot.nodes().count(), + 0, + "a contested device id resolves nothing" + ); + assert!(!m.graph_ready(), "and it is an outstanding obligation"); + + // Accounting for the Device's removal leaves the Port holding the id + // alone — still not a Device, so the node stays withheld. + m.apply(RegEvent::Removed { id: gid(42) }); + assert_eq!(m.project().snapshot.nodes().count(), 0); +} + #[test] fn model_ambiguous_device_id_withholds_its_nodes() { // Two live Devices on one recycled id: there is no way to know whose