From b3d71724ae9dc0547fa0d48d701070ea1b5c1ef6 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sat, 25 Jul 2026 18:17:48 -0400 Subject: [PATCH] =?UTF-8?q?host/observer:=20phase=203r=20pure=20core=20?= =?UTF-8?q?=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" + ); +}