//! Pure Phase-6 fan-out planning.
//!
//! This module converts one coherent taint decision into exact per-port link
//! specifications and owns the pure state around the mutation edge. PipeWire
//! object creation remains behind [`LinkMutation`], implemented by the registry
//! adapter on its main-loop thread. That split keeps policy and the link-manager
//! matrix deterministic while making it impossible for Tokio to address a
//! recyclable global id directly.
//!
//! Port identity is `audio.channel`, never registry enumeration order or a
//! recyclable global id. An unknown or incompatible layout makes that stream
//! unsupported; it never produces a guessed partial capture.

use std::collections::{BTreeMap, BTreeSet};

use super::aec::{AecConfig, AecState, AecValidator};
use super::observer::{EventKind, Millis, Projection};
use super::taint::snapshot::{
    GlobalId, GraphSnapshot, MediaRole, PortDirection, PortSnapshot, Serial,
};
use super::taint::{
    Decisions, ECHO_CANCEL_GROUP_PREFIX, ExclusionCtx, Reason, StickyState, evaluate,
};
use crate::common::output::AudioExclusionEvent;
use tokio::sync::mpsc;

const AEC_VALIDATION_TIMEOUT_MILLIS: Millis = 5_000;

/// One endpoint identity carried from a coherent graph snapshot to the
/// immediate mutation edge. IDs select the server objects; serials prove the
/// IDs still name those same objects. Neither may be omitted.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub(super) struct DesiredEndpoint {
    pub(super) node_id: GlobalId,
    pub(super) node_serial: Serial,
    pub(super) port_id: GlobalId,
    pub(super) port_serial: Serial,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub(super) struct DesiredLink {
    pub(super) output: DesiredEndpoint,
    pub(super) input: DesiredEndpoint,
}

impl DesiredLink {
    /// Revalidate every recyclable id against its never-recycled serial.
    /// Called by the production mutator immediately before `create_object`.
    pub(super) fn is_current(self, snapshot: &GraphSnapshot) -> bool {
        endpoint_is_current(snapshot, self.output) && endpoint_is_current(snapshot, self.input)
    }
}

fn endpoint_is_current(snapshot: &GraphSnapshot, endpoint: DesiredEndpoint) -> bool {
    snapshot
        .node(endpoint.node_serial)
        .is_some_and(|node| node.id == endpoint.node_id)
        && snapshot.ports().any(|port| {
            port.serial == endpoint.port_serial
                && port.id == endpoint.port_id
                && port.node == endpoint.node_id
        })
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum PlanIssue {
    CaptureSinkMissing,
    CaptureSinkNotASink,
    MissingOutputPorts,
    MissingCapturePorts,
    UnidentifiedChannel,
    AmbiguousCaptureChannel,
    DuplicateOutputChannel,
    IncompatibleChannelLayout,
    ExclusiveCapturePort,
    LinkCreationFailed,
}

impl PlanIssue {
    pub(super) fn code(self) -> &'static str {
        match self {
            Self::CaptureSinkMissing => "capture-sink-missing",
            Self::CaptureSinkNotASink => "capture-sink-not-a-sink",
            Self::MissingOutputPorts => "missing-output-ports",
            Self::MissingCapturePorts => "missing-capture-ports",
            Self::UnidentifiedChannel => "unidentified-channel",
            Self::AmbiguousCaptureChannel => "ambiguous-capture-channel",
            Self::DuplicateOutputChannel => "duplicate-output-channel",
            Self::IncompatibleChannelLayout => "incompatible-channel-layout",
            Self::ExclusiveCapturePort => "exclusive-capture-port",
            Self::LinkCreationFailed => "link-creation-failed",
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum StreamPlan {
    Excluded { reason: Reason },
    Unsupported { issue: PlanIssue },
    Capture { links: BTreeSet<DesiredLink> },
}

/// State of one retained link proxy as observed from its bound PipeWire Link.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum ManagedLinkState {
    Pending,
    Active,
    Failed,
}

/// The narrow mutation capability available to the Phase-6 controller. The
/// adapter owns the PipeWire implementation and all proxies. Tests inject a
/// ledger with no PipeWire dependency.
pub(super) trait LinkMutation {
    /// Replace the owned set with `desired`, dropping obsolete proxies first.
    /// Implementations must revalidate [`DesiredLink::is_current`] against
    /// this exact snapshot immediately before each creation.
    fn reconcile(
        &mut self,
        snapshot: &GraphSnapshot,
        desired: &BTreeSet<DesiredLink>,
    ) -> BTreeMap<DesiredLink, ManagedLinkState>;
}

/// A projection consumer that is explicitly allowed to mutate the graph.
/// Kept separate from the read-only observer sink trait so the Phase-5 audit
/// still has no mutation capability in its type signature.
pub(super) trait MutationProjectionSink: Send {
    fn on_projection(
        &mut self,
        projection: &Projection,
        kind: EventKind,
        now_us: u64,
        links: &mut dyn LinkMutation,
    );

    /// Retarget a running controller to the exact serial of a recreated
    /// connection-owned capture sink, then reconcile immediately against the
    /// observer's current coherent projection. The adapter invokes this only
    /// on its PipeWire loop thread, keeping old-link revocation and new-link
    /// creation ordered with ordinary registry observations.
    fn replace_capture_sink(
        &mut self,
        capture_sink: Serial,
        projection: &Projection,
        now_us: u64,
        links: &mut dyn LinkMutation,
    );
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum StreamCaptureState {
    Excluded { reason: Reason },
    Unsupported { issue: PlanIssue },
    Gated { aec: AecState },
    Linking,
    Captured,
}

/// Phases 2, 4, and the pure port planner folded into one per-event controller.
/// It never retains a global id between callbacks: only sticky serial identity
/// and the capture sink's serial survive a recompute.
pub(super) struct FanoutController {
    capture_sink: Serial,
    validator: AecValidator,
    sticky: StickyState,
    states: BTreeMap<Serial, StreamCaptureState>,
    foreign_aec_groups: BTreeSet<String>,
    status_tx: Option<mpsc::UnboundedSender<AudioExclusionEvent>>,
}

impl FanoutController {
    #[cfg(test)]
    pub(super) fn new(capture_sink: Serial, aec: AecConfig) -> Self {
        Self::build(capture_sink, aec, None)
    }

    /// Construct the production controller with a non-blocking status path.
    /// The PipeWire callback only enqueues owned records; JSON serialization
    /// and stdout I/O happen on a Tokio task outside the graph loop.
    pub(super) fn with_status_sender(
        capture_sink: Serial,
        aec: AecConfig,
        status_tx: mpsc::UnboundedSender<AudioExclusionEvent>,
    ) -> Self {
        Self::build(capture_sink, aec, Some(status_tx))
    }

    fn build(
        capture_sink: Serial,
        aec: AecConfig,
        status_tx: Option<mpsc::UnboundedSender<AudioExclusionEvent>>,
    ) -> Self {
        Self {
            capture_sink,
            validator: AecValidator::new(aec, AEC_VALIDATION_TIMEOUT_MILLIS),
            sticky: StickyState::default(),
            states: BTreeMap::new(),
            foreign_aec_groups: BTreeSet::new(),
            status_tx,
        }
    }

    fn emit_status(&self, event: AudioExclusionEvent) {
        if let Some(status_tx) = &self.status_tx {
            // A closed receiver means the machine front-end no longer accepts
            // notices. Capture ownership remains safe and should not be
            // poisoned merely because its best-effort stdout consumer ended.
            let _ = status_tx.send(event);
        }
    }

    #[cfg(test)]
    fn states(&self) -> &BTreeMap<Serial, StreamCaptureState> {
        &self.states
    }
}

impl MutationProjectionSink for FanoutController {
    fn on_projection(
        &mut self,
        projection: &Projection,
        _kind: EventKind,
        now_us: u64,
        links: &mut dyn LinkMutation,
    ) {
        let prior_aec_state = self.validator.state();
        self.validator
            .observe(&projection.snapshot, projection.graph_ready, now_us / 1_000);
        let aec_state = self.validator.state();
        let aec_status = if prior_aec_state != aec_state {
            self.validator
                .configured_module_id()
                .and_then(|module_index| match aec_state {
                    AecState::Failed => Some(AudioExclusionEvent::AecFailed { module_index }),
                    AecState::Revoked => Some(AudioExclusionEvent::AecRevoked { module_index }),
                    _ => None,
                })
        } else {
            None
        };
        let configured_aec = self.validator.configured_module_id();
        let foreign_aec_groups: BTreeSet<String> = projection
            .snapshot
            .nodes()
            .filter_map(|node| {
                let group = node.props.link_group.as_ref()?;
                (group.starts_with(ECHO_CANCEL_GROUP_PREFIX)
                    && configured_aec
                        .is_none_or(|module| node.props.pulse_module_id != Some(module)))
                .then(|| group.clone())
            })
            .collect();
        let new_foreign_groups: Vec<String> = foreign_aec_groups
            .difference(&self.foreign_aec_groups)
            .cloned()
            .collect();
        let ctx = ExclusionCtx {
            aec_module_id: self.validator.validated_module_id(),
            pipewire_pulse_pid: projection.pipewire_pulse_pid,
            pixelpass_owned: BTreeSet::from([self.capture_sink]),
            graph_ready: projection.graph_ready,
        };
        let (decisions, sticky) = evaluate(&projection.snapshot, &ctx, &self.sticky);
        self.sticky = sticky;
        let plans = plan(&projection.snapshot, &decisions, self.capture_sink);

        let desired: BTreeSet<DesiredLink> = if self.validator.fan_out_permitted() {
            plans
                .values()
                .filter_map(|plan| match plan {
                    StreamPlan::Capture { links } => Some(links.iter().copied()),
                    _ => None,
                })
                .flatten()
                .collect()
        } else {
            BTreeSet::new()
        };
        let managed = links.reconcile(&projection.snapshot, &desired);

        let next_states: BTreeMap<Serial, StreamCaptureState> = plans
            .into_iter()
            .map(|(serial, plan)| {
                let state = match plan {
                    StreamPlan::Excluded { reason } => StreamCaptureState::Excluded { reason },
                    StreamPlan::Unsupported { issue } => StreamCaptureState::Unsupported { issue },
                    StreamPlan::Capture { .. } if !self.validator.fan_out_permitted() => {
                        StreamCaptureState::Gated { aec: aec_state }
                    }
                    StreamPlan::Capture { links: planned } => {
                        let link_states: Vec<ManagedLinkState> = planned
                            .iter()
                            .filter_map(|link| managed.get(link).copied())
                            .collect();
                        if link_states.len() != planned.len()
                            || link_states.contains(&ManagedLinkState::Failed)
                        {
                            StreamCaptureState::Unsupported {
                                issue: PlanIssue::LinkCreationFailed,
                            }
                        } else if link_states
                            .iter()
                            .all(|state| *state == ManagedLinkState::Active)
                        {
                            StreamCaptureState::Captured
                        } else {
                            StreamCaptureState::Linking
                        }
                    }
                };
                (serial, state)
            })
            .collect();
        let mut stream_statuses = Vec::new();
        for (&serial, state) in &next_states {
            if self.states.get(&serial) == Some(state) {
                continue;
            }
            match state {
                StreamCaptureState::Unsupported { issue } => {
                    tracing::warn!(
                        serial = serial.0,
                        reason = issue.code(),
                        "audio fan-out: stream unsupported"
                    );
                    stream_statuses.push(AudioExclusionEvent::StreamUnsupported {
                        stream_serial: serial.0,
                        reason: issue.code(),
                    });
                }
                StreamCaptureState::Captured => tracing::info!(
                    serial = serial.0,
                    "audio fan-out: every required link is active"
                ),
                _ => {}
            }
        }
        self.states = next_states;
        self.foreign_aec_groups = foreign_aec_groups;

        // `reconcile` above has already dropped every link forbidden by an AEC
        // transition or a newly unsafe graph. Only report after that mutation
        // boundary, so `aec_revoked` can never race ahead of link revocation.
        if let Some(event) = aec_status {
            self.emit_status(event);
        }
        for link_group in new_foreign_groups {
            self.emit_status(AudioExclusionEvent::ForeignAecWarning { link_group });
        }
        for event in stream_statuses {
            self.emit_status(event);
        }
    }

    fn replace_capture_sink(
        &mut self,
        capture_sink: Serial,
        projection: &Projection,
        now_us: u64,
        links: &mut dyn LinkMutation,
    ) {
        if self.capture_sink == capture_sink {
            return;
        }
        self.capture_sink = capture_sink;
        // This is an explicit owner control message rather than a model event,
        // so it must not be coalesced behind the next registry callback. The
        // normal projection path is still the single place that evaluates,
        // plans, revokes stale proxies, and determines all-links-ACTIVE.
        self.on_projection(projection, EventKind::Graph, now_us, links);
    }
}

/// Plan every candidate in the decision universe against one capture sink.
///
/// The caller must pass `decisions` produced from this exact `snapshot`. The
/// mutation edge upholds that condition structurally by evaluating and planning
/// inline in one observer callback; keeping this function pure makes the port
/// matrix independently falsifiable.
pub(super) fn plan(
    snapshot: &GraphSnapshot,
    decisions: &Decisions,
    capture_sink: Serial,
) -> BTreeMap<Serial, StreamPlan> {
    let sink = snapshot.node(capture_sink);
    let sink_issue = match sink {
        None => Some(PlanIssue::CaptureSinkMissing),
        Some(node) if node.role != MediaRole::Sink => Some(PlanIssue::CaptureSinkNotASink),
        Some(_) => None,
    };

    decisions
        .candidates
        .iter()
        .map(|(&serial, decision)| {
            let stream_plan = if let Some(reason) = decision.reason() {
                StreamPlan::Excluded { reason }
            } else if let Some(issue) = sink_issue {
                StreamPlan::Unsupported { issue }
            } else {
                let node = snapshot
                    .node(serial)
                    .expect("a decision from this snapshot names a live node");
                match pair_ports(
                    snapshot.ports_of(node.id),
                    snapshot.ports_of(sink.expect("validated above").id),
                    node.id,
                    node.serial,
                    sink.expect("validated above").id,
                    sink.expect("validated above").serial,
                ) {
                    Ok(links) => StreamPlan::Capture { links },
                    Err(issue) => StreamPlan::Unsupported { issue },
                }
            };
            (serial, stream_plan)
        })
        .collect()
}

fn pair_ports<'a>(
    candidate_ports: impl Iterator<Item = &'a PortSnapshot>,
    capture_ports: impl Iterator<Item = &'a PortSnapshot>,
    output_node: GlobalId,
    output_node_serial: Serial,
    input_node: GlobalId,
    input_node_serial: Serial,
) -> Result<BTreeSet<DesiredLink>, PlanIssue> {
    let outputs: Vec<&PortSnapshot> = candidate_ports
        .filter(|port| port.direction == PortDirection::Out && !port.monitor)
        .collect();
    if outputs.is_empty() {
        return Err(PlanIssue::MissingOutputPorts);
    }
    let inputs: Vec<&PortSnapshot> = capture_ports
        .filter(|port| port.direction == PortDirection::In)
        .collect();
    if inputs.is_empty() {
        return Err(PlanIssue::MissingCapturePorts);
    }
    if inputs.iter().any(|port| port.exclusive) {
        return Err(PlanIssue::ExclusiveCapturePort);
    }

    let mut inputs_by_channel: BTreeMap<&str, &PortSnapshot> = BTreeMap::new();
    for port in &inputs {
        let channel = port
            .channel
            .as_deref()
            .ok_or(PlanIssue::UnidentifiedChannel)?;
        if inputs_by_channel.insert(channel, port).is_some() {
            return Err(PlanIssue::AmbiguousCaptureChannel);
        }
    }

    let mut output_channels = BTreeSet::new();
    let mut links = BTreeSet::new();
    for output in outputs {
        let channel = output
            .channel
            .as_deref()
            .ok_or(PlanIssue::UnidentifiedChannel)?;
        if !output_channels.insert(channel) {
            return Err(PlanIssue::DuplicateOutputChannel);
        }
        if channel == "MONO" {
            for &input in inputs_by_channel.values() {
                links.insert(DesiredLink {
                    output: DesiredEndpoint {
                        node_id: output_node,
                        node_serial: output_node_serial,
                        port_id: output.id,
                        port_serial: output.serial,
                    },
                    input: DesiredEndpoint {
                        node_id: input_node,
                        node_serial: input_node_serial,
                        port_id: input.id,
                        port_serial: input.serial,
                    },
                });
            }
            continue;
        }
        let Some(&input) = inputs_by_channel.get(channel) else {
            return Err(PlanIssue::IncompatibleChannelLayout);
        };
        links.insert(DesiredLink {
            output: DesiredEndpoint {
                node_id: output_node,
                node_serial: output_node_serial,
                port_id: output.id,
                port_serial: output.serial,
            },
            input: DesiredEndpoint {
                node_id: input_node,
                node_serial: input_node_serial,
                port_id: input.id,
                port_serial: input.serial,
            },
        });
    }
    Ok(links)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::host::observer::Readiness;
    use crate::host::taint::fixture::{Graph, PULSE_PID};
    use crate::host::taint::snapshot::{MediaRole, NodeProps, PortDirection};
    use crate::host::taint::{ExclusionCtx, StickyState, evaluate};

    fn decisions(graph: &Graph) -> (GraphSnapshot, Decisions) {
        let snapshot = graph.build();
        let (decisions, _) = evaluate(
            &snapshot,
            &ExclusionCtx {
                pipewire_pulse_pid: Some(PULSE_PID),
                graph_ready: true,
                ..ExclusionCtx::default()
            },
            &StickyState::default(),
        );
        (snapshot, decisions)
    }

    fn stereo_graph() -> (Graph, Serial, Serial, [GlobalId; 4]) {
        let mut graph = Graph::new();
        let app = graph.app_node("music", MediaRole::StreamOutput, 42);
        let sink = graph.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43);
        let app_fl = graph.port_on_channel(app, PortDirection::Out, false, Some("FL"));
        let app_fr = graph.port_on_channel(app, PortDirection::Out, false, Some("FR"));
        let sink_fl = graph.port_on_channel(sink, PortDirection::In, false, Some("FL"));
        let sink_fr = graph.port_on_channel(sink, PortDirection::In, false, Some("FR"));
        (
            graph,
            app.serial,
            sink.serial,
            [app_fl, app_fr, sink_fl, sink_fr],
        )
    }

    fn endpoint(
        snapshot: &GraphSnapshot,
        node_serial: Serial,
        port_id: GlobalId,
    ) -> DesiredEndpoint {
        let node = snapshot.node(node_serial).expect("fixture node");
        let port = snapshot
            .ports()
            .find(|port| port.id == port_id && port.node == node.id)
            .expect("fixture port");
        DesiredEndpoint {
            node_id: node.id,
            node_serial,
            port_id,
            port_serial: port.serial,
        }
    }

    fn projection(graph: &Graph) -> Projection {
        projection_from(graph.build())
    }

    fn projection_from(snapshot: GraphSnapshot) -> Projection {
        Projection {
            snapshot,
            pipewire_pulse_pid: Some(PULSE_PID),
            graph_ready: true,
            readiness: Readiness::Complete,
        }
    }

    #[derive(Default)]
    struct FakeLinks {
        held: BTreeSet<DesiredLink>,
        creates: usize,
        drops: usize,
        next_state: BTreeMap<DesiredLink, ManagedLinkState>,
    }

    impl LinkMutation for FakeLinks {
        fn reconcile(
            &mut self,
            snapshot: &GraphSnapshot,
            desired: &BTreeSet<DesiredLink>,
        ) -> BTreeMap<DesiredLink, ManagedLinkState> {
            let current: BTreeSet<DesiredLink> = desired
                .iter()
                .copied()
                .filter(|link| link.is_current(snapshot))
                .collect();
            self.drops += self.held.difference(&current).count();
            self.creates += current.difference(&self.held).count();
            self.held = current.clone();
            current
                .into_iter()
                .map(|link| {
                    let state = self
                        .next_state
                        .get(&link)
                        .copied()
                        .unwrap_or(ManagedLinkState::Active);
                    (link, state)
                })
                .collect()
        }
    }

    fn drive(controller: &mut FanoutController, graph: &Graph, links: &mut FakeLinks) {
        controller.on_projection(&projection(graph), EventKind::Graph, 1_000, links);
    }

    fn controller_with_status(
        capture_sink: Serial,
        aec: AecConfig,
    ) -> (
        FanoutController,
        mpsc::UnboundedReceiver<AudioExclusionEvent>,
    ) {
        let (status_tx, status_rx) = mpsc::unbounded_channel();
        (
            FanoutController::with_status_sender(capture_sink, aec, status_tx),
            status_rx,
        )
    }

    #[test]
    fn stereo_ports_pair_by_channel_not_enumeration_order() {
        let (graph, app, sink, [app_fl, app_fr, sink_fl, sink_fr]) = stereo_graph();
        let (snapshot, decisions) = decisions(&graph);
        assert_eq!(
            plan(&snapshot, &decisions, sink).get(&app),
            Some(&StreamPlan::Capture {
                links: BTreeSet::from([
                    DesiredLink {
                        output: endpoint(&snapshot, app, app_fl),
                        input: endpoint(&snapshot, sink, sink_fl),
                    },
                    DesiredLink {
                        output: endpoint(&snapshot, app, app_fr),
                        input: endpoint(&snapshot, sink, sink_fr),
                    },
                ]),
            })
        );
    }

    #[test]
    fn mono_fans_to_both_stereo_inputs() {
        let mut graph = Graph::new();
        let app = graph.app_node("mono", MediaRole::StreamOutput, 42);
        let sink = graph.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43);
        let mono = graph.port_on_channel(app, PortDirection::Out, false, Some("MONO"));
        let left = graph.port_on_channel(sink, PortDirection::In, false, Some("FL"));
        let right = graph.port_on_channel(sink, PortDirection::In, false, Some("FR"));
        let (snapshot, decisions) = decisions(&graph);
        let StreamPlan::Capture { links } =
            plan(&snapshot, &decisions, sink.serial)[&app.serial].clone()
        else {
            panic!("mono stream must be plannable");
        };
        assert_eq!(links.len(), 2);
        assert_eq!(
            links
                .iter()
                .map(|link| link.output.port_id)
                .collect::<BTreeSet<_>>(),
            BTreeSet::from([mono])
        );
        assert_eq!(
            links
                .iter()
                .map(|link| link.input.port_id)
                .collect::<BTreeSet<_>>(),
            BTreeSet::from([left, right])
        );
    }

    #[test]
    fn unknown_or_incompatible_layout_is_unsupported_not_guessed() {
        let (mut graph, app, sink, _) = stereo_graph();
        let surround = graph.app_node("surround", MediaRole::StreamOutput, 44);
        graph.port_on_channel(surround, PortDirection::Out, false, Some("FC"));
        let unknown = graph.app_node("unknown", MediaRole::StreamOutput, 45);
        graph.port_on_channel(unknown, PortDirection::Out, false, None);
        let (snapshot, decisions) = decisions(&graph);
        let plans = plan(&snapshot, &decisions, sink);
        assert!(matches!(plans[&app], StreamPlan::Capture { .. }));
        assert_eq!(
            plans[&surround.serial],
            StreamPlan::Unsupported {
                issue: PlanIssue::IncompatibleChannelLayout
            }
        );
        assert_eq!(
            plans[&unknown.serial],
            StreamPlan::Unsupported {
                issue: PlanIssue::UnidentifiedChannel
            }
        );
    }

    #[test]
    fn excluded_candidate_never_produces_a_link_plan() {
        let (mut graph, app, sink, _) = stereo_graph();
        let call = graph.peerspeak_node("call", 7);
        graph.port_on_channel(call, PortDirection::Out, false, Some("FL"));
        let (snapshot, decisions) = decisions(&graph);
        let plans = plan(&snapshot, &decisions, sink);
        assert!(matches!(plans[&app], StreamPlan::Capture { .. }));
        assert_eq!(
            plans[&call.serial],
            StreamPlan::Excluded {
                reason: Reason::PeerspeakOwned
            }
        );
    }

    #[test]
    fn replacing_the_capture_sink_changes_every_target_port() {
        let (mut graph, app, old_sink, _) = stereo_graph();
        let new_sink = graph.native_virtual_node("pixelpass_capture_new", MediaRole::Sink, 46);
        let new_fl = graph.port_on_channel(new_sink, PortDirection::In, false, Some("FL"));
        let new_fr = graph.port_on_channel(new_sink, PortDirection::In, false, Some("FR"));
        let (snapshot, decisions) = decisions(&graph);
        let old = plan(&snapshot, &decisions, old_sink);
        let new = plan(&snapshot, &decisions, new_sink.serial);
        let StreamPlan::Capture { links: old } = &old[&app] else {
            panic!("old sink plan");
        };
        let StreamPlan::Capture { links: new } = &new[&app] else {
            panic!("new sink plan");
        };
        assert_ne!(old, new);
        assert_eq!(
            new.iter()
                .map(|link| link.input.port_id)
                .collect::<BTreeSet<_>>(),
            BTreeSet::from([new_fl, new_fr])
        );
    }

    #[test]
    fn replacing_the_capture_sink_drops_stale_links_and_recaptures() {
        let (mut graph, app, old_sink, _) = stereo_graph();
        let new_sink = graph.native_virtual_node("pixelpass_capture_new", MediaRole::Sink, 46);
        let new_fl = graph.port_on_channel(new_sink, PortDirection::In, false, Some("FL"));
        let new_fr = graph.port_on_channel(new_sink, PortDirection::In, false, Some("FR"));
        let projection = projection(&graph);
        let mut links = FakeLinks::default();
        let mut controller = FanoutController::new(old_sink, AecConfig::Off);

        controller.on_projection(&projection, EventKind::Graph, 1_000, &mut links);
        assert_eq!(links.creates, 2);
        assert_eq!(controller.states()[&app], StreamCaptureState::Captured);

        controller.replace_capture_sink(new_sink.serial, &projection, 2_000, &mut links);

        assert_eq!(links.drops, 2, "every old-sink proxy must be dropped");
        assert_eq!(links.creates, 4, "both replacement links must be created");
        assert_eq!(controller.states()[&app], StreamCaptureState::Captured);
        assert_eq!(
            links
                .held
                .iter()
                .map(|link| link.input.port_id)
                .collect::<BTreeSet<_>>(),
            BTreeSet::from([new_fl, new_fr]),
            "only replacement-sink ports may remain held"
        );
    }

    #[test]
    fn serial_revalidation_rejects_a_recycled_node_id_before_mutation() {
        let (mut graph, app, sink, _) = stereo_graph();
        let (snapshot, decisions) = decisions(&graph);
        let StreamPlan::Capture { links } = &plan(&snapshot, &decisions, sink)[&app] else {
            panic!("fixture must plan links");
        };
        assert!(links.iter().all(|link| link.is_current(&snapshot)));

        let old_app_id = snapshot.node(app).expect("old app").id;
        let reborn = graph.node_with_id(
            "reborn",
            MediaRole::StreamOutput,
            old_app_id,
            NodeProps::default(),
        );
        graph.port_on_channel(reborn, PortDirection::Out, false, Some("FL"));
        graph.port_on_channel(reborn, PortDirection::Out, false, Some("FR"));
        let recycled = graph.build_without(&[crate::host::taint::fixture::NodeRef {
            serial: app,
            id: old_app_id,
        }]);
        assert!(
            links.iter().all(|link| !link.is_current(&recycled)),
            "the same global id with a new serial must not be addressable"
        );
    }

    #[test]
    fn duplicate_projection_is_idempotent_and_all_links_must_activate() {
        let (graph, app, sink, _) = stereo_graph();
        let first_projection = projection(&graph);
        let decisions = evaluate(
            &first_projection.snapshot,
            &ExclusionCtx {
                graph_ready: true,
                pipewire_pulse_pid: Some(PULSE_PID),
                ..ExclusionCtx::default()
            },
            &StickyState::default(),
        )
        .0;
        let StreamPlan::Capture { links: planned } =
            &plan(&first_projection.snapshot, &decisions, sink)[&app]
        else {
            panic!("fixture must plan links");
        };
        let pending = *planned.iter().next().expect("stereo link");

        let mut links = FakeLinks::default();
        links.next_state.insert(pending, ManagedLinkState::Pending);
        let mut controller = FanoutController::new(sink, AecConfig::Off);
        drive(&mut controller, &graph, &mut links);
        assert_eq!(links.creates, 2);
        assert_eq!(
            controller.states()[&app],
            StreamCaptureState::Linking,
            "one pending channel prevents a captured verdict"
        );

        drive(&mut controller, &graph, &mut links);
        assert_eq!(
            links.creates, 2,
            "the same node must not get a second link set"
        );
        links.next_state.clear();
        drive(&mut controller, &graph, &mut links);
        assert_eq!(controller.states()[&app], StreamCaptureState::Captured);
    }

    #[test]
    fn newly_unsafe_ancestry_drops_owned_links_before_any_replacement() {
        let (mut graph, app, sink, _) = stereo_graph();
        let mut links = FakeLinks::default();
        let mut controller = FanoutController::new(sink, AecConfig::Off);
        drive(&mut controller, &graph, &mut links);
        assert_eq!(links.held.len(), 2);

        let call = graph.peerspeak_node("call", 7);
        let app_id = graph.build().node(app).expect("app remains live").id;
        graph.link(
            call,
            crate::host::taint::fixture::NodeRef {
                serial: app,
                id: app_id,
            },
        );
        drive(&mut controller, &graph, &mut links);
        assert!(links.held.is_empty());
        assert_eq!(links.drops, 2);
        assert!(matches!(
            controller.states()[&app],
            StreamCaptureState::Excluded { .. }
        ));
    }

    #[test]
    fn eligible_late_arrival_is_positively_captured() {
        let mut graph = Graph::new();
        let sink = graph.native_virtual_node("pixelpass_capture_test", MediaRole::Sink, 43);
        graph.port_on_channel(sink, PortDirection::In, false, Some("FL"));
        graph.port_on_channel(sink, PortDirection::In, false, Some("FR"));
        let mut links = FakeLinks::default();
        let mut controller = FanoutController::new(sink.serial, AecConfig::Off);
        drive(&mut controller, &graph, &mut links);
        assert!(links.held.is_empty());

        let late = graph.app_node("late", MediaRole::StreamOutput, 44);
        graph.port_on_channel(late, PortDirection::Out, false, Some("FL"));
        graph.port_on_channel(late, PortDirection::Out, false, Some("FR"));
        drive(&mut controller, &graph, &mut links);
        assert_eq!(links.creates, 2);
        assert_eq!(
            controller.states()[&late.serial],
            StreamCaptureState::Captured
        );
    }

    #[test]
    fn configured_but_unvalidated_aec_gates_every_mutation() {
        let (graph, app, sink, _) = stereo_graph();
        let mut links = FakeLinks::default();
        let mut controller = FanoutController::new(sink, AecConfig::PulseModule(999));
        drive(&mut controller, &graph, &mut links);
        assert_eq!(links.creates, 0);
        assert!(matches!(
            controller.states()[&app],
            StreamCaptureState::Gated {
                aec: AecState::Validating
            }
        ));
    }

    #[test]
    fn aec_validation_deadline_emits_the_failed_cause_once() {
        let (graph, app, sink, _) = stereo_graph();
        let mut links = FakeLinks::default();
        let (mut controller, mut status_rx) =
            controller_with_status(sink, AecConfig::PulseModule(999));
        drive(&mut controller, &graph, &mut links);
        assert!(matches!(
            controller.states()[&app],
            StreamCaptureState::Gated {
                aec: AecState::Validating
            }
        ));

        controller.on_projection(
            &projection(&graph),
            EventKind::Tick,
            (AEC_VALIDATION_TIMEOUT_MILLIS + 2) * 1_000,
            &mut links,
        );
        assert_eq!(
            status_rx.try_recv(),
            Ok(AudioExclusionEvent::AecFailed { module_index: 999 })
        );
        assert!(matches!(
            controller.states()[&app],
            StreamCaptureState::Gated {
                aec: AecState::Failed
            }
        ));

        controller.on_projection(
            &projection(&graph),
            EventKind::Tick,
            (AEC_VALIDATION_TIMEOUT_MILLIS + 3) * 1_000,
            &mut links,
        );
        assert_eq!(
            status_rx.try_recv(),
            Err(mpsc::error::TryRecvError::Empty),
            "a sticky failed state must not repeat the event on every tick"
        );
    }

    #[test]
    fn link_creation_failure_is_sticky_for_the_same_intent() {
        let (graph, app, sink, _) = stereo_graph();
        let first_projection = projection(&graph);
        let decisions = evaluate(
            &first_projection.snapshot,
            &ExclusionCtx {
                graph_ready: true,
                pipewire_pulse_pid: Some(PULSE_PID),
                ..ExclusionCtx::default()
            },
            &StickyState::default(),
        )
        .0;
        let StreamPlan::Capture { links: planned } =
            &plan(&first_projection.snapshot, &decisions, sink)[&app]
        else {
            panic!("fixture must plan links");
        };

        let mut links = FakeLinks::default();
        links.next_state.insert(
            *planned.iter().next().expect("stereo link"),
            ManagedLinkState::Failed,
        );
        let (mut controller, mut status_rx) = controller_with_status(sink, AecConfig::Off);
        drive(&mut controller, &graph, &mut links);
        assert_eq!(
            controller.states()[&app],
            StreamCaptureState::Unsupported {
                issue: PlanIssue::LinkCreationFailed
            }
        );
        assert_eq!(links.creates, 2);
        assert_eq!(
            status_rx.try_recv(),
            Ok(AudioExclusionEvent::StreamUnsupported {
                stream_serial: app.0,
                reason: "link-creation-failed",
            })
        );
        drive(&mut controller, &graph, &mut links);
        assert_eq!(
            links.creates, 2,
            "an identical failed intent must not hot-loop"
        );
        assert_eq!(
            status_rx.try_recv(),
            Err(mpsc::error::TryRecvError::Empty),
            "an unchanged unsupported stream must not repeat its event"
        );
    }

    #[test]
    fn aec_revocation_drops_previously_active_fanout() {
        let (mut graph, app, sink, _) = stereo_graph();
        let aec = graph.module_node("aec-playback", MediaRole::StreamOutput, 999);
        graph.port_on_channel(aec, PortDirection::Out, false, Some("FL"));
        let mut links = FakeLinks::default();
        let (mut controller, mut status_rx) =
            controller_with_status(sink, AecConfig::PulseModule(999));
        drive(&mut controller, &graph, &mut links);
        assert_eq!(controller.states()[&app], StreamCaptureState::Captured);
        assert_eq!(links.held.len(), 2);

        controller.on_projection(
            &projection_from(graph.build_without(&[aec])),
            EventKind::Graph,
            2_000,
            &mut links,
        );
        assert!(links.held.is_empty());
        assert_eq!(links.drops, 2);
        assert_eq!(
            controller.states()[&app],
            StreamCaptureState::Gated {
                aec: AecState::Revoked
            }
        );
        assert_eq!(
            status_rx.try_recv(),
            Ok(AudioExclusionEvent::AecRevoked { module_index: 999 })
        );
    }

    #[test]
    fn second_echo_cancel_group_emits_one_foreign_warning() {
        use crate::host::taint::fixture::pulse_module;

        let (mut graph, _app, sink, _) = stereo_graph();
        let own_client = graph.client_of_module(999);
        let mut own_props = pulse_module(own_client, 999, PULSE_PID);
        own_props.link_group = Some("echo-cancel-owned-13".to_string());
        graph.node("owned-aec-playback", MediaRole::StreamOutput, own_props);

        let foreign_client = graph.client_of_module(1_000);
        let mut foreign_props = pulse_module(foreign_client, 1_000, PULSE_PID);
        foreign_props.link_group = Some("echo-cancel-foreign-14".to_string());
        graph.node(
            "foreign-aec-playback",
            MediaRole::StreamOutput,
            foreign_props.clone(),
        );
        graph.node("foreign-aec-capture", MediaRole::StreamInput, foreign_props);

        let mut links = FakeLinks::default();
        let (mut controller, mut status_rx) =
            controller_with_status(sink, AecConfig::PulseModule(999));
        drive(&mut controller, &graph, &mut links);
        assert_eq!(
            status_rx.try_recv(),
            Ok(AudioExclusionEvent::ForeignAecWarning {
                link_group: "echo-cancel-foreign-14".to_string(),
            })
        );
        assert_eq!(
            status_rx.try_recv(),
            Err(mpsc::error::TryRecvError::Empty),
            "all legs in one foreign group collapse to one warning"
        );

        drive(&mut controller, &graph, &mut links);
        assert_eq!(
            status_rx.try_recv(),
            Err(mpsc::error::TryRecvError::Empty),
            "a still-live foreign group must not warn on every graph event"
        );
    }

    #[test]
    fn unindexed_native_echo_cancel_group_is_still_foreign() {
        let (mut graph, _app, sink, _) = stereo_graph();
        graph.group_node(
            "native-aec-playback",
            MediaRole::StreamOutput,
            "echo-cancel-native-1",
            77,
        );
        let mut links = FakeLinks::default();
        let (mut controller, mut status_rx) = controller_with_status(sink, AecConfig::Off);

        drive(&mut controller, &graph, &mut links);
        assert_eq!(
            status_rx.try_recv(),
            Ok(AudioExclusionEvent::ForeignAecWarning {
                link_group: "echo-cancel-native-1".to_string(),
            })
        );
    }
}