host/taint: pure graph model + taint engine (phase 2)

Implements design v3.4 §6.1–§6.1.3 behind a fixture test surface. No
PipeWire types in any signature; nothing here links against libpipewire.
Not wired into anything yet — phase 3's registry observer is what will
feed it, so the module is `#![allow(dead_code)]` for now.

    evaluate(&GraphSnapshot, &ExclusionCtx, &StickyState)
        -> (Decisions, StickyState)

- snapshot.rs: owned Node/Port/Link/Client model keyed on `Serial`
  (object.serial, 64-bit, identity) with `GlobalId` retained strictly as
  a snapshot-local lookup key. Two live objects claiming one id resolve
  as `Ambiguous`, which fails closed.
- owner.rs: the owner bridge — the key union (link-group, pulse.module.id,
  client.id, application.process.id) with equality-not-first-present
  semantics, transitive union-find components, and both suppression rules.
- mod.rs: monotone fixpoint over link edges, the conditional owner bridge
  (gated on the tainted member being one that *receives* audio) and the
  unbounded-owner backstop, then sticky merge. Stable `Reason` codes with
  an explicit priority so the reported reason never depends on traversal
  order.

Three judgement calls that go beyond what v3.4 spells out, all flagged
in the source:

1. Coarse keys (client.id, application.process.id) may not bridge
   device-role nodes. Every ALSA device is created by one WirePlumber
   process, so they share a client and a PID; peerspeak's playback taints
   the default sink on every recompute, and without this rule that taint
   reaches the microphone source and then every app holding a mic loses
   its playback — the §6.1.1 catastrophe by another route.
2. "Owner is bounded" is not "has a usable key": client.id alone does not
   bound an owner (the measured GStreamer split-client refutation), so
   the fail-closed backstop keys on strong keys or a usable PID.
3. Sticky entries record a reason per node rather than one per owner, so
   a forwarder's output leg keeps `tainted-owner-bridge` instead of
   inheriting its input leg's `tainted-upstream`.

32 fixture tests, each asserting an exact partition of the full candidate
universe rather than spot-checking named nodes: v3.4 §12's matrix, the
impl plan's degenerate-snapshot boundary, and the eligible half of every
scenario so an exclude-everything build fails.

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

src/host/taint/fixture.rs

@@ -0,0 +1,282 @@
+//! Synthetic graph builders for the taint-engine tests.
+//!
+//! Serials are handed out monotonically and never reused, exactly as
+//! PipeWire does; global ids are handed out separately and **may be reused
+//! on purpose**, which is what the recycling tests need.
+
+use std::collections::BTreeMap;
+
+use super::snapshot::{
+    ClientSnapshot, GlobalId, GraphSnapshot, LinkSnapshot, MediaRole, NodeProps, NodeSnapshot,
+    PortDirection, PortSnapshot, Serial,
+};
+
+/// pipewire-pulse's PID, as measured on the target machine.
+pub const PULSE_PID: u32 = 2541;
+/// WirePlumber's PID — one process owning every device node on the box.
+pub const SESSION_PID: u32 = 900;
+
+/// A node's identity in a fixture: what tests pass around.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub struct NodeRef {
+    pub serial: Serial,
+    pub id: GlobalId,
+}
+
+#[derive(Default)]
+pub struct Graph {
+    next_serial: u64,
+    next_id: u32,
+    nodes: Vec<NodeSnapshot>,
+    ports: Vec<PortSnapshot>,
+    links: Vec<LinkSnapshot>,
+    clients: Vec<ClientSnapshot>,
+    /// One client connection per process / per module, which is what the
+    /// live graph looks like. Tests that need the *split*-client shape
+    /// (GStreamer opens one per stream) pass clients explicitly instead.
+    client_by_app: BTreeMap<u32, GlobalId>,
+    client_by_module: BTreeMap<u64, GlobalId>,
+    session_client: Option<GlobalId>,
+}
+
+impl Graph {
+    pub fn new() -> Self {
+        Self {
+            // Start past u32::MAX so every fixture also exercises the phase
+            // 0a widening: a serial that a u32 model would have truncated.
+            next_serial: u64::from(u32::MAX) + 1,
+            next_id: 1,
+            ..Self::default()
+        }
+    }
+
+    fn serial(&mut self) -> Serial {
+        self.next_serial += 1;
+        Serial(self.next_serial)
+    }
+
+    fn id(&mut self) -> GlobalId {
+        self.next_id += 1;
+        GlobalId(self.next_id)
+    }
+
+    /// A client object. `sec_pid` is `pipewire.sec.pid` — pipewire-pulse's
+    /// PID for Pulse-emulated clients.
+    pub fn client(&mut self, sec_pid: Option<u32>) -> GlobalId {
+        let serial = self.serial();
+        let id = self.id();
+        self.clients.push(ClientSnapshot {
+            serial,
+            id,
+            sec_pid,
+        });
+        id
+    }
+
+    /// The client connection an ordinary process holds — one per PID,
+    /// created on demand.
+    pub fn client_of_app(&mut self, pid: u32) -> GlobalId {
+        if let Some(id) = self.client_by_app.get(&pid) {
+            return *id;
+        }
+        let id = self.client(Some(PULSE_PID));
+        self.client_by_app.insert(pid, id);
+        id
+    }
+
+    /// An ordinary application stream: its own client, its own PID.
+    pub fn app_node(&mut self, name: &str, role: MediaRole, pid: u32) -> NodeRef {
+        let client = self.client_of_app(pid);
+        self.node(name, role, app(client, pid))
+    }
+
+    /// A leg of a pactl-loaded module: one client per module, and the
+    /// node's `application.process.id` is **pipewire-pulse's own**, because
+    /// pipewire-pulse genuinely is the client.
+    pub fn module_node(&mut self, name: &str, role: MediaRole, module: u64) -> NodeRef {
+        let client = match self.client_by_module.get(&module) {
+            Some(id) => *id,
+            None => {
+                let id = self.client(Some(PULSE_PID));
+                self.client_by_module.insert(module, id);
+                id
+            }
+        };
+        self.node(name, role, pulse_module(client, module, PULSE_PID))
+    }
+
+    /// A leg joined to its siblings by `node.link-group` — loopback,
+    /// filter-chain, echo-cancel.
+    pub fn group_node(&mut self, name: &str, role: MediaRole, group: &str, pid: u32) -> NodeRef {
+        let client = self.client_of_app(pid);
+        self.node(name, role, link_group(group, client, pid))
+    }
+
+    /// A device node as the session manager creates it: no strong key, and
+    /// WirePlumber's client and PID — shared with every other device.
+    pub fn device_node(&mut self, name: &str, role: MediaRole) -> NodeRef {
+        let session = match self.session_client {
+            Some(id) => id,
+            None => {
+                let id = self.client(None);
+                self.session_client = Some(id);
+                id
+            }
+        };
+        self.node(name, role, device(session, SESSION_PID))
+    }
+
+    pub fn peerspeak_node(&mut self, name: &str, pid: u32) -> NodeRef {
+        let client = self.client_of_app(pid);
+        self.node(name, MediaRole::StreamOutput, peerspeak_owned(client, pid))
+    }
+
+    pub fn node(&mut self, name: &str, role: MediaRole, props: NodeProps) -> NodeRef {
+        let id = self.id();
+        self.node_with_id(name, role, id, props)
+    }
+
+    /// Force a global id — for reproducing id recycling after teardown.
+    pub fn node_with_id(
+        &mut self,
+        name: &str,
+        role: MediaRole,
+        id: GlobalId,
+        props: NodeProps,
+    ) -> NodeRef {
+        let serial = self.serial();
+        self.nodes.push(NodeSnapshot {
+            serial,
+            id,
+            name: Some(name.to_string()),
+            role,
+            props,
+        });
+        NodeRef { serial, id }
+    }
+
+    pub fn port(&mut self, node: NodeRef, direction: PortDirection, exclusive: bool) {
+        let serial = self.serial();
+        let id = self.id();
+        self.ports.push(PortSnapshot {
+            serial,
+            id,
+            node: node.id,
+            direction,
+            exclusive,
+            monitor: false,
+        });
+    }
+
+    /// A signal edge: audio flows `from → to`.
+    pub fn link(&mut self, from: NodeRef, to: NodeRef) {
+        self.link_ids(from.id, to.id);
+    }
+
+    /// A link naming raw ids, so a test can dangle an endpoint.
+    pub fn link_ids(&mut self, from: GlobalId, to: GlobalId) {
+        let serial = self.serial();
+        let id = self.id();
+        self.links.push(LinkSnapshot {
+            serial,
+            id,
+            output_node: from,
+            input_node: to,
+            output_port: None,
+            input_port: None,
+        });
+    }
+
+    /// An id that belongs to nothing — for unresolved-endpoint tests.
+    pub fn dangling_id(&mut self) -> GlobalId {
+        self.id()
+    }
+
+    pub fn build(&self) -> GraphSnapshot {
+        self.build_without(&[])
+    }
+
+    /// A later snapshot in which some nodes have gone away, along with
+    /// their ports and every link touching them. Surviving objects keep
+    /// their serials, which is what makes sticky-taint sequences testable.
+    pub fn build_without(&self, dropped: &[NodeRef]) -> GraphSnapshot {
+        let gone_serials: Vec<Serial> = dropped.iter().map(|n| n.serial).collect();
+        let nodes: Vec<NodeSnapshot> = self
+            .nodes
+            .iter()
+            .filter(|n| !gone_serials.contains(&n.serial))
+            .cloned()
+            .collect();
+        // Filter by what was *dropped*, not by what is live: a link to an id
+        // that never had a node is a dangling endpoint, and dropping those
+        // here would quietly disarm every unresolved-ancestry test.
+        let gone_ids: Vec<GlobalId> = dropped.iter().map(|n| n.id).collect();
+        GraphSnapshot::new(
+            nodes,
+            self.ports
+                .iter()
+                .filter(|p| !gone_ids.contains(&p.node))
+                .cloned()
+                .collect(),
+            self.links
+                .iter()
+                .filter(|l| !gone_ids.contains(&l.output_node) && !gone_ids.contains(&l.input_node))
+                .cloned()
+                .collect(),
+            self.clients.clone(),
+        )
+    }
+
+    /// Drop clients too — full owner teardown.
+    pub fn drop_clients(&mut self, ids: &[GlobalId]) {
+        self.clients.retain(|c| !ids.contains(&c.id));
+    }
+}
+
+/// An ordinary application stream: real PID, one client connection.
+pub fn app(client: GlobalId, pid: u32) -> NodeProps {
+    NodeProps {
+        client_id: Some(client),
+        process_id: Some(pid),
+        ..NodeProps::default()
+    }
+}
+
+/// A pactl-module-created stream: the daemon is the client, so the node's
+/// `application.process.id` is pipewire-pulse's own.
+pub fn pulse_module(client: GlobalId, module: u64, pulse_pid: u32) -> NodeProps {
+    NodeProps {
+        pulse_module_id: Some(module),
+        client_id: Some(client),
+        process_id: Some(pulse_pid),
+        ..NodeProps::default()
+    }
+}
+
+/// A PipeWire-module leg joined to its siblings by `node.link-group`
+/// (loopback, filter-chain, echo-cancel).
+pub fn link_group(group: &str, client: GlobalId, pid: u32) -> NodeProps {
+    NodeProps {
+        link_group: Some(group.to_string()),
+        client_id: Some(client),
+        process_id: Some(pid),
+        ..NodeProps::default()
+    }
+}
+
+/// A device node as the session manager creates it: no strong key, and the
+/// session manager's own client and PID — shared with every other device.
+pub fn device(session_client: GlobalId, session_pid: u32) -> NodeProps {
+    NodeProps {
+        client_id: Some(session_client),
+        process_id: Some(session_pid),
+        ..NodeProps::default()
+    }
+}
+
+pub fn peerspeak_owned(client: GlobalId, pid: u32) -> NodeProps {
+    NodeProps {
+        peerspeak_owned: true,
+        ..app(client, pid)
+    }
+}