host/aec: phase 4 — AEC identity validation state machine (pure core)

Bounded, read-only state machine that validates peerspeak's live
echo-cancel module identity before the taint engine trusts it, filling
the last ExclusionCtx field (aec_module_id). Design v3.4 §5.2/§5.3,
impl plan §4.

States (v3.4 §5.3 verbatim): NotConfigured / Validating / Validated /
Failed / Revoked. No fan-out while Validating; Failed and Revoked are
sticky terminals so a reused module index (indices ARE reused, §5.2
correction 3) cannot alias a Revoked epoch onto an unrelated reload.
Revocation is loss of the whole identity (every node bearing the index
gone), never one leg corking. The Failed deadline is armed only on the
first graph_ready, so a slow initial enumeration is "unknown" not
"absent" and never times out spuriously.

parse_aec_arg handles --aec=off|pulse-module:<idx> (D5): bare-u64
decimal accepted past u32::MAX, rejecting sign/whitespace/non-digit/
overflow/unknown-form.

22 pure tests (the exit-gate transition matrix), cargo test --bins
green (143 unit + 1 ignored live), clippy -D warnings + fmt clean.

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

src/host/aec/mod.rs

@@ -0,0 +1,261 @@
+//! Phase 4 — the AEC identity validation state machine (impl plan §4, design
+//! v3.4 §5.2/§5.3).
+//!
+//! peerspeak's echo canceller (`module-echo-cancel`) creates four graph nodes
+//! that all carry `pulse.module.id == <the index pactl returned>`, and the
+//! playback leg among them is a `Stream/Output/Audio` node wired straight to
+//! the speakers — a fan-out candidate that would copy the whole remote call
+//! into the share unless it is excluded (v3.4 §5.2, measured ≈desktop level).
+//! The taint engine (phase 2) already excludes it *given* the module index in
+//! [`ExclusionCtx::aec_module_id`](crate::host::taint::ExclusionCtx); this
+//! module is what decides, at runtime and fail-closed, whether that index may
+//! be trusted and handed over.
+//!
+//! **Why a state machine and not a one-shot check (v3.4 §5.3).** The identity
+//! is an *observed correlation on PipeWire 1.6.8*, not a documented contract,
+//! and a start-time enumeration races in both directions: peerspeak's
+//! `enable()` returns before the playback hazard leg is even in the graph, and
+//! pixelpass's capture spawns lazily on the first viewer, at a moment peerspeak
+//! does not control. So validation is a bounded epoch, and the identity can be
+//! *lost* mid-share (the module unloads) as well as *gained*.
+//!
+//! **The two traps this is shaped around:**
+//!
+//! - **Revocation is loss of the whole module identity, not one leg corking**
+//!   (v3.4 §5.3). Each [`AecValidator::observe`] rescans the snapshot for *any*
+//!   node bearing the index; [`AecState::Validated`] drops to
+//!   [`AecState::Revoked`] only when that set becomes **empty**. A single leg
+//!   corking or relinking (still ≥1 present) stays `Validated` — getting this
+//!   wrong turns a normal cork into a spurious share-wide audio stop.
+//! - **Module indices are reused verbatim across unload/reload** (v3.4 §5.2
+//!   correction 3 — both a reload's module index *and* its `node.link-group`
+//!   came back byte-identical, and node ids were recycled *and reassigned
+//!   across legs*). So [`AecState::Failed`] and [`AecState::Revoked`] are
+//!   **sticky terminal**: a later node reappearing with the same index does
+//!   **not** un-revoke and alias onto the new module. A genuine reload gets a
+//!   *fresh* [`AecValidator`] (peerspeak re-tells pixelpass the index on every
+//!   load), never a resurrected one.
+//!
+//! **Scope.** This is the validation state machine + `--aec` parsing only.
+//! Foreign / second-AEC detection (a non-owned `echo-cancel-*` group, v3.4
+//! §5.4 / D3) and the `foreign_aec_warning`/`aec_failed`/`aec_revoked` status
+//! *events* are phase 6's, which reads this machine's [`AecState`]. Wiring the
+//! parsed [`AecConfig`] out of the CLI and calling [`AecValidator::observe`]
+//! in the recompute loop is integration (phases 5/8). The node-side
+//! `pulse.module.id` parse (JSON-number-vs-string, u64-not-u32) is phase 3's
+//! adapter; this module consumes the already-parsed
+//! [`NodeProps::pulse_module_id`](crate::host::taint::snapshot::NodeProps).
+
+#![allow(dead_code)] // Wired into `--aec` parsing + the recompute loop by later phases.
+
+#[cfg(test)]
+mod tests;
+
+use crate::host::observer::Millis;
+use crate::host::taint::snapshot::GraphSnapshot;
+
+/// The parsed `--aec=off|pulse-module:<idx>` argument (decision D5).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AecConfig {
+    /// `--aec=off` — peerspeak's AEC is not in play, so there is nothing to
+    /// exclude and fan-out proceeds with no AEC identity. Not the same as an
+    /// *absent* argument (that default is the caller's; see [`parse_aec_arg`]).
+    Off,
+    /// `--aec=pulse-module:<idx>` — validate this live module index before
+    /// trusting it. The index is compared as `u64`, never `u32` (v3.4 §5.2).
+    PulseModule(u64),
+}
+
+/// Why an `--aec` argument was rejected. Rejection is fatal at the CLI edge —
+/// there is no fail-closed *default* index, because a wrong index would exclude
+/// the wrong node (or nothing), so a malformed value must not silently become
+/// "no AEC".
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AecParseError {
+    /// The value was empty.
+    Empty,
+    /// Not `off` and not `pulse-module:...`.
+    UnknownForm,
+    /// `pulse-module:` with nothing after the colon.
+    MissingIndex,
+    /// The index was not a bare `u64` decimal (sign, whitespace, non-digit, or
+    /// `> u64::MAX`).
+    InvalidIndex,
+}
+
+/// Parse one `--aec` value. `off` and `pulse-module:<idx>` are the only forms.
+///
+/// The index accepts values `> u32::MAX` (v3.4 §5.2: `pulse.module.id` sits
+/// next to the `object.serial` u32-truncation bug, so it is only ever compared
+/// as `u64`) and rejects anything Rust's [`u64`] parser rejects — a leading
+/// sign, surrounding or interior whitespace, non-decimal digits, and overflow.
+/// Matching is exact and case-sensitive: the argument is machine-generated by
+/// peerspeak from `EchoCancelGuard::module_index`, not typed by a user.
+pub fn parse_aec_arg(value: &str) -> Result<AecConfig, AecParseError> {
+    if value.is_empty() {
+        return Err(AecParseError::Empty);
+    }
+    if value == "off" {
+        return Ok(AecConfig::Off);
+    }
+    if let Some(index) = value.strip_prefix("pulse-module:") {
+        if index.is_empty() {
+            return Err(AecParseError::MissingIndex);
+        }
+        // A bare decimal only: reject a leading sign (Rust's `u64` parser
+        // accepts `+7`), interior/surrounding whitespace, and any non-digit,
+        // before letting the parser catch overflow. Leading zeros are harmless.
+        if !index.bytes().all(|b| b.is_ascii_digit()) {
+            return Err(AecParseError::InvalidIndex);
+        }
+        return index
+            .parse::<u64>()
+            .map(AecConfig::PulseModule)
+            .map_err(|_| AecParseError::InvalidIndex);
+    }
+    Err(AecParseError::UnknownForm)
+}
+
+/// The validation epoch (v3.4 §5.3, verbatim).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AecState {
+    /// `--aec=off` — no AEC identity, fan-out proceeds with no exclusion.
+    /// Terminal.
+    NotConfigured,
+    /// Waiting for the first node bearing the index. **No fan-out occurs here**
+    /// — silence is the safe direction. Ends at `Validated` on first sight, or
+    /// `Failed` once the graph is fully enumerated and the bounded deadline
+    /// passes with the index never seen.
+    Validating,
+    /// The index was observed live. Fan-out is permitted, excluding that
+    /// identity transitively (phase 2 / v3.4 §6.1).
+    Validated,
+    /// The deadline expired with the index never observed. **Fail closed** — no
+    /// fan-out; the caller reports a capability failure rather than sharing.
+    /// Sticky terminal.
+    Failed,
+    /// The whole module identity disappeared mid-share (every node bearing the
+    /// index gone). **Stop fan-out now** and drop the owned link proxies; do
+    /// not keep the numeric index and hope, because it is reused. Sticky
+    /// terminal — see the module header's second trap.
+    Revoked,
+}
+
+/// The bounded, read-only AEC identity validator. Fold the live graph in with
+/// [`AecValidator::observe`] once per recompute; read the result with
+/// [`AecValidator::state`], [`AecValidator::fan_out_permitted`], and
+/// [`AecValidator::validated_module_id`].
+#[derive(Clone, Debug)]
+pub struct AecValidator {
+    /// The index to validate. `None` iff [`AecConfig::Off`] (state stays
+    /// [`AecState::NotConfigured`] forever).
+    target: Option<u64>,
+    state: AecState,
+    /// The `Validating → Failed` budget, applied *after* the deadline is armed.
+    timeout: Millis,
+    /// The absolute `Failed` deadline, armed the first time the graph reports
+    /// ready (the "registry sync barrier" of v3.4 §5.3) and never re-armed —
+    /// `graph_ready` is dynamic and can flap, but the epoch budget must not
+    /// restart. `None` until then: while the initial enumeration is still in
+    /// flight, a not-yet-seen index is *unknown*, not *absent*, so it must not
+    /// time out to `Failed`.
+    deadline: Option<Millis>,
+}
+
+impl AecValidator {
+    /// `timeout` is the `Validating → Failed` budget, counted from the moment
+    /// the graph first becomes ready (not from construction). An `Off` config
+    /// starts (and stays) [`AecState::NotConfigured`].
+    pub fn new(config: AecConfig, timeout: Millis) -> Self {
+        match config {
+            AecConfig::Off => Self {
+                target: None,
+                state: AecState::NotConfigured,
+                timeout,
+                deadline: None,
+            },
+            AecConfig::PulseModule(index) => Self {
+                target: Some(index),
+                state: AecState::Validating,
+                timeout,
+                deadline: None,
+            },
+        }
+    }
+
+    pub fn state(&self) -> AecState {
+        self.state
+    }
+
+    /// The validated index to place in
+    /// [`ExclusionCtx::aec_module_id`](crate::host::taint::ExclusionCtx) —
+    /// `Some` **only** in [`AecState::Validated`]. `None` everywhere else,
+    /// including `NotConfigured` (no AEC ⇒ nothing to exclude) and the
+    /// fail-closed states (whose `None` must be paired with
+    /// [`Self::fan_out_permitted`] `== false`, i.e. no fan-out at all — *not*
+    /// a fan-out that merely skips AEC exclusion).
+    pub fn validated_module_id(&self) -> Option<u64> {
+        match self.state {
+            AecState::Validated => self.target,
+            _ => None,
+        }
+    }
+
+    /// Whether fan-out may proceed at all right now. True only in
+    /// [`AecState::NotConfigured`] (fan out, no exclusion) and
+    /// [`AecState::Validated`] (fan out, excluding the identity). `Validating`,
+    /// `Failed` and `Revoked` all forbid it — silence over echo.
+    pub fn fan_out_permitted(&self) -> bool {
+        matches!(self.state, AecState::NotConfigured | AecState::Validated)
+    }
+
+    /// Fold one recompute's view of the graph into the machine.
+    ///
+    /// `graph_ready` is the observer's dynamic readiness
+    /// ([`Projection::graph_ready`](crate::host::observer::Projection)); `now`
+    /// is a monotonic millisecond clock. Positive evidence (a node bearing the
+    /// index) is authoritative and validates regardless of `graph_ready` —
+    /// seeing the node *is* seeing it — but the `Failed` deadline only begins
+    /// once `graph_ready` has first become true, so a slow initial enumeration
+    /// can never masquerade as a genuinely-absent module.
+    pub fn observe(&mut self, snapshot: &GraphSnapshot, graph_ready: bool, now: Millis) {
+        // `Off` (NotConfigured) and both sticky terminals are no-ops: there is
+        // nothing to look for, and a reappearing reused index must not revive a
+        // Failed/Revoked epoch (v3.4 §5.2 correction 3).
+        let Some(target) = self.target else {
+            return;
+        };
+        match self.state {
+            AecState::Validating => {
+                if self.index_present(snapshot, target) {
+                    self.state = AecState::Validated;
+                    return;
+                }
+                // Arm the deadline once, on the first ready graph.
+                if self.deadline.is_none() && graph_ready {
+                    self.deadline = Some(now.saturating_add(self.timeout));
+                }
+                if self.deadline.is_some_and(|deadline| now >= deadline) {
+                    self.state = AecState::Failed;
+                }
+            }
+            AecState::Validated => {
+                // Revocation is the whole identity gone, not one leg corking.
+                if !self.index_present(snapshot, target) {
+                    self.state = AecState::Revoked;
+                }
+            }
+            AecState::NotConfigured | AecState::Failed | AecState::Revoked => {}
+        }
+    }
+
+    /// Whether any node in the snapshot bears the target module index. The same
+    /// exact-`u64`-equality predicate the taint engine roots on
+    /// (`taint/mod.rs`), kept here so "is the identity live?" has one
+    /// definition.
+    fn index_present(&self, snapshot: &GraphSnapshot, target: u64) -> bool {
+        snapshot
+            .nodes()
+            .any(|node| node.props.pulse_module_id == Some(target))
+    }
+}