pixelpass/src/host/observer/mod.rs

//! The registry observer's **pure core** (impl plan §4, phase 3).
//!
//! This is my half of the phase-3 split: a reducer that folds a stream of
//! typed [`RegEvent`]s into a live model of the PipeWire graph and projects
//! the [`GraphSnapshot`] + context the taint engine (phase 2) consumes. **No
//! PipeWire types appear here** — the I/O adapter (Codex's half) translates
//! live registry callbacks, Link/Device binds, `/proc` reads, and the
//! `core.sync`/`done` round-trip into these events and feeds them in. Every
//! test in this module builds the event stream by hand.
//!
//! Three things this core is shaped to get right, each an exit-gate row:
//!
//! - **Removal by recycled id.** `global_remove` names only a 32-bit global
//!   id, and those recycle. The model keeps an insertion-ordered index per id
//!   so a removal accounts for the *oldest* generation first, and the
//!   snapshot projection treats any id still claimed by two live objects as
//!   [`IdLookup::Ambiguous`] — fail closed (v3.4 §6.1.3).
//! - **The readiness epoch.** `graph_ready` is false until the initial graph
//!   is fully observed: the server has synced **and** no binds/withheld nodes
//!   remain outstanding. A bounded timeout makes it fail closed. It gates
//!   sticky *retirement* only; withholding after completion is per-object.
//! - **Withholding on unresolved devices.** A node claiming a `device.id`
//!   whose Device we have not observed is held out of the snapshot entirely
//!   rather than admitted with a provisional `session_device` (see
//!   [`classify`]).
//!
//! **Two accepted limitations (Codex phase-3 review, findings 3 and 4), both
//! low-reachability, owed to a later hardening round:**
//!
//! - *A Link dropped for a missing `object.serial`/props is unrepresented.*
//!   The adapter drops such a global before it reaches [`RegistryModel`], so
//!   readiness can reach `Complete` while permanently omitting that Link — an
//!   invisible edge that could hide tainted ancestry. **Not reachable in
//!   practice:** PipeWire's native protocol defines `object.serial` as the
//!   unique identity every global carries, so a Link without one requires a
//!   protocol/server failure, not ordinary churn. (The live gate is
//!   consistent with this but does not *prove* it — it only counts Links the
//!   strict parser already admitted.) A full fix needs a pure
//!   "required-observation-failed" token that holds readiness false; deferred
//!   rather than built for a case that does not occur.
//! - *Removal generation ordering assumes no removal is silently lost.* On a
//!   recycled id with two live claimants, [`Self::on_removed`] retires the
//!   oldest generation first; if the *first* generation's removal was never
//!   delivered, a later removal is misattributed. PipeWire's registry does not
//!   silently drop `global_remove`, so this needs callback loss to trigger.
//!   The snapshot treats the two-claimant window as [`IdLookup::Ambiguous`]
//!   (fail closed) meanwhile.

#![allow(dead_code)] // Wired by the phase-3 adapter (Codex's half) and consumed by later phases.

pub mod adapter;
pub mod classify;
pub mod pulse_pid;

#[cfg(test)]
mod tests;

use crate::host::taint::snapshot::{
    ClientSnapshot, GlobalId, GraphSnapshot, LinkSnapshot, MediaRole, NodeProps, NodeSnapshot,
    PortSnapshot, Serial,
};
use classify::{Classification, DeviceClaim};
use std::collections::{BTreeMap, VecDeque};

/// A monotonic millisecond clock value, supplied by the adapter via
/// [`RegEvent::Tick`]. Kept as a bare integer rather than
/// [`std::time::Instant`] so the readiness timeout is deterministic in tests.
pub type Millis = u64;

/// A Node as observed off the registry, before `session_device` has been
/// decided. The adapter fills [`NodeProps`] with everything it can parse and
/// leaves `session_device` at its `false` default; the model overwrites it
/// from the [`classify`] result once the backing Device (if any) is resolved.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct NodeObservation {
    pub serial: Serial,
    pub id: GlobalId,
    pub name: Option<String>,
    pub role: MediaRole,
    pub props: NodeProps,
    pub device_claim: DeviceClaim,
}

/// The four endpoint references a Link carries. Node endpoints are required —
/// a Link with unknown nodes is useless — so this whole struct is what the
/// adapter must resolve (from the global's props if present, else by binding
/// `LinkInfoRef`, the correctness path) before a Link enters the snapshot.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct LinkEndpoints {
    pub output_node: GlobalId,
    pub input_node: GlobalId,
    pub output_port: Option<GlobalId>,
    pub input_port: Option<GlobalId>,
}

/// A typed observation of the live graph. The adapter produces these; the
/// model consumes them in [`RegistryModel::apply`].
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RegEvent {
    /// A Node global appeared. Admitted immediately unless it claims an
    /// unresolved Device (then withheld — see [`classify`]).
    NodeAdded(NodeObservation),
    /// A Port global appeared.
    PortAdded(PortSnapshot),
    /// A Client global appeared. Feeds pulse-PID derivation via `sec_pid`.
    ClientAdded(ClientSnapshot),
    /// A Device global appeared. Resolves any nodes withheld on its id.
    DeviceAdded { id: GlobalId },
    /// A Link global appeared. `endpoints` is `Some` when the global carried
    /// them (the optimisation) and `None` when the adapter must bind to learn
    /// them (the correctness path) — the latter is an outstanding obligation
    /// until a matching [`RegEvent::LinkEndpointsResolved`] arrives.
    LinkAdded {
        serial: Serial,
        id: GlobalId,
        endpoints: Option<LinkEndpoints>,
    },
    /// The bind-`LinkInfoRef` fallback resolved a Link's endpoints.
    LinkEndpointsResolved {
        serial: Serial,
        endpoints: LinkEndpoints,
    },
    /// The adapter read `/proc/<pid>/comm` (`None` = the read failed / the
    /// process is gone). Validates the pulse-PID candidate.
    ProcCommProbed { pid: u32, comm: Option<String> },
    /// Any global was removed. Only its 32-bit id is known.
    Removed { id: GlobalId },
    /// A `core.sync()` issued after the initial enumeration completed its
    /// round-trip (`done`). One half of readiness; the other is that no
    /// binds/withheld nodes are still outstanding.
    ServerSynced,
    /// A monotonic clock sample. Drives the readiness timeout only.
    Tick { now: Millis },
}

/// Which slot in the id index a live object occupies. `global_remove` gives
/// only the id, so the index remembers what each id currently holds. A Node
/// slot's serial may live in either the admitted or the withheld map.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Slot {
    Node(Serial),
    Port(Serial),
    Link(Serial),
    Client(Serial),
    Device,
}

/// The readiness epoch. A one-time transition out of [`Readiness::Waiting`];
/// both terminal states are sticky (a completed graph is not un-completed by
/// later per-object withholding, and a timed-out observer stays fail-closed
/// for its lifetime).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Readiness {
    /// The initial enumeration is still in flight.
    Waiting,
    /// The initial enumeration finished at least once (server synced with no
    /// obligations then outstanding). **Sticky** — later per-object
    /// withholding does not revert it. Note this is *not* the same as
    /// [`RegistryModel::graph_ready`], which additionally requires no *current*
    /// obligation (Codex finding 1); `Complete` only records that the epoch
    /// was reached.
    Complete,
    /// The bounded deadline passed with obligations outstanding.
    /// `graph_ready` stays false — fail closed.
    TimedOut,
}

/// The pure handoff to the taint engine: a coherent [`GraphSnapshot`] plus the
/// two context fields phase 3 owns. The caller merges these into
/// [`crate::host::taint::ExclusionCtx`] alongside `aec_module_id` (phase 4)
/// and `pixelpass_owned` (pixelpass's own tracking).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Projection {
    pub snapshot: GraphSnapshot,
    pub pipewire_pulse_pid: Option<u32>,
    pub graph_ready: bool,
}

/// The live model. Folds [`RegEvent`]s; project with [`RegistryModel::project`].
#[derive(Clone, Debug)]
pub struct RegistryModel {
    // Admitted objects, keyed by their never-recycled serial.
    nodes: BTreeMap<Serial, NodeSnapshot>,
    ports: BTreeMap<Serial, PortSnapshot>,
    links: BTreeMap<Serial, LinkSnapshot>,
    clients: BTreeMap<Serial, ClientSnapshot>,

    /// Nodes held out of the snapshot pending their Device's resolution.
    withheld: BTreeMap<Serial, NodeObservation>,
    /// Links whose endpoints the adapter is still binding; the id is kept so
    /// removal and resolution can find them.
    pending_links: BTreeMap<Serial, GlobalId>,

    /// Live Device global ids, ref-counted so a recycled id is only
    /// considered resolved while a Device actually holds it.
    resolved_devices: BTreeMap<GlobalId, usize>,

    /// Insertion-ordered holders of each live global id. `global_remove`
    /// accounts for the oldest generation first (v3.4 §6.1.3).
    live_ids: BTreeMap<GlobalId, VecDeque<Slot>>,

    /// `/proc/<pid>/comm` reads keyed by pid, for pulse-PID validation.
    probed_comm: BTreeMap<u32, Option<String>>,

    server_synced: bool,
    readiness: Readiness,
    deadline: Millis,
    last_now: Millis,
}

impl RegistryModel {
    /// `now` seeds the clock; `timeout` is the readiness budget. The deadline
    /// is `now + timeout`; a [`RegEvent::Tick`] at or past it while still
    /// [`Readiness::Waiting`] fails the epoch closed.
    pub fn new(now: Millis, timeout: Millis) -> Self {
        Self {
            nodes: BTreeMap::new(),
            ports: BTreeMap::new(),
            links: BTreeMap::new(),
            clients: BTreeMap::new(),
            withheld: BTreeMap::new(),
            pending_links: BTreeMap::new(),
            resolved_devices: BTreeMap::new(),
            live_ids: BTreeMap::new(),
            probed_comm: BTreeMap::new(),
            server_synced: false,
            readiness: Readiness::Waiting,
            deadline: now.saturating_add(timeout),
            last_now: now,
        }
    }

    pub fn readiness(&self) -> Readiness {
        self.readiness
    }

    /// Whether the graph is trustworthy enough to make eligibility and sticky
    /// **retirement** decisions right now.
    ///
    /// This is **dynamic**, not the sticky [`Readiness::Complete`] flag: it is
    /// true only when the initial enumeration has completed **and** there are
    /// no current obligations outstanding (a node withheld on an unresolved
    /// Device, or a Link still being bound). The distinction is the fix for
    /// Codex phase-3 review finding 1: a Link whose endpoints are still
    /// resolving is an **invisible edge** — it is absent from the snapshot,
    /// not merely dangling — so a decision made while one exists can miss real
    /// tainted ancestry and wrongly report a candidate eligible. Unresolved
    /// ancestry ⇒ fail closed is the governing invariant (v3.4 §6.1), and an
    /// unresolved Link is unresolved ancestry, so `graph_ready` must drop back
    /// to false whenever one is pending — even after the initial epoch.
    ///
    /// [`Readiness::Complete`] stays sticky (it records that the initial
    /// enumeration happened, for logging and to distinguish "not started" from
    /// "momentarily churning"); `graph_ready` layers the dynamic obligation
    /// check on top. Downstream (phase 6) may debounce the brief blips a
    /// normal Link bind causes; the observer's job is to report the truth.
    pub fn graph_ready(&self) -> bool {
        matches!(self.readiness, Readiness::Complete) && !self.obligations_outstanding()
    }

    /// The pulse-PID candidate the adapter should be probing (`None` = no
    /// repeated `sec_pid`, nothing to probe). Exposed so the adapter re-probes
    /// only when the candidate changes.
    pub fn pulse_pid_candidate(&self) -> Option<u32> {
        let clients: Vec<ClientSnapshot> = self.clients.values().cloned().collect();
        pulse_pid::candidate(&clients)
    }

    /// Fold one observation into the model.
    pub fn apply(&mut self, event: RegEvent) {
        match event {
            RegEvent::NodeAdded(obs) => self.on_node_added(obs),
            RegEvent::PortAdded(port) => {
                self.push_id(port.id, Slot::Port(port.serial));
                self.ports.insert(port.serial, port);
            }
            RegEvent::ClientAdded(client) => {
                self.push_id(client.id, Slot::Client(client.serial));
                self.clients.insert(client.serial, client);
                // A new client can change the pulse candidate; the adapter
                // learns that via `pulse_pid_candidate`. No readiness effect.
            }
            RegEvent::DeviceAdded { id } => self.on_device_added(id),
            RegEvent::LinkAdded {
                serial,
                id,
                endpoints,
            } => self.on_link_added(serial, id, endpoints),
            RegEvent::LinkEndpointsResolved { serial, endpoints } => {
                self.on_link_resolved(serial, endpoints)
            }
            RegEvent::ProcCommProbed { pid, comm } => {
                self.probed_comm.insert(pid, comm);
            }
            RegEvent::Removed { id } => self.on_removed(id),
            RegEvent::ServerSynced => {
                self.server_synced = true;
                self.maybe_complete();
            }
            RegEvent::Tick { now } => {
                self.last_now = now;
                self.maybe_timeout(now);
            }
        }
    }

    fn on_node_added(&mut self, obs: NodeObservation) {
        self.push_id(obs.id, Slot::Node(obs.serial));
        let resolved = obs
            .device_claim
            .device_id
            .is_some_and(|id| self.device_resolved(id));
        match classify::classify(&obs.device_claim, resolved) {
            Classification::Withhold { .. } => {
                self.withheld.insert(obs.serial, obs);
            }
            Classification::SessionDevice => self.admit_node(obs, true),
            Classification::NotADevice | Classification::NotSessionDevice => {
                self.admit_node(obs, false)
            }
        }
        // Withholding a node adds an obligation; admitting one can never
        // complete readiness on its own, but re-check is cheap and keeps the
        // invariant local.
        self.maybe_complete();
    }

    fn admit_node(&mut self, obs: NodeObservation, session_device: bool) {
        let mut props = obs.props;
        props.session_device = session_device;
        self.nodes.insert(
            obs.serial,
            NodeSnapshot {
                serial: obs.serial,
                id: obs.id,
                name: obs.name,
                role: obs.role,
                props,
            },
        );
    }

    fn on_device_added(&mut self, id: GlobalId) {
        self.push_id(id, Slot::Device);
        *self.resolved_devices.entry(id).or_insert(0) += 1;
        // Admit every node that was withheld waiting on exactly this Device.
        let ready: Vec<Serial> = self
            .withheld
            .iter()
            .filter(|(_, obs)| obs.device_claim.device_id == Some(id))
            .map(|(&serial, _)| serial)
            .collect();
        for serial in ready {
            if let Some(obs) = self.withheld.remove(&serial) {
                // Resolved now, so classify yields a terminal answer, never
                // Withhold again.
                let session_device = matches!(
                    classify::classify(&obs.device_claim, true),
                    Classification::SessionDevice
                );
                self.admit_node(obs, session_device);
            }
        }
        self.maybe_complete();
    }

    fn on_link_added(&mut self, serial: Serial, id: GlobalId, endpoints: Option<LinkEndpoints>) {
        self.push_id(id, Slot::Link(serial));
        match endpoints {
            Some(e) => {
                self.links.insert(serial, link_snapshot(serial, id, e));
            }
            None => {
                // Correctness path: withhold the Link until the bind fallback
                // resolves it. Counts as an outstanding obligation.
                self.pending_links.insert(serial, id);
            }
        }
        self.maybe_complete();
    }

    fn on_link_resolved(&mut self, serial: Serial, endpoints: LinkEndpoints) {
        // `remove` also guards against a stale resolution for a Link already
        // gone: unknown serial ⇒ ignore.
        if let Some(id) = self.pending_links.remove(&serial) {
            self.links
                .insert(serial, link_snapshot(serial, id, endpoints));
            self.maybe_complete();
        }
    }

    fn on_removed(&mut self, id: GlobalId) {
        let Some(queue) = self.live_ids.get_mut(&id) else {
            tracing::warn!(global_id = id.0, "observer: remove for an id we never saw");
            return;
        };
        // Oldest generation first — the id may be shared during a
        // missed-removal window.
        let slot = queue.pop_front();
        if queue.is_empty() {
            self.live_ids.remove(&id);
        }
        match slot {
            Some(Slot::Node(serial)) => {
                if self.nodes.remove(&serial).is_none() {
                    // Was still withheld — drop the obligation.
                    self.withheld.remove(&serial);
                }
            }
            Some(Slot::Port(serial)) => {
                self.ports.remove(&serial);
            }
            Some(Slot::Link(serial)) => {
                self.links.remove(&serial);
                self.pending_links.remove(&serial);
            }
            Some(Slot::Client(serial)) => {
                self.clients.remove(&serial);
            }
            Some(Slot::Device) => {
                if let Some(count) = self.resolved_devices.get_mut(&id) {
                    *count -= 1;
                    if *count == 0 {
                        self.resolved_devices.remove(&id);
                    }
                }
            }
            None => {
                tracing::warn!(global_id = id.0, "observer: empty id slot on remove");
            }
        }
        // A removal can drain the last obligation (a withheld node or pending
        // link vanished before it resolved).
        self.maybe_complete();
    }

    fn push_id(&mut self, id: GlobalId, slot: Slot) {
        self.live_ids.entry(id).or_default().push_back(slot);
    }

    fn device_resolved(&self, id: GlobalId) -> bool {
        self.resolved_devices.get(&id).is_some_and(|&n| n > 0)
    }

    /// Every obligation that must clear before the initial graph is trusted:
    /// no node withheld on an unresolved Device, no Link awaiting its bind.
    fn obligations_outstanding(&self) -> bool {
        !self.withheld.is_empty() || !self.pending_links.is_empty()
    }

    /// Completion needs no clock — only the sync flag and an empty obligation
    /// set — so it may fire on any mutating event. Sticky once reached.
    fn maybe_complete(&mut self) {
        if self.readiness != Readiness::Waiting {
            return;
        }
        if self.server_synced && !self.obligations_outstanding() {
            self.readiness = Readiness::Complete;
            tracing::info!("observer: readiness epoch reached (synced + no obligations)");
        }
    }

    /// Only the timeout consults the clock.
    fn maybe_timeout(&mut self, now: Millis) {
        if self.readiness != Readiness::Waiting {
            return;
        }
        if now >= self.deadline {
            self.readiness = Readiness::TimedOut;
            tracing::warn!(
                withheld = self.withheld.len(),
                pending_links = self.pending_links.len(),
                "observer: readiness epoch timed out with obligations outstanding — fail closed"
            );
        }
    }

    /// pipewire-pulse's PID from the current clients, validated against the
    /// probed `comm`. `None` whenever anything is ambiguous or unconfirmed —
    /// the safe answer (key 4 unusable).
    fn pulse_pid(&self) -> Option<u32> {
        let candidate = self.pulse_pid_candidate()?;
        let comm = self.probed_comm.get(&candidate).and_then(|c| c.as_deref());
        pulse_pid::validate(candidate, comm)
    }

    /// Project the current state into the taint engine's inputs.
    pub fn project(&self) -> Projection {
        let snapshot = GraphSnapshot::new(
            self.nodes.values().cloned().collect(),
            self.ports.values().cloned().collect(),
            self.links.values().cloned().collect(),
            self.clients.values().cloned().collect(),
        );
        Projection {
            snapshot,
            pipewire_pulse_pid: self.pulse_pid(),
            graph_ready: self.graph_ready(),
        }
    }
}

fn link_snapshot(serial: Serial, id: GlobalId, e: LinkEndpoints) -> LinkSnapshot {
    LinkSnapshot {
        serial,
        id,
        output_node: e.output_node,
        input_node: e.input_node,
        output_port: e.output_port,
        input_port: e.input_port,
    }
}