From bbf67444447a231af4494875e738935d4a537459 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sat, 25 Jul 2026 15:43:10 -0400 Subject: [PATCH] =?UTF-8?q?host/audit:=20phase=205=20=E2=80=94=20dry-run?= =?UTF-8?q?=20audit=20mode=20(read-only)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Runs phases 2-4 against the live PipeWire graph on every registry event and reports the complete eligible/excluded candidate partition with stable reason codes. Creates no links, loads no modules, changes no routing. Impl plan §5. Two entry points behind the hidden PIXELPASS_AUDIO_AUDIT=1 trigger: inside a real `pixelpass host` run (the plan-literal reading, proves the path phase 6 will mutate), and a hidden `--audit-audio` standalone mode with no iroh endpoint or capture pipeline, which is what drives the §5.1 matrix. The recompute runs inline on the observer thread via a new ProjectionSink hook, once per applied event. Polling `latest()` was rejected: it coalesces, and phase 4 detects a module unload by observing the empty gap before the next module appears — with indices reused verbatim (v3.4 §5.2 correction 3), a missed gap aliases a fresh module onto a dead identity. Running inline is what makes phase 4's "one observe per graph event" contract true, and it puts the cost where O5 can measure it. Split as usual: the auditor and the metrics are pure and unit-tested; the clock, the writer and the env parsing are the thin edge in `sink`/`run`. - audit/mod.rs Auditor: AEC validator + taint engine + record building. The AEC gate and the engine's own reasons stay distinguishable — a shut gate must not erase the reason codes the §5.1 rows assert. - audit/metrics.rs O5: event rate, bucketed recompute distribution + exact max, busy fraction, and a documented lower-bound queueing proxy (libpipewire exposes no queue depth). - audit/sink.rs JSON Lines to stderr, or PIXELPASS_AUDIO_AUDIT_FILE. Never stdout — peerspeak parses that stream. - audit/run.rs Env parsing; a malformed AEC value is fatal, matching phase 4's rule that it must not silently become "no AEC". Observer gains `EventKind` (derived from RegEvent, so a consumer's view of "was this a real graph change?" cannot disagree with the model's) and `Projection::readiness`, which distinguishes the three ways graph_ready can be false. taint::fixture is now pub(crate) so audit tests share one graph vocabulary with the taint tests. 33 new tests, 178 green, clippy -D warnings and fmt clean. Co-Authored-By: Claude Opus 5 --- src/cli.rs | 12 + src/common/signal.rs | 10 + src/host/audit/metrics.rs | 269 +++++++++++ src/host/audit/mod.rs | 455 ++++++++++++++++++ src/host/audit/run.rs | 170 +++++++ src/host/audit/sink.rs | 184 +++++++ src/host/audit/tests.rs | 896 +++++++++++++++++++++++++++++++++++ src/host/mod.rs | 7 + src/host/observer/adapter.rs | 72 ++- src/host/observer/mod.rs | 45 ++ src/host/taint/mod.rs | 5 +- src/main.rs | 7 + 12 files changed, 2124 insertions(+), 8 deletions(-) create mode 100644 src/host/audit/metrics.rs create mode 100644 src/host/audit/mod.rs create mode 100644 src/host/audit/run.rs create mode 100644 src/host/audit/sink.rs create mode 100644 src/host/audit/tests.rs diff --git a/src/cli.rs b/src/cli.rs index 7409189..aa7e4e8 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -118,6 +118,18 @@ pub struct Cli { /// or if the previously saved test result is stale. #[arg(long)] pub reconfigure: bool, + + /// Run the read-only audio-exclusion dry-run audit against the live + /// PipeWire graph, then exit on ctrl-c. Emits one JSON object per line to + /// stderr (or to `PIXELPASS_AUDIO_AUDIT_FILE`) describing which audio + /// streams would be eligible for a screen share and why the rest would not. + /// Creates no links and changes no routing. + /// + /// Hidden: this is development instrumentation for the screen-share audio + /// exclusion work (impl plan phase 5), not a user-facing feature, and the + /// record schema is free to change until phase 6 fixes it. + #[arg(long, hide = true)] + pub audit_audio: bool, } #[derive(ValueEnum, Clone, Copy, Debug)] diff --git a/src/common/signal.rs b/src/common/signal.rs index 9d8ec2a..be48693 100644 --- a/src/common/signal.rs +++ b/src/common/signal.rs @@ -1,5 +1,15 @@ +use anyhow::{Context, Result}; +use tokio::signal::unix::{Signal, SignalKind}; use tokio_util::sync::CancellationToken; +/// A stream of SIGTERMs, for the callers that need to shut down cleanly when +/// something other than a human at a terminal asks them to (`timeout`, a test +/// harness, a service manager). Ctrl-c alone covers only the interactive case. +pub fn terminate_stream() -> Result { + tokio::signal::unix::signal(SignalKind::terminate()) + .context("could not install a SIGTERM handler") +} + /// Install a ctrl-c handler that triggers the returned token. /// /// The first ctrl-c cancels gracefully; a second ctrl-c terminates the process. diff --git a/src/host/audit/metrics.rs b/src/host/audit/metrics.rs new file mode 100644 index 0000000..a95421c --- /dev/null +++ b/src/host/audit/metrics.rs @@ -0,0 +1,269 @@ +//! O5 measurement, pure (impl plan §5.2). +//! +//! v3.4 §6.4 asserts "a full recompute per graph event is fine for v1". The +//! impl plan closes O5 by refusing to let that rest on a node count: what has +//! to be recorded is the **graph-event rate**, the **recompute duration +//! distribution and maximum**, and **whether events queue behind recompute or +//! logging**. +//! +//! Everything here is arithmetic over samples the caller supplies. The clock +//! reads live at the I/O edge ([`super::sink`]), which is what keeps the +//! statistics unit-testable: a test feeds a hand-written sample sequence and +//! asserts the summary exactly, with no timing flake. +//! +//! **The queueing measure is a proxy, and a one-directional one.** libpipewire +//! dispatches registry callbacks serially on its own loop thread and exposes no +//! queue depth, so nothing here can read a backlog directly. What it can see is +//! that the observer thread was *continuously busy*: if an event begins being +//! handled within [`QUEUE_THRESHOLD_US`] of the previous sample's completion, +//! it was almost certainly already waiting while that recompute ran. That makes +//! [`Summary::queued_events`] a **lower bound** — a genuine backlog always shows +//! up in it, but a burst that happens to arrive exactly as the loop goes idle is +//! counted as un-queued. Combined with [`Summary::busy_fraction`] (which needs +//! no inference at all) it is enough to answer O5 in the direction that matters: +//! a low busy fraction with zero queued events is headroom, and anything else is +//! a number to argue about rather than an assumption to inherit. + +use serde::Serialize; + +use crate::host::observer::EventKind; + +/// An event beginning this close behind the previous sample's completion is +/// counted as having queued. Deliberately tight: the cost of being wrong in the +/// generous direction is a metric that overstates backlog and sends a later +/// round chasing a non-problem. +pub const QUEUE_THRESHOLD_US: u64 = 100; + +/// Upper bounds of the duration histogram, microseconds. A twelfth (overflow) +/// bucket catches everything at or above the last bound. Log-ish spacing: the +/// interesting question is which order of magnitude a recompute lands in, not +/// its exact microsecond. +pub const BUCKET_BOUNDS_US: [u64; 11] = [ + 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000, 25_000, 50_000, 100_000, +]; + +/// Human labels for the histogram buckets, parallel to [`BUCKET_BOUNDS_US`] +/// plus the overflow bucket. +pub const BUCKET_LABELS: [&str; 12] = [ + "<50us", "<100us", "<250us", "<500us", "<1ms", "<2.5ms", "<5ms", "<10ms", "<25ms", "<50ms", + "<100ms", ">=100ms", +]; + +/// A bucketed duration distribution with exact count, sum and maximum. +/// +/// Bounded memory by construction — the audit runs for as long as a share does, +/// and keeping every sample to compute an exact percentile would grow without +/// limit. The maximum, which is the number O5 actually cares about, is kept +/// exactly; percentiles are reported as the bucket they fall in. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct Histogram { + buckets: [u64; 12], + count: u64, + sum_us: u64, + max_us: u64, +} + +impl Histogram { + pub fn record(&mut self, us: u64) { + let index = BUCKET_BOUNDS_US + .iter() + .position(|&bound| us < bound) + .unwrap_or(BUCKET_BOUNDS_US.len()); + self.buckets[index] += 1; + self.count += 1; + self.sum_us = self.sum_us.saturating_add(us); + self.max_us = self.max_us.max(us); + } + + pub fn count(&self) -> u64 { + self.count + } + + pub fn max_us(&self) -> u64 { + self.max_us + } + + pub fn sum_us(&self) -> u64 { + self.sum_us + } + + pub fn mean_us(&self) -> Option { + (self.count > 0).then(|| self.sum_us / self.count) + } + + /// The label of the bucket the `q`-quantile falls in (`q` in `0.0..=1.0`), + /// or `None` when nothing has been recorded. + /// + /// Uses the *nearest-rank* definition: the bucket containing the + /// `ceil(q · count)`-th sample in ascending order. Reported as a bucket + /// rather than a number because interpolating inside a bucket would invent + /// precision the histogram does not have. + pub fn quantile_bucket(&self, q: f64) -> Option<&'static str> { + if self.count == 0 { + return None; + } + let q = q.clamp(0.0, 1.0); + // Rank is 1-based; q = 0 still names the bucket holding the smallest + // sample rather than degenerating to "no samples". + let rank = ((q * self.count as f64).ceil() as u64).max(1); + let mut cumulative = 0u64; + for (index, &n) in self.buckets.iter().enumerate() { + cumulative += n; + if cumulative >= rank { + return Some(BUCKET_LABELS[index]); + } + } + // Unreachable while `count` is the sum of the buckets, but returning the + // top bucket is the fail-loud answer rather than a panic in a metric. + Some(BUCKET_LABELS[BUCKET_LABELS.len() - 1]) + } + + /// Non-empty buckets as `(label, count)`, ascending. Empty buckets are + /// dropped so a summary line stays readable. + pub fn distribution(&self) -> Vec<(&'static str, u64)> { + self.buckets + .iter() + .enumerate() + .filter(|&(_, &n)| n > 0) + .map(|(index, &n)| (BUCKET_LABELS[index], n)) + .collect() + } +} + +/// One handled event, as timed by the I/O edge. +/// +/// Ticks are the AEC validator's clock, not graph changes, so [`Metrics`] counts +/// them separately — folding them into the event rate would inflate it by a +/// constant 4 Hz and hide the real graph churn. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Sample { + /// Monotonic microseconds (since observer start) at which handling began. + pub at_us: u64, + /// Microseconds between the previous sample's completion and `at_us`. Zero + /// for the first sample. + pub gap_us: u64, + /// Time spent in the AEC observe + taint recompute. + pub recompute_us: u64, + /// Time spent serialising and writing the record, zero when nothing was + /// emitted. Separate from `recompute_us` because O5 asks about queueing + /// behind recompute **or logging** — and if logging turns out to dominate, + /// that is a fixable problem of a different kind. + pub emit_us: u64, + pub kind: EventKind, +} + +/// Rolling O5 state. Fold samples in with [`Metrics::record`]; read with +/// [`Metrics::summary`]. +#[derive(Clone, Debug, Default)] +pub struct Metrics { + graph_events: u64, + tick_events: u64, + emitted_records: u64, + recompute: Histogram, + emit: Histogram, + busy_us: u64, + queued_events: u64, + first_event_us: Option, + last_completion_us: u64, +} + +impl Metrics { + pub fn record(&mut self, sample: Sample) { + match sample.kind { + EventKind::Graph => self.graph_events += 1, + EventKind::Tick => self.tick_events += 1, + } + self.recompute.record(sample.recompute_us); + if sample.emit_us > 0 { + self.emitted_records += 1; + self.emit.record(sample.emit_us); + } + self.busy_us = self + .busy_us + .saturating_add(sample.recompute_us) + .saturating_add(sample.emit_us); + + // The first sample has no predecessor to have queued behind. + if self.first_event_us.is_some() && sample.gap_us <= QUEUE_THRESHOLD_US { + self.queued_events += 1; + } + self.first_event_us.get_or_insert(sample.at_us); + self.last_completion_us = sample + .at_us + .saturating_add(sample.recompute_us) + .saturating_add(sample.emit_us); + } + + pub fn summary(&self) -> Summary { + let span_us = self + .first_event_us + .map(|first| self.last_completion_us.saturating_sub(first)) + .unwrap_or(0); + // A rate needs a span to divide by; one event in zero elapsed time has + // no rate, and reporting a made-up one is worse than reporting none. + let graph_events_per_sec = (span_us > 0) + .then(|| self.graph_events as f64 * 1_000_000.0 / span_us as f64) + .map(round_2); + let busy_fraction = (span_us > 0).then(|| round_4(self.busy_us as f64 / span_us as f64)); + + Summary { + graph_events: self.graph_events, + tick_events: self.tick_events, + emitted_records: self.emitted_records, + span_us, + graph_events_per_sec, + recompute_max_us: self.recompute.max_us(), + recompute_mean_us: self.recompute.mean_us(), + recompute_p50: self.recompute.quantile_bucket(0.50), + recompute_p90: self.recompute.quantile_bucket(0.90), + recompute_p99: self.recompute.quantile_bucket(0.99), + recompute_distribution: self.recompute.distribution(), + emit_max_us: self.emit.max_us(), + emit_mean_us: self.emit.mean_us(), + emit_distribution: self.emit.distribution(), + busy_us: self.busy_us, + busy_fraction, + queued_events: self.queued_events, + queue_threshold_us: QUEUE_THRESHOLD_US, + } + } +} + +/// The O5 answer, as emitted. +#[derive(Clone, Debug, PartialEq, Serialize)] +pub struct Summary { + pub graph_events: u64, + pub tick_events: u64, + pub emitted_records: u64, + /// First event to last completion, microseconds. + pub span_us: u64, + pub graph_events_per_sec: Option, + pub recompute_max_us: u64, + pub recompute_mean_us: Option, + pub recompute_p50: Option<&'static str>, + pub recompute_p90: Option<&'static str>, + pub recompute_p99: Option<&'static str>, + pub recompute_distribution: Vec<(&'static str, u64)>, + pub emit_max_us: u64, + pub emit_mean_us: Option, + pub emit_distribution: Vec<(&'static str, u64)>, + /// Total observer-thread time spent recomputing and logging. + pub busy_us: u64, + /// `busy_us / span_us` — the share of wall time the observer thread could + /// not be servicing PipeWire. Needs no inference, unlike `queued_events`. + pub busy_fraction: Option, + /// Events that began within `queue_threshold_us` of the previous sample's + /// completion — a **lower bound** on backlog, see the module header. + pub queued_events: u64, + pub queue_threshold_us: u64, +} + +/// Keep the JSON readable: a rate to two decimals and a fraction to four are +/// well past the precision any of this is good to. +fn round_2(value: f64) -> f64 { + (value * 100.0).round() / 100.0 +} + +fn round_4(value: f64) -> f64 { + (value * 10_000.0).round() / 10_000.0 +} diff --git a/src/host/audit/mod.rs b/src/host/audit/mod.rs new file mode 100644 index 0000000..996805a --- /dev/null +++ b/src/host/audit/mod.rs @@ -0,0 +1,455 @@ +//! Phase 5 — dry-run audit mode 🚦 (impl plan §5). +//! +//! **This phase adds no capability. Its entire purpose is to be wrong loudly +//! and safely.** It runs phases 2–4 against the *live* graph on every graph +//! event and reports what they conclude. It creates no links, loads no modules, +//! and changes no routing — the only thing it produces is a line of JSON. +//! +//! Why this is the gate the plan marks 🚦: the defects that matter here are +//! graph-*reasoning* defects. The 57 phase-2 fixture tests prove the engine +//! matches my model of PipeWire; only a live run proves my model matches +//! PipeWire. A wrong answer at this phase costs a log line. The same wrong +//! answer in phase 6 costs an echo — the sharer's own voice, copied back into +//! the share, which is the failure this whole design exists to prevent. +//! +//! ## The one structural requirement (§5.1) +//! +//! Every emitted record carries the **complete candidate universe partitioned +//! into exact eligible and excluded sets**, with a stable reason code on each +//! excluded row — never a spot check on named nodes. Checking only the nodes a +//! row names constrains nothing about the rest, and it lets the degenerate +//! "exclude everything" implementation pass: that build is silent, produces no +//! echo, and satisfies any assertion phrased purely as *this must be excluded*. +//! Asserting the eligible half of each row is what fails it. That requirement is +//! also the plan's answer to open question O7 (over-exclusion needs no separate +//! gate — it is subsumed by this one). +//! +//! ## What is deliberately *not* here +//! +//! - **No link creation, and no code path that could reach one.** The auditor +//! consumes a [`Projection`] and returns a record. It has no handle to +//! anything mutable. +//! - **No stdout.** Records go to stderr as JSON Lines +//! ([`sink`]) because peerspeak parses pixelpass's stdout event stream +//! (`screenshare/mod.rs:92`); a stray line there corrupts it. +//! - **No `--aec` CLI flag.** That surface is phase 7's mode selector. The audit +//! takes its AEC identity from `PIXELPASS_AUDIO_AUDIT_AEC` through the +//! *same* [`parse_aec_arg`] the real flag will use, so the parser and the +//! validator are both exercised without committing to a public interface +//! before it is designed. +//! +//! ## Fan-out gating vs. taint (read before interpreting a record) +//! +//! Two independent things can exclude a candidate and the record keeps them +//! distinguishable: +//! +//! - The **taint engine** (phase 2) excludes individual nodes with its own +//! reason codes — `peerspeak-owned`, `aec-identity`, `tainted-upstream`, … +//! - The **AEC validator** (phase 4) can forbid fan-out *entirely*, regardless +//! of taint, whenever the configured identity is unvalidated, failed or +//! revoked. Silence over echo. +//! +//! When the gate is shut, a candidate the engine would have called eligible is +//! reported excluded with an audit-level reason ([`GateReason`]); a candidate +//! the engine excluded on its own keeps *its* reason, because that names the +//! mechanism that actually applies to it. `fan_out_permitted` on the record +//! carries the gate state, so the two cases are always tellable apart. +//! +//! **Consequence for the §5.1 matrix:** every row whose point is the +//! eligible/excluded partition must run with `PIXELPASS_AUDIO_AUDIT_AEC=off` +//! (state `NotConfigured`, gate open). Row 12 — the AEC lifecycle row — is the +//! one that runs with a real `pulse-module:`, and the gate slamming shut is +//! precisely what it asserts. + +#![allow(dead_code)] // Trigger paths are wired by `sink` + `run`; rows are read by tests. + +pub mod metrics; +pub mod run; +pub mod sink; + +#[cfg(test)] +mod tests; + +use serde::Serialize; + +use crate::host::aec::{AecConfig, AecState, AecValidator}; +use crate::host::observer::{EventKind, Millis, Projection, Readiness}; +use crate::host::taint::snapshot::Serial; +use crate::host::taint::{Decisions, Eligibility, ExclusionCtx, StickyState, evaluate}; + +/// How long the AEC validator may sit in `Validating` after the graph first +/// reports ready before failing closed. Generous relative to the observer's own +/// 2 s readiness budget: in the audit a `Failed` is a diagnostic, and timing out +/// early would report an absent module that was merely slow to appear. +pub const AEC_VALIDATION_TIMEOUT_MILLIS: Millis = 5_000; + +/// Everything the auditor needs beyond the live graph. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct AuditConfig { + /// The AEC identity to validate, as parsed from + /// `PIXELPASS_AUDIO_AUDIT_AEC`. Defaults to [`AecConfig::Off`] — an audit + /// run is not a share, so "there is no echo canceller in play" is the + /// honest default, and it is what leaves the fan-out gate open for the + /// partition rows. + pub aec: AecConfig, + pub aec_timeout: Millis, +} + +impl Default for AuditConfig { + fn default() -> Self { + Self { + aec: AecConfig::Off, + aec_timeout: AEC_VALIDATION_TIMEOUT_MILLIS, + } + } +} + +/// An audit-level exclusion: the AEC validator has shut the fan-out gate. These +/// codes are disjoint from the taint engine's +/// [`Reason::code`](crate::host::taint::Reason::code) values, so a reader never +/// has to know which layer produced a code to interpret it. +// The shared `Aec` prefix is the point: `GateReason::Validating` and +// `AecState::Validating` would be one careless glob import away from being +// confused, and these three are the *audit's* view of that machine, not the +// machine itself. +#[allow(clippy::enum_variant_names)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum GateReason { + /// The configured AEC identity has not been seen yet. Not an error — the + /// module may still be loading — but no fan-out happens meanwhile. + AecValidating, + /// The deadline passed with the identity never observed. + AecFailed, + /// The whole identity disappeared mid-run: every node bearing the index is + /// gone (v3.4 §5.3). + AecRevoked, +} + +impl GateReason { + pub fn code(self) -> &'static str { + match self { + Self::AecValidating => "aec-validating", + Self::AecFailed => "aec-failed", + Self::AecRevoked => "aec-revoked", + } + } + + /// The gate reason implied by a validator state, or `None` when fan-out is + /// permitted. Mirrors [`AecValidator::fan_out_permitted`] — kept as one + /// `match` over the same enum so the two cannot drift: every state that + /// permits fan-out maps to `None` and every state that forbids it maps to a + /// code. + pub fn from_state(state: AecState) -> Option { + match state { + AecState::NotConfigured | AecState::Validated => None, + AecState::Validating => Some(Self::AecValidating), + AecState::Failed => Some(Self::AecFailed), + AecState::Revoked => Some(Self::AecRevoked), + } + } +} + +/// Stable string for an [`AecState`], for the record's `aec_state` field. +/// +/// Defined here rather than on [`AecState`] to keep the merged phase-4 module +/// untouched by a reporting concern. +fn aec_state_code(state: AecState) -> &'static str { + match state { + AecState::NotConfigured => "not-configured", + AecState::Validating => "validating", + AecState::Validated => "validated", + AecState::Failed => "failed", + AecState::Revoked => "revoked", + } +} + +/// Stable string for the observer's readiness epoch. +fn readiness_code(readiness: Readiness) -> &'static str { + match readiness { + Readiness::Waiting => "waiting", + Readiness::Complete => "complete", + Readiness::TimedOut => "timed-out", + } +} + +/// One candidate node's effective answer. `reason` is `None` exactly when +/// `eligible` is true. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct AuditRow { + pub serial: u64, + pub name: Option, + pub eligible: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub reason: Option<&'static str>, + /// The exclusion was carried over from a previous snapshot rather than + /// derived from the current topology (phase-2 stickiness). + pub sticky: bool, +} + +/// A tainted node of *any* media role, not just fan-out candidates. Candidates +/// already appear in [`AuditBody::candidates`]; this is the diagnostic view — +/// when a candidate's exclusion is a surprise, the taint that reached it is the +/// next question, and it usually sits on a node that is not itself a candidate. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct TaintRow { + pub serial: u64, + pub name: Option, + pub reason: &'static str, + pub sticky: bool, +} + +/// The decision content of one recompute — everything except which recompute it +/// was. Split out from [`AuditRecord`] so "did anything actually change?" is a +/// derived `==` rather than a hand-maintained field comparison that a later +/// field addition could silently fall out of. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct AuditBody { + /// The observer's dynamic readiness. False ⇒ every candidate is excluded + /// `graph-not-ready`; no decision from a partial graph is a decision. + pub graph_ready: bool, + /// The sticky readiness epoch, which distinguishes the three ways + /// `graph_ready` can be false (see [`Projection::readiness`]). + pub epoch: &'static str, + pub aec_state: &'static str, + /// The index handed to the taint engine — `Some` only while `Validated`. + #[serde(skip_serializing_if = "Option::is_none")] + pub aec_module_id: Option, + /// Whether the AEC validator permits fan-out at all right now. + pub fan_out_permitted: bool, + /// The audit-level reason fan-out is forbidden, when it is. + #[serde(skip_serializing_if = "Option::is_none")] + pub gate_reason: Option<&'static str>, + /// **The complete candidate universe**, ascending by serial — every + /// `Stream/Output/Audio` node in the snapshot, partitioned. §5.1's exact + /// partition is `candidates`, not a subset of it. + pub candidates: Vec, + pub eligible_count: usize, + pub excluded_count: usize, + /// Taint across all node roles, ascending by serial. + pub taint: Vec, +} + +impl AuditBody { + /// Serials of eligible candidates, ascending — the half of the partition an + /// exclude-everything build fails. + pub fn eligible(&self) -> Vec { + self.candidates + .iter() + .filter(|row| row.eligible) + .map(|row| row.serial) + .collect() + } + + /// `(serial, reason code)` for excluded candidates, ascending. + pub fn excluded(&self) -> Vec<(u64, &'static str)> { + self.candidates + .iter() + .filter(|row| !row.eligible) + .map(|row| (row.serial, row.reason.unwrap_or("?"))) + .collect() + } + + /// The eligible candidate with this name, if any. Convenience for the + /// matrix rows, which name nodes rather than serials. + pub fn row_named(&self, name: &str) -> Option<&AuditRow> { + self.candidates + .iter() + .find(|row| row.name.as_deref() == Some(name)) + } +} + +/// One recompute, as emitted. +#[derive(Clone, Debug, PartialEq, Eq, Serialize)] +pub struct AuditRecord { + /// Monotonic per-run counter over *every* recompute, emitted or suppressed, + /// so a gap in the emitted sequence is visibly a suppression rather than a + /// lost line. + pub seq: u64, + pub trigger: &'static str, + /// Observer-clock milliseconds at which this recompute ran. + pub at_ms: Millis, + #[serde(flatten)] + pub body: AuditBody, +} + +/// What one [`Auditor::observe`] produced. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AuditOutcome { + pub record: AuditRecord, + /// Whether the record should be written. See [`Auditor::observe`]. + pub emit: bool, +} + +/// The dry-run auditor: phases 2–4 folded together over a live projection. +/// +/// Read-only by construction — it borrows a [`Projection`] and owns only the +/// state phases 2 and 4 thread explicitly ([`StickyState`], [`AecValidator`]). +/// There is no field here through which a link could be created. +#[derive(Clone, Debug)] +pub struct Auditor { + validator: AecValidator, + sticky: StickyState, + seq: u64, + /// The body of the last record actually written, for change suppression. + last_emitted: Option, +} + +impl Auditor { + pub fn new(config: AuditConfig) -> Self { + Self { + validator: AecValidator::new(config.aec, config.aec_timeout), + sticky: StickyState::default(), + seq: 0, + last_emitted: None, + } + } + + pub fn aec_state(&self) -> AecState { + self.validator.state() + } + + pub fn sticky(&self) -> &StickyState { + &self.sticky + } + + /// Fold one projection into the audit. + /// + /// **Called once per applied registry event — never on a coalesced batch.** + /// That is not a performance preference, it is the phase-4 integration + /// contract (`aec/mod.rs`, the `Validated` arm): revocation is detected by + /// observing the *empty gap* between a module unload and the next reload, + /// and module indices are reused verbatim (v3.4 §5.2 correction 3). Coalesce + /// across that gap and a fresh module silently inherits a dead module's + /// validated identity. [`sink`] is what upholds this, by running the + /// recompute inline on the observer thread rather than polling + /// [`RegistryObserverHandle::latest`](crate::host::observer::adapter::RegistryObserverHandle::latest), + /// which coalesces by nature. + /// + /// `emit` is true for every graph-triggered recompute, and for a + /// tick-triggered one only when the decision content changed. Ticks arrive + /// at a constant 4 Hz purely to drive the AEC deadline; emitting an + /// identical record four times a second would bury the graph events the + /// audit exists to show. `seq` still advances on suppressed records, so + /// nothing about the run is silently unaccounted for. + pub fn observe( + &mut self, + projection: &Projection, + kind: EventKind, + now: Millis, + ) -> AuditOutcome { + self.seq += 1; + + // Phase 4 first: its verdict is an *input* to phase 2 via + // `ExclusionCtx::aec_module_id`, so observing the graph in the other + // order would evaluate taint against the previous recompute's identity. + self.validator + .observe(&projection.snapshot, projection.graph_ready, now); + let aec_state = self.validator.state(); + let gate_reason = GateReason::from_state(aec_state); + + let ctx = ExclusionCtx { + aec_module_id: self.validator.validated_module_id(), + pipewire_pulse_pid: projection.pipewire_pulse_pid, + // The audit creates nothing, so it owns nothing. Another host's + // capture sink is still caught — by the `pixelpass_capture_*` name + // prefix (v3.4 §6.2), which is what §5.1 row 7 exercises — so an + // empty set costs the matrix nothing. + pixelpass_owned: Default::default(), + graph_ready: projection.graph_ready, + }; + + let (decisions, sticky) = evaluate(&projection.snapshot, &ctx, &self.sticky); + self.sticky = sticky; + + let body = build_body( + projection, + &decisions, + aec_state, + self.validator.validated_module_id(), + gate_reason, + ); + + let emit = kind == EventKind::Graph || self.last_emitted.as_ref() != Some(&body); + if emit { + self.last_emitted = Some(body.clone()); + } + + AuditOutcome { + record: AuditRecord { + seq: self.seq, + trigger: kind.code(), + at_ms: now, + body, + }, + emit, + } + } +} + +fn build_body( + projection: &Projection, + decisions: &Decisions, + aec_state: AecState, + aec_module_id: Option, + gate_reason: Option, +) -> AuditBody { + let candidates: Vec = decisions + .candidates + .values() + .map(|decision| { + // The engine's own reason wins when it has one: it names the + // mechanism that actually excluded *this* node, which is what the + // §5.1 rows assert. The gate reason applies only to candidates the + // engine would have passed — otherwise a shut gate would erase every + // reason code in the record and the matrix would stop constraining + // the engine at all. + let (eligible, reason, sticky) = match decision.eligibility { + Eligibility::NotEligible { reason, sticky } => (false, Some(reason.code()), sticky), + Eligibility::Eligible => match gate_reason { + Some(gate) => (false, Some(gate.code()), false), + None => (true, None, false), + }, + }; + AuditRow { + serial: decision.serial.0, + name: decision.name.clone(), + eligible, + reason, + sticky, + } + }) + .collect(); + + let eligible_count = candidates.iter().filter(|row| row.eligible).count(); + + let taint: Vec = decisions + .taint + .iter() + .map(|(&serial, entry)| TaintRow { + serial: serial.0, + name: node_name(projection, serial), + reason: entry.reason.code(), + sticky: entry.sticky, + }) + .collect(); + + AuditBody { + graph_ready: projection.graph_ready, + epoch: readiness_code(projection.readiness), + aec_state: aec_state_code(aec_state), + aec_module_id, + fan_out_permitted: gate_reason.is_none(), + gate_reason: gate_reason.map(GateReason::code), + excluded_count: candidates.len() - eligible_count, + eligible_count, + candidates, + taint, + } +} + +fn node_name(projection: &Projection, serial: Serial) -> Option { + projection + .snapshot + .node(serial) + .and_then(|node| node.name.clone()) +} diff --git a/src/host/audit/run.rs b/src/host/audit/run.rs new file mode 100644 index 0000000..c7d8067 --- /dev/null +++ b/src/host/audit/run.rs @@ -0,0 +1,170 @@ +//! Triggering the dry-run audit: environment parsing and the two entry points. +//! +//! The impl plan §5 specifies a **hidden trigger**, `PIXELPASS_AUDIO_AUDIT=1`. +//! It is honoured in two places, which answer two different questions: +//! +//! - **Inside a real `pixelpass host` run** ([`spawn_if_enabled`]) — proves the +//! audit works in the code path phase 6 will actually mutate. This is the +//! plan-literal reading of the trigger. +//! - **Standalone** ([`run_standalone`], behind the hidden `--audit-audio` +//! flag) — observer plus auditor and nothing else: no iroh endpoint, no +//! display-server detection, no capture pipeline, no ticket. This is what +//! drives the §5.1 matrix, because a row that fails should fail for a reason +//! about *audio*, not because a relay was unreachable. +//! +//! Both paths run the same [`AuditSink`] over the same observer, so neither is a +//! simulation of the other. + +use std::fs::OpenOptions; +use std::io::Write; + +use anyhow::{Context, Result, bail}; + +use super::sink::AuditSink; +use super::{AEC_VALIDATION_TIMEOUT_MILLIS, AuditConfig}; +use crate::common::signal; +use crate::host::aec::{AecConfig, AecParseError, parse_aec_arg}; +use crate::host::observer::adapter::RegistryObserverHandle; + +/// The hidden trigger (impl plan §5). Exactly `1` enables the audit; anything +/// else, including `true` or `yes`, does not. +/// +/// Deliberately strict. This variable can only arrive by someone typing it, and +/// a value that *looks* enabling but is not would produce a silent no-op — the +/// single most annoying failure mode for a diagnostic tool. A mistyped value +/// gets a warning (see [`enabled`]) rather than silence. +pub const AUDIT_ENV: &str = "PIXELPASS_AUDIO_AUDIT"; + +/// The AEC identity for the audit, in the `--aec` grammar (`off` or +/// `pulse-module:`). Absent ⇒ `off`. +pub const AUDIT_AEC_ENV: &str = "PIXELPASS_AUDIO_AUDIT_AEC"; + +/// Redirect the JSON Lines stream to this file instead of stderr. +pub const AUDIT_FILE_ENV: &str = "PIXELPASS_AUDIO_AUDIT_FILE"; + +/// Whether the hidden trigger is set. +pub fn enabled() -> bool { + match std::env::var(AUDIT_ENV) { + Ok(value) if value == "1" => true, + Ok(value) => { + tracing::warn!( + "{AUDIT_ENV}={value:?} is not `1`; the audio audit stays off. \ + Set {AUDIT_ENV}=1 to enable it." + ); + false + } + Err(_) => false, + } +} + +/// Build the audit configuration from the environment. +/// +/// A malformed `PIXELPASS_AUDIO_AUDIT_AEC` is **fatal**, matching the phase-4 +/// rule that a bad `--aec` value must not silently become "no AEC": there is no +/// fail-closed default index, so a wrong or dropped one would exclude the wrong +/// node (or nothing at all) and the audit would confidently report a partition +/// computed against an identity nobody asked for. +pub fn config_from_env() -> Result { + let aec = match std::env::var(AUDIT_AEC_ENV) { + Ok(raw) => parse_aec_arg(&raw).map_err(|e| { + anyhow::anyhow!( + "{AUDIT_AEC_ENV}={raw:?} is not a valid AEC argument ({}). \ + Expected `off` or `pulse-module:`, where the index is a bare decimal.", + describe(e) + ) + })?, + Err(std::env::VarError::NotPresent) => AecConfig::Off, + Err(e) => bail!("{AUDIT_AEC_ENV} is not readable: {e}"), + }; + Ok(AuditConfig { + aec, + aec_timeout: AEC_VALIDATION_TIMEOUT_MILLIS, + }) +} + +fn describe(error: AecParseError) -> &'static str { + match error { + AecParseError::Empty => "the value was empty", + AecParseError::UnknownForm => "not `off` and not `pulse-module:...`", + AecParseError::MissingIndex => "`pulse-module:` with no index after the colon", + AecParseError::InvalidIndex => { + "the index was not a bare decimal (no sign, whitespace, or non-digits) that fits in u64" + } + } +} + +/// Where the JSON Lines go. Stderr unless `PIXELPASS_AUDIO_AUDIT_FILE` names a +/// file, which is appended to rather than truncated — a matrix run that restarts +/// the process mid-scenario should not lose the rows it already recorded. +fn writer_from_env() -> Result> { + match std::env::var(AUDIT_FILE_ENV) { + Ok(path) if !path.is_empty() => { + let file = OpenOptions::new() + .create(true) + .append(true) + .open(&path) + .with_context(|| format!("{AUDIT_FILE_ENV}={path:?} could not be opened"))?; + tracing::info!("audio audit: writing records to {path}"); + Ok(Box::new(file)) + } + _ => Ok(Box::new(std::io::stderr())), + } +} + +/// Construct the sink and spawn the observer behind it. +fn spawn_audit() -> Result { + let config = config_from_env()?; + let sink = AuditSink::new(config, writer_from_env()?); + tracing::info!( + aec = ?config.aec, + "audio audit: dry run active — decisions are logged, no links are created" + ); + RegistryObserverHandle::spawn_with_sink(Some(Box::new(sink))) +} + +/// Start the audit if the hidden trigger is set, for a `pixelpass host` run. +/// +/// The returned handle must be held for the lifetime of the run: dropping it +/// stops the observer thread and flushes the final O5 summary. +/// +/// Returns `Err` only when the trigger *was* set and starting failed — a +/// misconfigured audit is worth failing the run over, because the alternative is +/// a host that silently is not being audited while its operator believes it is. +pub fn spawn_if_enabled() -> Result> { + if !enabled() { + return Ok(None); + } + spawn_audit().map(Some) +} + +/// The standalone audit: run the observer and the auditor, and nothing else, +/// until ctrl-c. +/// +/// Does not consult [`AUDIT_ENV`] — reaching this function required passing the +/// hidden `--audit-audio` flag, which is already an explicit request. The +/// environment still supplies the AEC identity and the output file. +pub async fn run_standalone() -> Result<()> { + let cancel = signal::install_ctrl_c(); + let handle = spawn_audit()?; + + eprintln!( + "pixelpass audio audit (dry run): observing the live PipeWire graph.\n\ + No links are created and no routing changes. Ctrl-C to stop." + ); + + // SIGTERM as well as ctrl-c, because this mode is driven by scripts as much + // as by hand — `timeout`, a matrix harness, and systemd all send SIGTERM, + // and the default disposition would kill the process before the sink's + // `Drop` writes the final O5 summary. Losing that summary is losing the + // whole §5.2 measurement for that run. + let mut sigterm = signal::terminate_stream()?; + tokio::select! { + _ = cancel.cancelled() => {} + _ = sigterm.recv() => tracing::info!("SIGTERM received, shutting down"), + } + // Explicit rather than incidental: this drop stops the PipeWire thread, + // which drops the sink, which writes the final metrics line. Letting it fall + // out of scope would do the same thing, but the ordering is the point. + drop(handle); + Ok(()) +} diff --git a/src/host/audit/sink.rs b/src/host/audit/sink.rs new file mode 100644 index 0000000..dae2d35 --- /dev/null +++ b/src/host/audit/sink.rs @@ -0,0 +1,184 @@ +//! The audit's I/O edge: timing, JSON Lines emission, O5 accounting. +//! +//! Everything impure about phase 5 lives here, and it is deliberately thin — +//! read the clock, call [`Auditor::observe`], write a line, fold a +//! [`metrics::Sample`]. The decisions are all upstream in the pure core, which +//! is why the matrix can be argued about in unit tests rather than only in front +//! of a live daemon. +//! +//! ## Why this runs on the observer thread +//! +//! [`AuditSink`] is a [`ProjectionSink`], invoked inline from the PipeWire +//! observer thread once per applied registry event. The obvious alternative — +//! a consumer task polling +//! [`RegistryObserverHandle::latest`](super::super::observer::adapter::RegistryObserverHandle::latest) +//! — was rejected: polling **coalesces**, and phase 4's revocation logic +//! detects a module unload by observing the *empty gap* before the next module +//! appears. Module indices are reused verbatim across an unload/reload (v3.4 +//! §5.2 correction 3), so a poller that misses the gap silently aliases a fresh +//! module onto a dead module's validated identity. Running inline is what makes +//! "one `observe` per graph event, no coalescing" — the contract phase 4 +//! documents as owed — actually true. +//! +//! The cost of that choice is that recompute and logging happen on the thread +//! servicing PipeWire, which is precisely the risk O5 asks about. That is not an +//! accident: this arrangement puts the cost exactly where the measurement can +//! see it. See [`metrics`]. +//! +//! ## Output contract +//! +//! One JSON object per line, to **stderr** by default, each tagged with a `kind` +//! discriminator (`"audit"` or `"metrics"`). Never stdout: peerspeak parses +//! pixelpass's stdout event stream, and the impl plan §5 is explicit that +//! unstructured output must not go there. `PIXELPASS_AUDIO_AUDIT_FILE` +//! redirects the records to a file instead, which is how the §5.1 matrix is +//! driven — it separates the audit stream from interleaved `tracing` output +//! without needing either side to change format. + +use std::io::Write; +use std::time::Instant; + +use serde::Serialize; + +use super::metrics::{self, Metrics, Summary}; +use super::{AuditConfig, AuditRecord, Auditor}; +use crate::host::observer::adapter::ProjectionSink; +use crate::host::observer::{EventKind, Millis, Projection}; + +/// Emit a rolling metrics line every this many ticks. Ticks are 250 ms, so this +/// is every 10 s — often enough that a run killed abruptly still leaves a +/// usable O5 record, rare enough that it does not crowd out the audit records. +const SUMMARY_INTERVAL_TICKS: u64 = 40; + +/// The live audit: pure auditor + clock + writer. +pub struct AuditSink { + auditor: Auditor, + metrics: Metrics, + writer: Box, + /// Set once the first sample has completed, so the first event is not + /// counted as having queued behind a predecessor that does not exist. + last_completion_us: Option, + ticks_since_summary: u64, + /// Wall-clock origin for the microsecond timings. Only used for durations, + /// never for the AEC deadline — that runs on the observer's own clock, + /// handed in as `now_us`, so the validator and the readiness epoch cannot + /// disagree about what time it is. + epoch: Instant, +} + +impl AuditSink { + pub fn new(config: AuditConfig, writer: Box) -> Self { + Self { + auditor: Auditor::new(config), + metrics: Metrics::default(), + writer, + last_completion_us: None, + ticks_since_summary: 0, + epoch: Instant::now(), + } + } + + fn elapsed_us(&self) -> u64 { + u64::try_from(self.epoch.elapsed().as_micros()).unwrap_or(u64::MAX) + } + + /// Write one line. Failures are logged once per occurrence and otherwise + /// ignored: a broken stderr must not take down the observer thread, and the + /// audit is diagnostic — losing a line is a worse audit, not a worse share. + fn write_line(&mut self, line: &T) { + match serde_json::to_string(line) { + Ok(json) => { + if let Err(e) = writeln!(self.writer, "{json}") { + tracing::warn!("audit: failed to write record: {e}"); + } + } + Err(e) => tracing::warn!("audit: failed to serialise record: {e}"), + } + } + + fn write_summary(&mut self, at_ms: Millis) { + let summary = self.metrics.summary(); + self.write_line(&MetricsLine { + kind: "metrics", + at_ms, + summary: &summary, + }); + let _ = self.writer.flush(); + } +} + +impl ProjectionSink for AuditSink { + fn on_projection(&mut self, projection: &Projection, kind: EventKind, now_us: u64) { + let at_us = self.elapsed_us(); + let gap_us = self + .last_completion_us + .map(|previous| at_us.saturating_sub(previous)) + .unwrap_or(0); + + let recompute_start = self.elapsed_us(); + let outcome = self.auditor.observe(projection, kind, now_us / 1_000); + let recompute_us = self.elapsed_us().saturating_sub(recompute_start); + + let emit_us = if outcome.emit { + let emit_start = self.elapsed_us(); + self.write_line(&AuditLine { + kind: "audit", + recompute_us, + record: &outcome.record, + }); + // Flushed per record so a run ended with SIGKILL (or a matrix row + // that reads the file while the process is still up) still shows + // every decision made before that instant. The cost is measured, not + // assumed — it is inside `emit_us`. + let _ = self.writer.flush(); + self.elapsed_us().saturating_sub(emit_start).max(1) + } else { + 0 + }; + + self.metrics.record(metrics::Sample { + at_us, + gap_us, + recompute_us, + emit_us, + kind, + }); + self.last_completion_us = Some(self.elapsed_us()); + + if kind == EventKind::Tick { + self.ticks_since_summary += 1; + if self.ticks_since_summary >= SUMMARY_INTERVAL_TICKS { + self.ticks_since_summary = 0; + self.write_summary(now_us / 1_000); + } + } + } +} + +impl Drop for AuditSink { + /// The final O5 record. The observer thread drops its sink when the main + /// loop quits, so an ordinary ctrl-c leaves a complete summary behind + /// without the runner having to ask for one. + fn drop(&mut self) { + let at_ms = self.elapsed_us() / 1_000; + self.write_summary(at_ms); + } +} + +#[derive(Serialize)] +struct AuditLine<'a> { + kind: &'static str, + /// This record's own recompute cost, so a surprising row can be correlated + /// with a cost spike without cross-referencing the periodic summary. + recompute_us: u64, + #[serde(flatten)] + record: &'a AuditRecord, +} + +#[derive(Serialize)] +struct MetricsLine<'a> { + kind: &'static str, + at_ms: Millis, + #[serde(flatten)] + summary: &'a Summary, +} diff --git a/src/host/audit/tests.rs b/src/host/audit/tests.rs new file mode 100644 index 0000000..ce254df --- /dev/null +++ b/src/host/audit/tests.rs @@ -0,0 +1,896 @@ +//! Pure tests for the phase-5 auditor and its O5 metrics. +//! +//! Two things are being tested here and they are worth keeping distinct: +//! +//! - **Audit-layer behaviour** — the fan-out gate, record suppression, sequence +//! accounting, epoch reporting, and above all that every record carries the +//! *complete* candidate universe (§5.1). These are properties nothing else +//! tests, because nothing else exists at this layer. +//! - **A few §5.1 matrix shapes in fixture form** — row 1 (owner-bridge +//! forwarder), row 3 (two modules, one tainted), row 12 (AEC lifecycle). These +//! are *not* re-litigating phase 2, whose 57 tests already own those verdicts. +//! They exist so that a plumbing mistake between the engine and the record — +//! a dropped reason code, an inverted partition — fails here, at compile-time +//! speed, rather than only in front of a live daemon. +//! +//! The live half of the gate cannot live in this file by definition: a fixture +//! tests my model against my own assumptions, and §5's whole argument is that +//! only a live run tests my model against PipeWire. See the matrix runs recorded +//! in the phase-5 results file. + +use super::metrics::{BUCKET_LABELS, Metrics, QUEUE_THRESHOLD_US, Sample}; +use super::*; +use crate::host::aec::AecConfig; +use crate::host::observer::{EventKind, Readiness}; +use crate::host::taint::fixture::{self, Graph, NodeRef}; +use crate::host::taint::snapshot::{GraphSnapshot, MediaRole}; + +const AEC_MODULE: u64 = 7; +const TIMEOUT: Millis = 5_000; + +fn ready(snapshot: GraphSnapshot) -> Projection { + Projection { + snapshot, + pipewire_pulse_pid: Some(fixture::PULSE_PID), + graph_ready: true, + readiness: Readiness::Complete, + } +} + +fn not_ready(snapshot: GraphSnapshot, readiness: Readiness) -> Projection { + Projection { + snapshot, + pipewire_pulse_pid: Some(fixture::PULSE_PID), + graph_ready: false, + readiness, + } +} + +fn auditor_off() -> Auditor { + Auditor::new(AuditConfig { + aec: AecConfig::Off, + aec_timeout: TIMEOUT, + }) +} + +fn auditor_aec(index: u64) -> Auditor { + Auditor::new(AuditConfig { + aec: AecConfig::PulseModule(index), + aec_timeout: TIMEOUT, + }) +} + +/// One graph-triggered recompute at `now`. +fn observe(auditor: &mut Auditor, projection: &Projection, now: Millis) -> AuditOutcome { + auditor.observe(projection, EventKind::Graph, now) +} + +/// Candidate names split into (eligible, excluded-with-reason), which is how the +/// §5.1 rows are phrased. Names rather than serials so a failure reads as the +/// scenario rather than as an integer. +fn partition(body: &AuditBody) -> (Vec<&str>, Vec<(&str, &str)>) { + let eligible = body + .candidates + .iter() + .filter(|row| row.eligible) + .map(|row| row.name.as_deref().unwrap_or("")) + .collect(); + let excluded = body + .candidates + .iter() + .filter(|row| !row.eligible) + .map(|row| { + ( + row.name.as_deref().unwrap_or(""), + row.reason.unwrap_or(""), + ) + }) + .collect(); + (eligible, excluded) +} + +// ── the §5.1 structural requirement ─────────────────────────────────────── + +/// The record must contain **every** `Stream/Output/Audio` node, not only the +/// interesting ones. This is the property the whole exact-partition requirement +/// rests on: if the record could omit a candidate, then asserting a complete +/// partition over the record would still not constrain the graph. +#[test] +fn the_record_carries_the_complete_candidate_universe() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + graph.app_node("game", MediaRole::StreamOutput, 101); + graph.app_node("recorder", MediaRole::StreamInput, 102); + graph.device_node("speakers", MediaRole::Sink); + let projection = ready(graph.build()); + + let outcome = observe(&mut auditor_off(), &projection, 0); + let (eligible, excluded) = partition(&outcome.record.body); + + // Both playback streams, neither the capture stream nor the sink. Ordered by + // serial (creation order), which is what makes the partition assertions in + // every other row stable rather than dependent on a hash iteration. + assert_eq!(eligible, vec!["music", "game"]); + assert!(excluded.is_empty(), "unexpected exclusions: {excluded:?}"); + assert_eq!(outcome.record.body.candidates.len(), 2); + assert_eq!(outcome.record.body.eligible_count, 2); + assert_eq!(outcome.record.body.excluded_count, 0); +} + +/// The fail-closed default asserted at the boundary (impl plan §4, phase 2's +/// "one addition"): nothing in, nothing eligible — and, just as importantly, no +/// panic and no invented row. +#[test] +fn an_empty_graph_yields_an_empty_partition() { + let projection = ready(Graph::new().build()); + let outcome = observe(&mut auditor_off(), &projection, 0); + + assert!(outcome.record.body.candidates.is_empty()); + assert!(outcome.record.body.taint.is_empty()); + assert_eq!(outcome.record.body.eligible_count, 0); + assert_eq!(outcome.record.body.excluded_count, 0); + assert!(outcome.record.body.fan_out_permitted); +} + +/// `eligible_count + excluded_count` is the candidate count, always. A partition +/// that does not partition would let a row's two assertions both pass while the +/// record described no coherent state. +#[test] +fn the_counts_always_partition_the_candidates() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + graph.peerspeak_node("peerspeak-playback", 200); + let projection = ready(graph.build()); + + let body = observe(&mut auditor_off(), &projection, 0).record.body; + assert_eq!( + body.eligible_count + body.excluded_count, + body.candidates.len() + ); + assert_eq!(body.eligible().len(), body.eligible_count); + assert_eq!(body.excluded().len(), body.excluded_count); +} + +/// Every excluded row names a reason and every eligible row does not. The +/// §5.1 rows assert "excluded, with reason code" — a `None` reason on an +/// excluded row would make that assertion unwritable. +#[test] +fn reason_presence_is_exactly_the_exclusion() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + graph.peerspeak_node("peerspeak-playback", 200); + let projection = ready(graph.build()); + + for row in observe(&mut auditor_off(), &projection, 0) + .record + .body + .candidates + { + assert_eq!( + row.eligible, + row.reason.is_none(), + "row {row:?} has eligibility and reason out of step" + ); + } +} + +// ── readiness ───────────────────────────────────────────────────────────── + +/// No decision made from a partial graph is a decision. Note this is asserted on +/// the *eligible* half too: an implementation that reported nothing at all while +/// not ready would also be wrong, because the audit must still show what it can +/// see. +#[test] +fn a_not_ready_graph_excludes_every_candidate() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + graph.app_node("game", MediaRole::StreamOutput, 101); + let projection = not_ready(graph.build(), Readiness::Waiting); + + let body = observe(&mut auditor_off(), &projection, 0).record.body; + let (eligible, excluded) = partition(&body); + + assert!(eligible.is_empty()); + assert_eq!( + excluded, + vec![("music", "graph-not-ready"), ("game", "graph-not-ready"),] + ); + assert!(!body.graph_ready); +} + +/// The three ways `graph_ready` can be false are distinguishable in the record. +/// Collapsing them would make a timed-out observer — a fail-closed *fault* — +/// indistinguishable from an enumeration that is merely still running. +#[test] +fn the_epoch_distinguishes_the_ways_a_graph_can_be_unready() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + let snapshot = graph.build(); + + for (readiness, expected) in [ + (Readiness::Waiting, "waiting"), + (Readiness::TimedOut, "timed-out"), + // A completed epoch momentarily blocked on a current obligation: the + // interesting one, because `graph_ready` alone makes it look like a + // brand-new observer. + (Readiness::Complete, "complete"), + ] { + let projection = not_ready(snapshot.clone(), readiness); + let body = observe(&mut auditor_off(), &projection, 0).record.body; + assert_eq!(body.epoch, expected); + assert!(!body.graph_ready); + } + + let body = observe(&mut auditor_off(), &ready(snapshot), 0).record.body; + assert_eq!(body.epoch, "complete"); + assert!(body.graph_ready); +} + +// ── the fan-out gate (phase 4 → audit) ──────────────────────────────────── + +/// `--aec=off` leaves the gate open: `NotConfigured` is "there is no echo +/// canceller", not "we failed to find one". +#[test] +fn aec_off_leaves_the_gate_open() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + let projection = ready(graph.build()); + + let body = observe(&mut auditor_off(), &projection, 0).record.body; + assert_eq!(body.aec_state, "not-configured"); + assert!(body.fan_out_permitted); + assert_eq!(body.gate_reason, None); + assert_eq!(body.aec_module_id, None); + assert_eq!(partition(&body).0, vec!["music"]); +} + +/// While the configured identity has not been seen, nothing may fan out — +/// silence over echo — and the record says why in a code, not in prose. +#[test] +fn a_validating_gate_excludes_every_engine_eligible_candidate() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + graph.app_node("game", MediaRole::StreamOutput, 101); + let projection = ready(graph.build()); + + let body = observe(&mut auditor_aec(AEC_MODULE), &projection, 0) + .record + .body; + let (eligible, excluded) = partition(&body); + + assert_eq!(body.aec_state, "validating"); + assert!(!body.fan_out_permitted); + assert_eq!(body.gate_reason, Some("aec-validating")); + assert!(eligible.is_empty()); + assert_eq!( + excluded, + vec![("music", "aec-validating"), ("game", "aec-validating")] + ); +} + +/// A shut gate must not erase the engine's own reason codes. If it did, every +/// §5.1 row run under a shut gate would report one uniform code and the matrix +/// would stop constraining the taint engine at all — the record would say +/// "nothing may fan out" while hiding *which* nodes were tainted and how. +#[test] +fn a_shut_gate_preserves_the_engines_own_reasons() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + graph.peerspeak_node("peerspeak-playback", 200); + let projection = ready(graph.build()); + + let body = observe(&mut auditor_aec(AEC_MODULE), &projection, 0) + .record + .body; + let (_, excluded) = partition(&body); + + assert!(!body.fan_out_permitted); + assert_eq!( + excluded, + vec![ + ("music", "aec-validating"), + // Tagged, so it keeps the reason that actually applies to it. + ("peerspeak-playback", "peerspeak-owned"), + ] + ); +} + +/// The deadline is armed on the first ready graph, so a slow enumeration reads +/// as "unknown", not "absent" (the phase-4 user design call). Past it with the +/// identity never seen, the gate latches shut. +#[test] +fn the_gate_fails_closed_after_the_deadline() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + let snapshot = graph.build(); + let mut auditor = auditor_aec(AEC_MODULE); + + // Still enumerating well past the timeout: not a failure, because absence + // has not been established. + let waiting = not_ready(snapshot.clone(), Readiness::Waiting); + let body = observe(&mut auditor, &waiting, TIMEOUT * 3).record.body; + assert_eq!(body.aec_state, "validating"); + + // Ready arms the deadline; the clock has to advance past it from here. + let projection = ready(snapshot); + let body = observe(&mut auditor, &projection, TIMEOUT * 3).record.body; + assert_eq!(body.aec_state, "validating"); + + let body = observe(&mut auditor, &projection, TIMEOUT * 6 + 1) + .record + .body; + assert_eq!(body.aec_state, "failed"); + assert_eq!(body.gate_reason, Some("aec-failed")); + assert_eq!(partition(&body).1, vec![("music", "aec-failed")]); +} + +// ── §5.1 row 12: the AEC lifecycle ──────────────────────────────────────── + +/// Build the four nodes `module-echo-cancel` creates, all bearing one index: +/// two `Stream/*` legs plus the virtual sink/source pair (v3.4 §5.2). The +/// playback leg is the hazard — a `Stream/Output/Audio` wired to the speakers. +fn aec_nodes(graph: &mut Graph, index: u64) -> Vec { + vec![ + graph.module_node("echo-cancel-playback", MediaRole::StreamOutput, index), + graph.module_node("echo-cancel-capture", MediaRole::StreamInput, index), + graph.module_node("echo-cancel-sink", MediaRole::Sink, index), + graph.module_node("echo-cancel-source", MediaRole::Source, index), + ] +} + +/// §5.1 row 12: AEC loaded → validated, its playback leg excluded by identity +/// while everything else stays eligible → unloaded → `Revoked`, gate shut. +/// +/// The eligible half is the load-bearing assertion in the first phase: an +/// implementation that excluded the whole graph the moment an AEC appeared would +/// satisfy "the four nodes are excluded" and still be wrong. +#[test] +fn row_12_aec_loaded_then_unloaded_validates_then_revokes() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + let aec = aec_nodes(&mut graph, AEC_MODULE); + let mut auditor = auditor_aec(AEC_MODULE); + + let loaded = ready(graph.build()); + let body = observe(&mut auditor, &loaded, 0).record.body; + let (eligible, excluded) = partition(&body); + + assert_eq!(body.aec_state, "validated"); + assert!(body.fan_out_permitted); + assert_eq!(body.aec_module_id, Some(AEC_MODULE)); + assert_eq!(eligible, vec!["music"]); + assert_eq!(excluded, vec![("echo-cancel-playback", "aec-identity")]); + + // Every node bearing the index goes away: a real unload. + let unloaded = ready(graph.build_without(&aec)); + let body = observe(&mut auditor, &unloaded, 1).record.body; + let (eligible, excluded) = partition(&body); + + assert_eq!(body.aec_state, "revoked"); + assert!(!body.fan_out_permitted); + assert_eq!(body.gate_reason, Some("aec-revoked")); + assert_eq!(body.aec_module_id, None); + assert!(eligible.is_empty()); + assert_eq!(excluded, vec![("music", "aec-revoked")]); +} + +/// One leg corking is not a revocation (v3.4 §5.3). Getting this wrong turns an +/// ordinary cork into a share-wide audio stop, so the audit must report the +/// identity as still live. +#[test] +fn row_12_partial_leg_loss_does_not_revoke() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + let aec = aec_nodes(&mut graph, AEC_MODULE); + let mut auditor = auditor_aec(AEC_MODULE); + + let body = observe(&mut auditor, &ready(graph.build()), 0).record.body; + assert_eq!(body.aec_state, "validated"); + + // The capture leg alone disappears; three nodes still bear the index. + let partial = ready(graph.build_without(&aec[1..2])); + let body = observe(&mut auditor, &partial, 1).record.body; + + assert_eq!(body.aec_state, "validated"); + assert!(body.fan_out_permitted); + assert_eq!(partition(&body).0, vec!["music"]); +} + +/// Revocation is sticky terminal: module indices are reused verbatim across an +/// unload/reload (v3.4 §5.2 correction 3), so a reappearing index must not +/// resurrect the epoch and alias onto an unrelated module. A genuine reload gets +/// a fresh validator, never this one. +#[test] +fn row_12_a_reused_index_does_not_resurrect_a_revoked_epoch() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + let aec = aec_nodes(&mut graph, AEC_MODULE); + let mut auditor = auditor_aec(AEC_MODULE); + + observe(&mut auditor, &ready(graph.build()), 0); + let unloaded = graph.build_without(&aec); + let body = observe(&mut auditor, &ready(unloaded), 1).record.body; + assert_eq!(body.aec_state, "revoked"); + + // A second module comes back with the same index — different objects, + // identical number. + let mut reloaded = Graph::new(); + reloaded.app_node("music", MediaRole::StreamOutput, 100); + aec_nodes(&mut reloaded, AEC_MODULE); + let body = observe(&mut auditor, &ready(reloaded.build()), 2) + .record + .body; + + assert_eq!(body.aec_state, "revoked"); + assert!(!body.fan_out_permitted); + assert_eq!(body.gate_reason, Some("aec-revoked")); +} + +// ── §5.1 rows in fixture form (plumbing, not phase-2 verdicts) ──────────── + +/// §5.1 row 1: a `module-null-sink` + `module-loopback` forwarder. The output +/// leg is excluded across the **owner bridge** — naming the mechanism, not a +/// link walk — while an identically-shaped forwarder with no tainted input stays +/// eligible. The second half is what an exclude-everything build fails. +#[test] +fn row_1_owner_bridge_forwarder_with_an_untainted_control() { + let mut graph = Graph::new(); + // Tainted root: peerspeak's own call playback, feeding a sink the forwarder + // reads back out. + let call = graph.peerspeak_node("peerspeak-call", 200); + let sink = graph.module_node("tainted-null-sink", MediaRole::Sink, 30); + graph.link(call, sink); + let capture = graph.module_node("tainted-loopback-capture", MediaRole::StreamInput, 30); + let playback = graph.module_node("tainted-loopback-playback", MediaRole::StreamOutput, 30); + graph.link(sink, capture); + let _ = playback; + + // Control: the same shape, fed by nothing tainted. + let clean_sink = graph.module_node("clean-null-sink", MediaRole::Sink, 31); + let clean_capture = graph.module_node("clean-loopback-capture", MediaRole::StreamInput, 31); + let clean_playback = graph.module_node("clean-loopback-playback", MediaRole::StreamOutput, 31); + graph.link(clean_sink, clean_capture); + let _ = clean_playback; + + let body = observe(&mut auditor_off(), &ready(graph.build()), 0) + .record + .body; + let (eligible, excluded) = partition(&body); + + assert_eq!(eligible, vec!["clean-loopback-playback"]); + assert_eq!( + excluded, + vec![ + ("peerspeak-call", "peerspeak-owned"), + ("tainted-loopback-playback", "tainted-owner-bridge"), + ] + ); +} + +/// §5.1 row 3: two Pulse modules, one tainted input. **The other module's output +/// must be eligible** — this is the row that makes a wrong pipewire-pulse-PID +/// fusion observable, because fusing all Pulse-created nodes into one owner +/// would drag the innocent module's output leg down with the tainted one. +#[test] +fn row_3_one_tainted_module_does_not_taint_the_other() { + let mut graph = Graph::new(); + let call = graph.peerspeak_node("peerspeak-call", 200); + let sink = graph.module_node("null-sink-a", MediaRole::Sink, 40); + graph.link(call, sink); + let capture_a = graph.module_node("module-a-capture", MediaRole::StreamInput, 40); + let playback_a = graph.module_node("module-a-playback", MediaRole::StreamOutput, 40); + graph.link(sink, capture_a); + let _ = playback_a; + + // A second, entirely independent module reading an untainted source. + let mic = graph.device_node("microphone", MediaRole::Source); + let capture_b = graph.module_node("module-b-capture", MediaRole::StreamInput, 41); + let playback_b = graph.module_node("module-b-playback", MediaRole::StreamOutput, 41); + graph.link(mic, capture_b); + let _ = playback_b; + + let body = observe(&mut auditor_off(), &ready(graph.build()), 0) + .record + .body; + let (eligible, excluded) = partition(&body); + + assert_eq!(eligible, vec!["module-b-playback"]); + assert_eq!( + excluded, + vec![ + ("peerspeak-call", "peerspeak-owned"), + ("module-a-playback", "tainted-owner-bridge"), + ] + ); +} + +/// §5.1 row 7 (cycle prevention, v3.4 §6.2): a forwarder reading *another* +/// pixelpass host's capture sink must be excluded by its **named output +/// serial**, or two hosts sharing to each other build an audio cycle. +#[test] +fn row_7_a_forwarder_reading_another_hosts_capture_sink_is_excluded() { + let mut graph = Graph::new(); + // The other host's own client, in *this* graph — a node pointing at a client + // that does not exist would exercise the unresolved-owner path instead of the + // capture-sink-name path this row is about. + let other_client = graph.client(Some(fixture::PULSE_PID)); + let other_sink = graph.node( + "pixelpass_capture_deadbeef", + MediaRole::Sink, + fixture::app(other_client, 300), + ); + let capture = graph.module_node("cycle-loopback-capture", MediaRole::StreamInput, 50); + let playback = graph.module_node("cycle-loopback-playback", MediaRole::StreamOutput, 50); + graph.link(other_sink, capture); + let _ = playback; + graph.app_node("music", MediaRole::StreamOutput, 100); + + let body = observe(&mut auditor_off(), &ready(graph.build()), 0) + .record + .body; + let (eligible, excluded) = partition(&body); + + assert_eq!(eligible, vec!["music"]); + assert_eq!( + excluded, + vec![("cycle-loopback-playback", "tainted-owner-bridge")] + ); + // The sink itself is tainted, by the mechanism that names it. + let sink_taint = body + .taint + .iter() + .find(|row| row.name.as_deref() == Some("pixelpass_capture_deadbeef")) + .expect("the other host's capture sink must be tainted"); + assert_eq!(sink_taint.reason, "pixelpass-owned"); +} + +/// Sticky taint (§5.1 row 10) is reported as sticky, not silently folded into +/// an ordinary exclusion. The flag is how the audit distinguishes "this is +/// tainted right now" from "this was tainted and its owner has not fully torn +/// down" — two different things to be surprised by. +#[test] +fn sticky_exclusions_are_flagged_as_sticky() { + let mut graph = Graph::new(); + let call = graph.peerspeak_node("peerspeak-call", 200); + let sink = graph.module_node("null-sink", MediaRole::Sink, 60); + graph.link(call, sink); + let capture = graph.module_node("loopback-capture", MediaRole::StreamInput, 60); + graph.module_node("loopback-playback", MediaRole::StreamOutput, 60); + graph.link(sink, capture); + + let mut auditor = auditor_off(); + let body = observe(&mut auditor, &ready(graph.build()), 0).record.body; + let playback = body + .row_named("loopback-playback") + .expect("the output leg must be a candidate"); + assert!(!playback.eligible); + assert!(!playback.sticky, "first sight is not sticky"); + + // The tainted input leg goes away; the output leg lives on. + let body = observe(&mut auditor, &ready(graph.build_without(&[capture])), 1) + .record + .body; + let playback = body + .row_named("loopback-playback") + .expect("the output leg must still be a candidate"); + assert!(!playback.eligible); + assert!(playback.sticky, "the taint is carried over, and says so"); +} + +// ── record accounting ───────────────────────────────────────────────────── + +/// Ticks exist to drive the AEC deadline, not to describe the graph. Emitting an +/// identical record four times a second would bury the graph events the audit +/// exists to show — but a tick that *does* change something must still be +/// emitted, or a `Validating → Failed` transition (which only a tick can cause) +/// would never appear in the log at all. +#[test] +fn an_unchanged_tick_is_suppressed_but_a_changed_one_is_not() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + let projection = ready(graph.build()); + let mut auditor = auditor_aec(AEC_MODULE); + + assert!(auditor.observe(&projection, EventKind::Graph, 0).emit); + assert!( + !auditor.observe(&projection, EventKind::Tick, 100).emit, + "an identical tick record is noise" + ); + assert!( + !auditor.observe(&projection, EventKind::Tick, 200).emit, + "still noise" + ); + + // The deadline expires on a tick: the state changes, so this one is emitted. + let outcome = auditor.observe(&projection, EventKind::Tick, TIMEOUT + 1); + assert!(outcome.emit); + assert_eq!(outcome.record.body.aec_state, "failed"); +} + +/// A graph event always emits, even when the decision content is identical — a +/// suppressed graph event would erase the evidence that the graph changed at all, +/// and "PipeWire told us something and nothing moved" is itself a finding. +#[test] +fn an_unchanged_graph_event_still_emits() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + let projection = ready(graph.build()); + let mut auditor = auditor_off(); + + assert!(observe(&mut auditor, &projection, 0).emit); + assert!(observe(&mut auditor, &projection, 1).emit); +} + +/// `seq` counts every recompute, emitted or not, so a gap in the emitted +/// sequence is visibly a suppression rather than a lost line. Without this, a +/// reader cannot tell a quiet audit from a broken one. +#[test] +fn seq_counts_suppressed_recomputes_too() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + let projection = ready(graph.build()); + let mut auditor = auditor_off(); + + assert_eq!(observe(&mut auditor, &projection, 0).record.seq, 1); + let suppressed = auditor.observe(&projection, EventKind::Tick, 1); + assert!(!suppressed.emit); + assert_eq!(suppressed.record.seq, 2); + assert_eq!(observe(&mut auditor, &projection, 2).record.seq, 3); +} + +/// Suppression compares against the last record actually *written*, not the last +/// one computed. Comparing against the last computed record would let a change +/// that appears and reverts between two ticks vanish from the log entirely, +/// leaving a reader with a record that no longer matches the state. +#[test] +fn suppression_compares_against_the_last_emitted_record() { + let mut graph = Graph::new(); + graph.app_node("music", MediaRole::StreamOutput, 100); + let with_music = ready(graph.build()); + let empty = ready(Graph::new().build()); + let mut auditor = auditor_off(); + + assert!(observe(&mut auditor, &with_music, 0).emit); + // A tick sees a different graph and emits. + assert!(auditor.observe(&empty, EventKind::Tick, 1).emit); + // The next tick sees the original graph again — different from what was last + // written, so it must be emitted. + assert!(auditor.observe(&with_music, EventKind::Tick, 2).emit); + // And now it matches the last written record. + assert!(!auditor.observe(&with_music, EventKind::Tick, 3).emit); +} + +/// The trigger and clock are reported verbatim, which is what lets the O5 event +/// rate be recomputed from the record stream alone rather than trusted from the +/// summary. +#[test] +fn the_record_reports_its_trigger_and_clock() { + let projection = ready(Graph::new().build()); + let mut auditor = auditor_off(); + + let outcome = auditor.observe(&projection, EventKind::Graph, 42); + assert_eq!(outcome.record.trigger, "graph"); + assert_eq!(outcome.record.at_ms, 42); + + let outcome = auditor.observe(&projection, EventKind::Tick, 43); + assert_eq!(outcome.record.trigger, "tick"); + assert_eq!(outcome.record.at_ms, 43); +} + +/// The taint view spans every media role, not just candidates. A candidate's +/// exclusion is usually explained by taint on a node that is not itself a +/// candidate — the sink in the middle of a forwarder — and without that the +/// record shows the verdict but not the evidence. +#[test] +fn the_taint_view_covers_non_candidate_roles() { + let mut graph = Graph::new(); + let call = graph.peerspeak_node("peerspeak-call", 200); + let sink = graph.module_node("null-sink", MediaRole::Sink, 70); + graph.link(call, sink); + + let body = observe(&mut auditor_off(), &ready(graph.build()), 0) + .record + .body; + let tainted: Vec<(&str, &str)> = body + .taint + .iter() + .map(|row| (row.name.as_deref().unwrap_or("?"), row.reason)) + .collect(); + + assert!( + tainted.contains(&("null-sink", "tainted-upstream")), + "the sink is not a candidate but its taint is what explains the row: {tainted:?}" + ); + assert!(tainted.contains(&("peerspeak-call", "peerspeak-owned"))); +} + +/// A record must serialise to a single line. Newlines inside a JSON Lines +/// record would split one record into two unparseable ones — and node names come +/// from PipeWire properties, which are attacker-adjacent free text. +#[test] +fn a_record_serialises_to_exactly_one_line() { + let mut graph = Graph::new(); + graph.app_node("evil\nname\r\nwith breaks", MediaRole::StreamOutput, 100); + let projection = ready(graph.build()); + + let outcome = observe(&mut auditor_off(), &projection, 0); + let json = serde_json::to_string(&outcome.record).expect("a record must serialise"); + assert_eq!(json.lines().count(), 1, "record split across lines: {json}"); + assert!( + json.contains(r"evil\nname"), + "the name must survive escaped" + ); +} + +// ── O5 metrics ──────────────────────────────────────────────────────────── + +fn sample(kind: EventKind, at_us: u64, gap_us: u64, recompute_us: u64, emit_us: u64) -> Sample { + Sample { + at_us, + gap_us, + recompute_us, + emit_us, + kind, + } +} + +/// Bucket bounds are exclusive upper bounds, so a value exactly on a bound lands +/// in the next bucket up. Asserted because an off-by-one here silently shifts +/// the whole distribution the O5 conclusion rests on. +#[test] +fn histogram_bounds_are_exclusive_upper_bounds() { + let mut metrics = Metrics::default(); + for us in [0, 49, 50, 99_999, 100_000, 1_000_000] { + metrics.record(sample(EventKind::Graph, 0, 1_000, us, 0)); + } + let summary = metrics.summary(); + + assert_eq!( + summary.recompute_distribution, + vec![ + ("<50us", 2), // 0 and 49 + ("<100us", 1), // 50 + ("<100ms", 1), // 99_999 + (">=100ms", 2), // 100_000 and 1_000_000 + ] + ); + assert_eq!(summary.recompute_max_us, 1_000_000); +} + +/// The maximum is exact, not bucketed. O5 asks for the maximum specifically, and +/// ">= 100 ms" is not an answer to "how bad does it get?". +#[test] +fn the_maximum_is_exact_not_bucketed() { + let mut metrics = Metrics::default(); + metrics.record(sample(EventKind::Graph, 0, 1_000, 137, 0)); + metrics.record(sample(EventKind::Graph, 0, 1_000, 4_211, 0)); + metrics.record(sample(EventKind::Graph, 0, 1_000, 90, 0)); + + let summary = metrics.summary(); + assert_eq!(summary.recompute_max_us, 4_211); + assert_eq!(summary.recompute_mean_us, Some((137 + 4_211 + 90) / 3)); +} + +/// Nearest-rank quantiles over the buckets. +#[test] +fn quantiles_use_nearest_rank_over_the_buckets() { + let mut metrics = Metrics::default(); + // 99 fast samples and one very slow one: the tail must show up at p99 and + // nowhere earlier, which is the whole reason for reporting p99 at all. + for _ in 0..99 { + metrics.record(sample(EventKind::Graph, 0, 1_000, 10, 0)); + } + metrics.record(sample(EventKind::Graph, 0, 1_000, 200_000, 0)); + + let summary = metrics.summary(); + assert_eq!(summary.recompute_p50, Some("<50us")); + assert_eq!(summary.recompute_p90, Some("<50us")); + assert_eq!(summary.recompute_p99, Some("<50us")); + assert_eq!(summary.recompute_max_us, 200_000); +} + +#[test] +fn an_empty_histogram_reports_no_quantiles_and_no_rate() { + let summary = Metrics::default().summary(); + assert_eq!(summary.recompute_p50, None); + assert_eq!(summary.recompute_mean_us, None); + assert_eq!(summary.graph_events_per_sec, None); + assert_eq!(summary.busy_fraction, None); + assert_eq!(summary.recompute_max_us, 0); + assert!(summary.recompute_distribution.is_empty()); +} + +/// Ticks are counted separately from graph events. Folding them in would inflate +/// the measured event rate by a constant 4 Hz and hide the real graph churn — +/// which is the number O5 is actually about. +#[test] +fn ticks_do_not_count_toward_the_graph_event_rate() { + let mut metrics = Metrics::default(); + // Two graph events one second apart, with ticks in between. + metrics.record(sample(EventKind::Graph, 0, 0, 100, 0)); + for i in 1..4 { + metrics.record(sample(EventKind::Tick, i * 250_000, 250_000, 100, 0)); + } + metrics.record(sample(EventKind::Graph, 1_000_000, 250_000, 100, 0)); + + let summary = metrics.summary(); + assert_eq!(summary.graph_events, 2); + assert_eq!(summary.tick_events, 3); + // Span runs to the last sample's completion: 1_000_000 + 100 µs. + assert_eq!(summary.span_us, 1_000_100); + assert_eq!(summary.graph_events_per_sec, Some(2.0)); +} + +/// The queueing proxy: an event beginning within the threshold of the previous +/// sample's completion was almost certainly already waiting. The first sample is +/// never counted — it has no predecessor to have queued behind, and counting it +/// would put a phantom backlog in every run. +#[test] +fn the_queueing_proxy_counts_back_to_back_events_only() { + let mut metrics = Metrics::default(); + metrics.record(sample(EventKind::Graph, 0, 0, 100, 0)); + metrics.record(sample(EventKind::Graph, 100, QUEUE_THRESHOLD_US, 100, 0)); + metrics.record(sample( + EventKind::Graph, + 200, + QUEUE_THRESHOLD_US + 1, + 100, + 0, + )); + metrics.record(sample(EventKind::Graph, 300, 0, 100, 0)); + + let summary = metrics.summary(); + assert_eq!( + summary.queued_events, 2, + "exactly the two within the threshold, never the first sample" + ); + assert_eq!(summary.queue_threshold_us, QUEUE_THRESHOLD_US); +} + +/// Emission cost is tracked separately from recompute cost, and a suppressed +/// record contributes neither an emitted-record count nor an emit sample — +/// otherwise the logging distribution would be diluted by every tick that wrote +/// nothing. +#[test] +fn emission_cost_is_tracked_separately_from_recompute() { + let mut metrics = Metrics::default(); + metrics.record(sample(EventKind::Graph, 0, 0, 300, 80)); + metrics.record(sample(EventKind::Tick, 1_000, 900, 200, 0)); + metrics.record(sample(EventKind::Graph, 2_000, 900, 400, 120)); + + let summary = metrics.summary(); + assert_eq!(summary.emitted_records, 2); + assert_eq!(summary.emit_max_us, 120); + assert_eq!(summary.emit_mean_us, Some(100)); + assert_eq!( + summary.emit_distribution, + vec![("<100us", 1), ("<250us", 1)] + ); + // Busy time is recompute *and* logging: 300+80+200+400+120. + assert_eq!(summary.busy_us, 1_100); +} + +/// The busy fraction needs no inference, unlike the queueing proxy, so it is the +/// number the O5 verdict should lean on. +#[test] +fn the_busy_fraction_is_the_share_of_wall_time_spent_working() { + let mut metrics = Metrics::default(); + metrics.record(sample(EventKind::Graph, 0, 0, 100, 0)); + // Ends at 1_000_000 + 900 → a span of 1_000_900 µs with 1_000 µs of work. + metrics.record(sample(EventKind::Graph, 1_000_000, 999_900, 900, 0)); + + let summary = metrics.summary(); + assert_eq!(summary.busy_us, 1_000); + assert_eq!(summary.span_us, 1_000_900); + assert_eq!(summary.busy_fraction, Some(0.001)); +} + +/// Bucket labels and bounds must stay parallel, or the distribution mislabels +/// itself — a silent failure that would misreport every O5 result. +#[test] +fn bucket_labels_cover_every_bound_plus_overflow() { + assert_eq!( + BUCKET_LABELS.len(), + super::metrics::BUCKET_BOUNDS_US.len() + 1 + ); +} diff --git a/src/host/mod.rs b/src/host/mod.rs index 8f8ca5d..f02b0b6 100644 --- a/src/host/mod.rs +++ b/src/host/mod.rs @@ -1,5 +1,6 @@ pub mod aec; pub mod audio; +pub mod audit; mod capture; mod observer; mod pipeline; @@ -77,6 +78,12 @@ pub async fn run(opts: HostOpts) -> Result<()> { let cancel = signal::install_ctrl_c(); + // Phase 5 dry-run audit, off unless `PIXELPASS_AUDIO_AUDIT=1`. Read-only: + // it observes the graph and logs what phases 2–4 conclude, creating no + // links. Bound to a name so the handle lives as long as the run — dropping + // it stops the observer thread and flushes the final O5 summary. + let _audio_audit = audit::run::spawn_if_enabled()?; + let endpoint = endpoint::bind(opts.relay.as_deref()).await?; // Relay-only ticket: wait for the home relay to connect, then keep only diff --git a/src/host/observer/adapter.rs b/src/host/observer/adapter.rs index de06f9e..25289cf 100644 --- a/src/host/observer/adapter.rs +++ b/src/host/observer/adapter.rs @@ -5,7 +5,7 @@ //! consumers running outside the PipeWire thread. use super::classify::DeviceClaim; -use super::{LinkEndpoints, NodeObservation, Projection, RegEvent, RegistryModel}; +use super::{EventKind, LinkEndpoints, NodeObservation, Projection, RegEvent, RegistryModel}; use crate::host::audio::parse_object_serial; use crate::host::taint::snapshot::{ ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial, @@ -22,6 +22,27 @@ use std::time::{Duration, Instant}; const READINESS_TIMEOUT_MILLIS: u64 = 2_000; const TICK_INTERVAL: Duration = Duration::from_millis(250); +/// A consumer that sees **every** projection, one per applied registry event, +/// on the observer thread. +/// +/// This exists because polling [`RegistryObserverHandle::latest`] coalesces, and +/// some consumers cannot tolerate that. Phase 4's AEC validator is the concrete +/// case: it detects a module unload by observing the *empty gap* before the next +/// module appears, and PipeWire reuses module indices verbatim across an +/// unload/reload (v3.4 §5.2 correction 3), so a consumer that misses the gap +/// silently aliases a fresh module onto a dead module's validated identity. +/// +/// **Implementations run inline on the PipeWire loop thread.** Whatever they do +/// delays the next registry callback, so they must be bounded and must not +/// block. The phase-5 audit is the only implementor and measures its own cost +/// for exactly this reason. +pub trait ProjectionSink: Send { + /// `now_us` is monotonic microseconds since the observer started — the same + /// clock that drives [`RegEvent::Tick`], so a sink's notion of time cannot + /// drift from the readiness epoch's. + fn on_projection(&mut self, projection: &Projection, kind: EventKind, now_us: u64); +} + /// Tokio-side access to the observer's most recent coherent projection. pub struct RegistryObserverHandle { latest: Arc>>, @@ -32,13 +53,22 @@ pub struct RegistryObserverHandle { impl RegistryObserverHandle { /// Spawn the read-only PipeWire registry observer. pub fn spawn() -> Result { + Self::spawn_with_sink(None) + } + + /// Spawn the observer with a per-event [`ProjectionSink`] attached. + /// + /// The sink is moved onto the observer thread and dropped when that thread + /// exits, which is what lets a sink emit a final summary on shutdown without + /// the caller arranging one. + pub fn spawn_with_sink(sink: Option>) -> Result { let latest = Arc::new(Mutex::new(None)); let latest_for_thread = Arc::clone(&latest); let (shutdown_tx, shutdown_rx) = pw::channel::channel::<()>(); let thread = std::thread::Builder::new() .name("pixelpass-pw-observer".to_string()) .spawn(move || { - if let Err(e) = run_observer(latest_for_thread, shutdown_rx) { + if let Err(e) = run_observer(latest_for_thread, shutdown_rx, sink) { tracing::warn!( "registry observer: libpipewire thread exited with error: {e:#}" ); @@ -89,19 +119,34 @@ struct ObserverState { latest: Arc>>, last_candidate: Option, live_globals: BTreeMap>, + sink: Option>, + started_at: Instant, } impl ObserverState { - fn new(latest: Arc>>) -> Self { + /// `started_at` is the observer's single time origin, shared with the + /// readiness tick timer — so a sink's `now_us` and a `RegEvent::Tick`'s + /// `now` are the same clock, not two that drift. + fn new( + latest: Arc>>, + sink: Option>, + started_at: Instant, + ) -> Self { Self { model: RegistryModel::new(0, READINESS_TIMEOUT_MILLIS), latest, last_candidate: None, live_globals: BTreeMap::new(), + sink, + started_at, } } fn apply(&mut self, event: RegEvent) { + // Taken before the model consumes the event: the sink is told what kind + // of observation produced the projection, and deriving that from the + // event itself is what stops the two from ever disagreeing. + let kind = event.kind(); self.model.apply(event); let candidate = self.model.pulse_pid_candidate(); @@ -111,18 +156,30 @@ impl ObserverState { let comm = std::fs::read_to_string(format!("/proc/{pid}/comm")) .ok() .map(|comm| comm.trim_end_matches(['\r', '\n']).to_string()); + // Folded into the model directly rather than through `apply`, so + // 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 }); } } - self.publish(); + self.publish(kind); } - fn publish(&self) { + fn publish(&mut self, kind: EventKind) { + let projection = self.model.project(); + if let Some(sink) = self.sink.as_mut() { + let now_us = u64::try_from(self.started_at.elapsed().as_micros()).unwrap_or(u64::MAX); + sink.on_projection(&projection, kind, now_us); + } + // Published after the sink has seen it, so the projection is moved + // rather than cloned — the snapshot is the largest thing the observer + // owns and this runs on every event. *self .latest .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(self.model.project()); + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(projection); } /// Record the global's id and apply its add event as one step, so the @@ -166,6 +223,7 @@ impl ObserverState { fn run_observer( latest: Arc>>, shutdown_rx: pw::channel::Receiver<()>, + sink: Option>, ) -> Result<()> { let started_at = Instant::now(); let main_loop = @@ -176,7 +234,7 @@ fn run_observer( .connect_rc(None) .context("pw core connect failed (is the daemon running?)")?; let registry = core.get_registry_rc().context("pw get_registry failed")?; - let state = Rc::new(RefCell::new(ObserverState::new(latest))); + let state = Rc::new(RefCell::new(ObserverState::new(latest, sink, started_at))); let main_loop_for_shutdown = main_loop.clone(); let _shutdown_receiver = shutdown_rx.attach(main_loop.loop_(), move |()| { diff --git a/src/host/observer/mod.rs b/src/host/observer/mod.rs index ced862f..e340de6 100644 --- a/src/host/observer/mod.rs +++ b/src/host/observer/mod.rs @@ -133,6 +133,42 @@ pub enum RegEvent { Tick { now: Millis }, } +/// What kind of observation drove a projection. +/// +/// Derived from the event itself ([`RegEvent::kind`]) rather than passed +/// alongside it, so a consumer's view of "was this a real graph change?" cannot +/// disagree with what the model was actually fed. The distinction matters to the +/// phase-5 audit twice over: ticks arrive at a constant rate and would inflate +/// any measured graph-event rate, and a record that is identical to the previous +/// 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` + /// probe, or the server sync. + Graph, + /// The periodic clock sample. Carries no graph information; it exists so the + /// readiness timeout and the AEC validation deadline have a clock. + Tick, +} + +impl EventKind { + pub fn code(self) -> &'static str { + match self { + Self::Graph => "graph", + Self::Tick => "tick", + } + } +} + +impl RegEvent { + pub fn kind(&self) -> EventKind { + match self { + Self::Tick { .. } => EventKind::Tick, + _ => EventKind::Graph, + } + } +} + /// 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. @@ -174,6 +210,14 @@ pub struct Projection { pub snapshot: GraphSnapshot, pub pipewire_pulse_pid: Option, pub graph_ready: bool, + /// The sticky readiness epoch behind `graph_ready`. Carried so a consumer + /// can tell the three not-ready causes apart — enumeration still in flight + /// ([`Readiness::Waiting`]), a fail-closed timeout ([`Readiness::TimedOut`]), + /// or a completed epoch momentarily blocked on a current obligation + /// ([`Readiness::Complete`] with `graph_ready == false`). `graph_ready` + /// alone collapses all three into "no". The phase-5 audit reports it as the + /// epoch column; nothing gates on it. + pub readiness: Readiness, } /// The live model. Folds [`RegEvent`]s; project with [`RegistryModel::project`]. @@ -496,6 +540,7 @@ impl RegistryModel { snapshot, pipewire_pulse_pid: self.pulse_pid(), graph_ready: self.graph_ready(), + readiness: self.readiness, } } } diff --git a/src/host/taint/mod.rs b/src/host/taint/mod.rs index bd39aa6..4d7f38a 100644 --- a/src/host/taint/mod.rs +++ b/src/host/taint/mod.rs @@ -100,8 +100,11 @@ pub mod owner; pub mod snapshot; +// `pub` so the phase-5 audit's pure tests can drive the auditor with the same +// graph builder the taint fixtures use — one fixture vocabulary, so an audit +// test and a taint test describing the same topology cannot drift apart. #[cfg(test)] -mod fixture; +pub mod fixture; #[cfg(test)] mod tests; diff --git a/src/main.rs b/src/main.rs index 24ca729..b01fddf 100644 --- a/src/main.rs +++ b/src/main.rs @@ -52,6 +52,13 @@ async fn main() -> Result<()> { return repair::run().await; } + // Read-only diagnostic: observe the graph, report what the audio-exclusion + // engine concludes, create nothing. Placed before the host/viewer dispatch + // because it is neither — it shares no screen and connects to no peer. + if cli.audit_audio { + return host::audit::run::run_standalone().await; + } + if cli.reconfigure { return interactive::run_reconfigure().await; }