pixelflow-core 0.1.0

//! Immutable graph construction and Phase 1 validation.

use crate::{
    ChromaSubsampling, FilterOptions, FormatDescriptor, FormatFamily, Rational, SampleType,
};

use semisafe::slice::get as semisafe_get;
use semisafe::slice::get_mut as semisafe_get_mut;

/// Stable graph node identifier.
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct NodeId(usize);

impl NodeId {
    /// Creates node ID from zero-based storage index.
    #[must_use]
    pub const fn new(index: usize) -> Self {
        Self(index)
    }

    /// Returns zero-based storage index.
    #[must_use]
    pub const fn index(self) -> usize {
        self.0
    }
}

/// Immutable script-facing clip handle.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub struct Clip {
    node_id: NodeId,
}

impl Clip {
    /// Creates clip handle for existing node.
    #[must_use]
    pub const fn new(node_id: NodeId) -> Self {
        Self { node_id }
    }

    /// Returns backing node ID.
    #[must_use]
    pub const fn node_id(self) -> NodeId {
        self.node_id
    }
}

/// Filter output axes that may differ from the first input clip.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct FilterChangeSet {
    /// Output format may differ from the first input format.
    pub format: bool,
    /// Output resolution may differ from the first input resolution.
    pub resolution: bool,
    /// Output frame count may differ from the first input frame count.
    pub frame_count: bool,
    /// Output frame rate may differ from the first input frame rate.
    pub frame_rate: bool,
}

/// Filter media compatibility policy used during validation.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FilterCompatibility {
    /// Output and all inputs must have identical media properties.
    Preserve,
    /// Output may differ from the first input only on declared axes.
    AllowChanges(FilterChangeSet),
    /// Planner owns input/output media compatibility validation for this filter.
    Custom,
}

/// Format state known for graph clip.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ClipFormat {
    /// Clip always uses one concrete format.
    Fixed(FormatDescriptor),
    /// Clip can vary format by frame. Unsupported in Phase 1.
    Variable,
}

/// Resolution state known for graph clip.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ClipResolution {
    /// Clip always uses one concrete resolution.
    Fixed {
        /// Width in pixels.
        width: usize,
        /// Height in pixels.
        height: usize,
    },
    /// Clip can vary resolution by frame. Unsupported in Phase 1.
    Variable,
}

/// Frame-count state known for graph clip.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FrameCount {
    /// Known finite number of frames.
    Finite(usize),
    /// Unknown or infinite count. Unsupported in Phase 1.
    Unknown,
}

/// Frame-rate state known for graph clip.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FrameRate {
    /// Known constant frame rate.
    Cfr(Rational),
    /// Unknown or variable frame rate. Unsupported in Phase 1.
    Unknown,
}

/// Declared media properties for one clip.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ClipMedia {
    format: ClipFormat,
    resolution: ClipResolution,
    frame_count: FrameCount,
    frame_rate: FrameRate,
}

impl ClipMedia {
    /// Creates fixed Phase 1 media properties.
    #[must_use]
    pub const fn fixed(
        format: FormatDescriptor,
        width: usize,
        height: usize,
        frame_count: usize,
        frame_rate: Rational,
    ) -> Self {
        Self {
            format: ClipFormat::Fixed(format),
            resolution: ClipResolution::Fixed { width, height },
            frame_count: FrameCount::Finite(frame_count),
            frame_rate: FrameRate::Cfr(frame_rate),
        }
    }

    /// Creates media properties from explicit state values.
    #[must_use]
    pub const fn new(
        format: ClipFormat,
        resolution: ClipResolution,
        frame_count: FrameCount,
        frame_rate: FrameRate,
    ) -> Self {
        Self {
            format,
            resolution,
            frame_count,
            frame_rate,
        }
    }

    /// Returns format state.
    #[must_use]
    pub const fn format(&self) -> &ClipFormat {
        &self.format
    }

    /// Returns resolution state.
    #[must_use]
    pub const fn resolution(&self) -> &ClipResolution {
        &self.resolution
    }

    /// Returns frame-count state.
    #[must_use]
    pub const fn frame_count(&self) -> FrameCount {
        self.frame_count
    }

    /// Returns frame-rate state.
    #[must_use]
    pub const fn frame_rate(&self) -> FrameRate {
        self.frame_rate
    }
}

/// Returns true when final clip format can be written as Phase 1 Y4M.
#[must_use]
pub const fn is_y4m_compatible_format(format: &FormatDescriptor) -> bool {
    let integer_sample = matches!(format.sample_type(), SampleType::U8 | SampleType::U16);
    let supported_family = match format.family() {
        FormatFamily::Gray => true,
        FormatFamily::Yuv => matches!(
            format.subsampling(),
            Some(ChromaSubsampling::Cs420 | ChromaSubsampling::Cs422 | ChromaSubsampling::Cs444)
        ),
        FormatFamily::PlanarRgb => false,
    };

    integer_sample && supported_family
}

/// Stored graph node kind.
#[derive(Clone, Debug, PartialEq)]
pub enum NodeKind {
    /// Source node, indexed only when reachable.
    Source {
        /// Stable source diagnostic name.
        name: String,
        /// Source request captured during graph construction.
        request: crate::SourceRequest,
        /// Source scheduling capabilities.
        capabilities: crate::SourceCapabilities,
    },
    /// Filter node produced by applying filter to input clips.
    Filter {
        /// Stable filter diagnostic name.
        name: String,
        /// Input clip handles.
        inputs: Vec<Clip>,
        /// Generic options captured during graph construction for runtime executor assembly.
        options: FilterOptions,
        /// Compatibility policy for this filter call.
        compatibility: FilterCompatibility,
        /// Declared upstream frame dependency contract.
        dependencies: crate::DependencyPattern,
        /// Declared scheduler concurrency class.
        concurrency: crate::ConcurrencyClass,
    },
}

/// One immutable graph node.
#[derive(Clone, Debug, PartialEq)]
pub struct GraphNode {
    id: NodeId,
    kind: NodeKind,
    media: ClipMedia,
}

impl GraphNode {
    /// Returns node ID.
    #[must_use]
    pub const fn id(&self) -> NodeId {
        self.id
    }

    /// Returns node kind.
    #[must_use]
    pub const fn kind(&self) -> &NodeKind {
        &self.kind
    }

    /// Returns declared media properties.
    #[must_use]
    pub const fn media(&self) -> &ClipMedia {
        &self.media
    }

    /// Returns stored filter options when this node is a filter.
    #[must_use]
    pub const fn filter_options(&self) -> Option<&FilterOptions> {
        match &self.kind {
            NodeKind::Filter { options, .. } => Some(options),
            NodeKind::Source { .. } => None,
        }
    }
}

/// Immutable graph ready for validation and future rendering.
#[derive(Clone, Debug, PartialEq)]
pub struct Graph {
    nodes: Vec<GraphNode>,
    outputs: Vec<Clip>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum VisitState {
    Visiting,
    Done,
}

/// Validated graph plus reachability plan, ready for later render scheduling.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ValidatedGraph {
    plan: ValidationPlan,
}

impl ValidatedGraph {
    /// Returns graph validation plan.
    #[must_use]
    pub const fn plan(&self) -> &ValidationPlan {
        &self.plan
    }
}

impl Graph {
    /// Returns graph clone with one source node media description replaced.
    pub fn with_source_media(mut self, node_id: NodeId, media: ClipMedia) -> crate::Result<Self> {
        let node = self.nodes.get_mut(node_id.index()).ok_or_else(|| {
            crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.invalid_clip"),
                format!(
                    "source media update references missing node {}",
                    node_id.index()
                ),
            )
        })?;

        if !matches!(node.kind, NodeKind::Source { .. }) {
            return Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.invalid_clip"),
                format!("node {} is not a source", node_id.index()),
            ));
        }

        node.media = media;
        self.propagate_filter_media()?;
        Ok(self)
    }

    fn propagate_filter_media(&mut self) -> crate::Result<()> {
        for index in 0..self.nodes.len() {
            let node = semisafe_get(&self.nodes, index);
            let (inputs, compatibility, existing_media) = match &node.kind {
                NodeKind::Filter {
                    inputs,
                    compatibility,
                    ..
                } => (inputs.clone(), *compatibility, node.media.clone()),
                NodeKind::Source { .. } => continue,
            };

            let Some(first_input) = inputs.first().copied() else {
                return Err(crate::PixelFlowError::new(
                    crate::ErrorCategory::Graph,
                    crate::ErrorCode::new("graph.missing_filter_input"),
                    format!("filter node {index} has no inputs"),
                ));
            };
            let input_media = self.input_media(first_input)?.clone();

            let node = semisafe_get_mut(&mut self.nodes, index);
            node.media = match compatibility {
                FilterCompatibility::Preserve => input_media,
                FilterCompatibility::AllowChanges(changes) => ClipMedia::new(
                    if changes.format {
                        existing_media.format().clone()
                    } else {
                        input_media.format().clone()
                    },
                    if changes.resolution {
                        existing_media.resolution().clone()
                    } else {
                        input_media.resolution().clone()
                    },
                    if changes.frame_count {
                        existing_media.frame_count()
                    } else {
                        input_media.frame_count()
                    },
                    if changes.frame_rate {
                        existing_media.frame_rate()
                    } else {
                        input_media.frame_rate()
                    },
                ),
                FilterCompatibility::Custom => existing_media,
            };
        }

        Ok(())
    }

    /// Returns node by ID.
    #[must_use]
    pub fn node(&self, id: NodeId) -> Option<&GraphNode> {
        self.nodes.get(id.index()).filter(|node| node.id == id)
    }

    /// Returns all stored nodes.
    #[must_use]
    pub fn nodes(&self) -> &[GraphNode] {
        &self.nodes
    }

    /// Returns final output clips declared by graph construction.
    #[must_use]
    pub fn outputs(&self) -> &[Clip] {
        &self.outputs
    }

    /// Finds reachable nodes and sources from single final output.
    pub fn validation_plan(&self) -> crate::Result<ValidationPlan> {
        match self.outputs.as_slice() {
            [] => Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.missing_output"),
                "graph has no final output",
            )),
            [output] => {
                let mut reachable = Vec::new();
                self.collect_reachable(output.node_id(), &mut reachable)?;
                let reachable_sources = reachable
                    .iter()
                    .copied()
                    .filter(|id| {
                        self.node(*id)
                            .is_some_and(|node| matches!(node.kind(), NodeKind::Source { .. }))
                    })
                    .collect();

                Ok(ValidationPlan {
                    reachable_nodes: reachable,
                    reachable_sources,
                })
            }
            _ => Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.multiple_outputs"),
                "graph has multiple final outputs",
            )),
        }
    }

    /// Validates topology and media constraints for reachable output graph.
    pub fn validate(&self) -> crate::Result<ValidatedGraph> {
        let plan = self.validation_plan()?;
        self.validate_acyclic(&plan)?;
        self.validate_media(&plan)?;
        Ok(ValidatedGraph { plan })
    }

    fn collect_reachable(&self, id: NodeId, reachable: &mut Vec<NodeId>) -> crate::Result<()> {
        let mut states = vec![None; self.nodes.len()];
        self.collect_reachable_checked(id, reachable, &mut states)
    }

    fn collect_reachable_checked(
        &self,
        id: NodeId,
        reachable: &mut Vec<NodeId>,
        states: &mut [Option<VisitState>],
    ) -> crate::Result<()> {
        match states.get(id.index()).copied().flatten() {
            Some(VisitState::Visiting) => {
                return Err(crate::PixelFlowError::new(
                    crate::ErrorCategory::Graph,
                    crate::ErrorCode::new("graph.cycle"),
                    format!("graph contains a cycle involving node {}", id.index()),
                ));
            }
            Some(VisitState::Done) => return Ok(()),
            None => {}
        }

        let node = self.node(id).ok_or_else(|| {
            crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.invalid_clip"),
                format!("clip references missing node {}", id.index()),
            )
        })?;

        *semisafe_get_mut(states, id.index()) = Some(VisitState::Visiting);

        if let NodeKind::Filter { inputs, .. } = node.kind() {
            for input in inputs {
                self.collect_reachable_checked(input.node_id(), reachable, states)?;
            }
        }

        *semisafe_get_mut(states, id.index()) = Some(VisitState::Done);
        reachable.push(id);
        Ok(())
    }

    fn validate_acyclic(&self, plan: &ValidationPlan) -> crate::Result<()> {
        let mut states = vec![None; self.nodes.len()];
        for id in plan.reachable_nodes() {
            self.visit_for_cycle(*id, &mut states)?;
        }
        Ok(())
    }

    fn visit_for_cycle(&self, id: NodeId, states: &mut [Option<VisitState>]) -> crate::Result<()> {
        match states.get(id.index()).copied().flatten() {
            Some(VisitState::Visiting) => {
                return Err(crate::PixelFlowError::new(
                    crate::ErrorCategory::Graph,
                    crate::ErrorCode::new("graph.cycle"),
                    format!("graph contains a cycle involving node {}", id.index()),
                ));
            }
            Some(VisitState::Done) => return Ok(()),
            None => {}
        }

        let state = states.get_mut(id.index()).ok_or_else(|| {
            crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.invalid_clip"),
                format!("clip references missing node {}", id.index()),
            )
        })?;
        *state = Some(VisitState::Visiting);

        let node = self.node(id).ok_or_else(|| {
            crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.invalid_clip"),
                format!("clip references missing node {}", id.index()),
            )
        })?;

        if let NodeKind::Filter { inputs, .. } = node.kind() {
            for input in inputs {
                self.visit_for_cycle(input.node_id(), states)?;
            }
        }

        *semisafe_get_mut(states, id.index()) = Some(VisitState::Done);
        Ok(())
    }

    fn validate_media(&self, plan: &ValidationPlan) -> crate::Result<()> {
        for id in plan.reachable_nodes() {
            let node = self.node(*id).ok_or_else(|| {
                crate::PixelFlowError::new(
                    crate::ErrorCategory::Graph,
                    crate::ErrorCode::new("graph.invalid_clip"),
                    format!("clip references missing node {}", id.index()),
                )
            })?;
            validate_phase1_media(node.media(), *id)?;
            if let NodeKind::Filter {
                inputs,
                compatibility,
                ..
            } = node.kind()
            {
                self.validate_filter_compatibility(node, inputs, *compatibility)?;
            }
        }

        let output = semisafe_get(&self.outputs, 0).node_id();
        let output_node = self.node(output).ok_or_else(|| {
            crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.invalid_clip"),
                format!("output references missing node {}", output.index()),
            )
        })?;
        let ClipFormat::Fixed(format) = output_node.media().format() else {
            return Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.variable_format"),
                format!("node {} has variable format", output.index()),
            ));
        };
        if !is_y4m_compatible_format(format) {
            return Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.unsupported_output_format"),
                format!(
                    "final output format '{}' is not Y4M-compatible",
                    format.name()
                ),
            ));
        }

        Ok(())
    }

    fn validate_filter_compatibility(
        &self,
        node: &GraphNode,
        inputs: &[Clip],
        compatibility: FilterCompatibility,
    ) -> crate::Result<()> {
        let Some(first_input) = inputs.first() else {
            return Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.missing_filter_input"),
                format!("filter node {} has no inputs", node.id().index()),
            ));
        };
        let first_media = self.input_media(*first_input)?;

        if compatibility == FilterCompatibility::Custom {
            return Ok(());
        }

        for input in inputs.iter().skip(1) {
            let input_media = self.input_media(*input)?;
            require_equal_format(first_media, input_media, input.node_id())?;
            require_equal_resolution(first_media, input_media, input.node_id())?;
            require_equal_frame_count(first_media, input_media, input.node_id())?;
            require_equal_frame_rate(first_media, input_media, input.node_id())?;
        }

        match compatibility {
            FilterCompatibility::Preserve => {
                require_equal_format(first_media, node.media(), node.id())?;
                require_equal_resolution(first_media, node.media(), node.id())?;
                require_equal_frame_count(first_media, node.media(), node.id())?;
                require_equal_frame_rate(first_media, node.media(), node.id())?;
            }
            FilterCompatibility::AllowChanges(changes) => {
                if !changes.format {
                    require_equal_format(first_media, node.media(), node.id())?;
                }
                if !changes.resolution {
                    require_equal_resolution(first_media, node.media(), node.id())?;
                }
                if !changes.frame_count {
                    require_equal_frame_count(first_media, node.media(), node.id())?;
                }
                if !changes.frame_rate {
                    require_equal_frame_rate(first_media, node.media(), node.id())?;
                }
            }
            FilterCompatibility::Custom => unreachable!("custom compatibility returned early"),
        }

        Ok(())
    }

    fn input_media(&self, clip: Clip) -> crate::Result<&ClipMedia> {
        self.node(clip.node_id())
            .map(GraphNode::media)
            .ok_or_else(|| {
                crate::PixelFlowError::new(
                    crate::ErrorCategory::Graph,
                    crate::ErrorCode::new("graph.invalid_clip"),
                    format!("clip references missing node {}", clip.node_id().index()),
                )
            })
    }
}

fn validate_phase1_media(media: &ClipMedia, id: NodeId) -> crate::Result<()> {
    match media.format() {
        ClipFormat::Fixed(_) => {}
        ClipFormat::Variable => {
            return Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.variable_format"),
                format!("node {} has variable format", id.index()),
            ));
        }
    }

    match media.resolution() {
        ClipResolution::Fixed { width, height } if *width > 0 && *height > 0 => {}
        ClipResolution::Fixed { .. } => {
            return Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.invalid_resolution"),
                format!("node {} has zero resolution dimension", id.index()),
            ));
        }
        ClipResolution::Variable => {
            return Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.variable_resolution"),
                format!("node {} has variable resolution", id.index()),
            ));
        }
    }

    match media.frame_count() {
        FrameCount::Finite(_) => {}
        FrameCount::Unknown => {
            return Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.unknown_frame_count"),
                format!("node {} has unknown frame count", id.index()),
            ));
        }
    }

    match media.frame_rate() {
        FrameRate::Cfr(rate) if rate.denominator > 0 && rate.numerator > 0 => {}
        FrameRate::Cfr(_) => {
            return Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.invalid_frame_rate"),
                format!("node {} has invalid frame rate", id.index()),
            ));
        }
        FrameRate::Unknown => {
            return Err(crate::PixelFlowError::new(
                crate::ErrorCategory::Graph,
                crate::ErrorCode::new("graph.unknown_frame_rate"),
                format!("node {} has unknown frame rate", id.index()),
            ));
        }
    }

    Ok(())
}

fn require_equal_format(expected: &ClipMedia, actual: &ClipMedia, id: NodeId) -> crate::Result<()> {
    if expected.format() == actual.format() {
        return Ok(());
    }
    Err(crate::PixelFlowError::new(
        crate::ErrorCategory::Graph,
        crate::ErrorCode::new("graph.incompatible_format"),
        format!(
            "node {} changes format without explicit conversion policy",
            id.index()
        ),
    ))
}

fn require_equal_resolution(
    expected: &ClipMedia,
    actual: &ClipMedia,
    id: NodeId,
) -> crate::Result<()> {
    if expected.resolution() == actual.resolution() {
        return Ok(());
    }
    Err(crate::PixelFlowError::new(
        crate::ErrorCategory::Graph,
        crate::ErrorCode::new("graph.incompatible_resolution"),
        format!(
            "node {} changes resolution without explicit resize policy",
            id.index()
        ),
    ))
}

fn require_equal_frame_count(
    expected: &ClipMedia,
    actual: &ClipMedia,
    id: NodeId,
) -> crate::Result<()> {
    if expected.frame_count() == actual.frame_count() {
        return Ok(());
    }
    Err(crate::PixelFlowError::new(
        crate::ErrorCategory::Graph,
        crate::ErrorCode::new("graph.incompatible_frame_count"),
        format!(
            "node {} changes frame count without explicit policy",
            id.index()
        ),
    ))
}

fn require_equal_frame_rate(
    expected: &ClipMedia,
    actual: &ClipMedia,
    id: NodeId,
) -> crate::Result<()> {
    if expected.frame_rate() == actual.frame_rate() {
        return Ok(());
    }
    Err(crate::PixelFlowError::new(
        crate::ErrorCategory::Graph,
        crate::ErrorCode::new("graph.incompatible_frame_rate"),
        format!(
            "node {} changes frame rate without explicit policy",
            id.index()
        ),
    ))
}

/// Reachability result used to index only reachable sources before render validation.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ValidationPlan {
    reachable_nodes: Vec<NodeId>,
    reachable_sources: Vec<NodeId>,
}

impl ValidationPlan {
    /// Returns reachable nodes in dependency-first order.
    #[must_use]
    pub fn reachable_nodes(&self) -> &[NodeId] {
        &self.reachable_nodes
    }

    /// Returns reachable source nodes in dependency-first order.
    #[must_use]
    pub fn reachable_sources(&self) -> &[NodeId] {
        &self.reachable_sources
    }
}

/// Mutable construction API. `build` freezes nodes into immutable graph.
#[derive(Clone, Debug, Default, PartialEq)]
pub struct GraphBuilder {
    nodes: Vec<GraphNode>,
    outputs: Vec<Clip>,
}

impl GraphBuilder {
    /// Creates empty graph builder.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            nodes: Vec::new(),
            outputs: Vec::new(),
        }
    }

    /// Adds source node and returns immutable clip handle.
    pub fn source(&mut self, name: impl Into<String>, media: ClipMedia) -> Clip {
        let name = name.into();
        self.source_with_request(crate::SourceRequest::new(name), media)
    }

    /// Adds source node with explicit request metadata.
    pub fn source_with_request(&mut self, request: crate::SourceRequest, media: ClipMedia) -> Clip {
        self.source_with_request_and_capabilities(
            request,
            media,
            crate::SourceCapabilities::random_access(),
        )
    }

    /// Adds source node with explicit scheduler capabilities.
    pub fn source_with_capabilities(
        &mut self,
        name: impl Into<String>,
        media: ClipMedia,
        capabilities: crate::SourceCapabilities,
    ) -> Clip {
        self.source_with_request_and_capabilities(
            crate::SourceRequest::new(name.into()),
            media,
            capabilities,
        )
    }

    /// Adds source node with explicit request metadata and scheduler capabilities.
    pub fn source_with_request_and_capabilities(
        &mut self,
        request: crate::SourceRequest,
        media: ClipMedia,
        capabilities: crate::SourceCapabilities,
    ) -> Clip {
        let id = NodeId::new(self.nodes.len());
        let clip = Clip::new(id);
        self.nodes.push(GraphNode {
            id,
            kind: NodeKind::Source {
                name: request.path().to_owned(),
                request,
                capabilities,
            },
            media,
        });
        clip
    }

    /// Adds filter node and returns new immutable clip handle.
    pub fn filter(
        &mut self,
        name: impl Into<String>,
        inputs: &[Clip],
        media: ClipMedia,
        compatibility: FilterCompatibility,
    ) -> crate::Result<Clip> {
        self.filter_with_schedule(
            name,
            inputs,
            media,
            compatibility,
            crate::DependencyPattern::same_frame(),
            crate::ConcurrencyClass::Stateless,
        )
    }

    /// Adds filter node with explicit scheduling contract.
    pub fn filter_with_schedule(
        &mut self,
        name: impl Into<String>,
        inputs: &[Clip],
        media: ClipMedia,
        compatibility: FilterCompatibility,
        dependencies: crate::DependencyPattern,
        concurrency: crate::ConcurrencyClass,
    ) -> crate::Result<Clip> {
        self.filter_with_schedule_and_options(
            name,
            inputs,
            media,
            compatibility,
            dependencies,
            concurrency,
            FilterOptions::new(),
        )
    }

    /// Adds filter node with explicit scheduling contract and stored options.
    pub fn filter_with_schedule_and_options(
        &mut self,
        name: impl Into<String>,
        inputs: &[Clip],
        media: ClipMedia,
        compatibility: FilterCompatibility,
        dependencies: crate::DependencyPattern,
        concurrency: crate::ConcurrencyClass,
        options: FilterOptions,
    ) -> crate::Result<Clip> {
        for input in inputs {
            if self.nodes.get(input.node_id().index()).is_none() {
                return Err(crate::PixelFlowError::new(
                    crate::ErrorCategory::Graph,
                    crate::ErrorCode::new("graph.invalid_clip"),
                    format!(
                        "filter input references missing node {}",
                        input.node_id().index()
                    ),
                ));
            }
        }

        let id = NodeId::new(self.nodes.len());
        let clip = Clip::new(id);
        self.nodes.push(GraphNode {
            id,
            kind: NodeKind::Filter {
                name: name.into(),
                inputs: inputs.to_vec(),
                options,
                compatibility,
                dependencies,
                concurrency,
            },
            media,
        });
        Ok(clip)
    }

    /// Replaces final output list with one clip.
    pub fn set_output(&mut self, output: Clip) {
        self.outputs.clear();
        self.outputs.push(output);
    }

    /// Adds final output clip. Phase 1 validation rejects multiple outputs.
    pub fn add_output(&mut self, output: Clip) {
        self.outputs.push(output);
    }

    /// Freezes builder into immutable graph.
    #[must_use]
    pub fn build(self) -> Graph {
        Graph {
            nodes: self.nodes,
            outputs: self.outputs,
        }
    }
}

#[cfg(test)]
impl Graph {
    fn from_parts_for_tests(nodes: Vec<GraphNode>, outputs: Vec<Clip>) -> Self {
        Self { nodes, outputs }
    }
}

#[cfg(test)]
impl GraphNode {
    fn filter_for_tests(
        id: NodeId,
        name: impl Into<String>,
        inputs: &[Clip],
        media: ClipMedia,
        compatibility: FilterCompatibility,
    ) -> Self {
        Self {
            id,
            kind: NodeKind::Filter {
                name: name.into(),
                inputs: inputs.to_vec(),
                options: crate::FilterOptions::new(),
                compatibility,
                dependencies: crate::DependencyPattern::same_frame(),
                concurrency: crate::ConcurrencyClass::Stateless,
            },
            media,
        }
    }
}

#[cfg(test)]
mod tests {
    #![expect(clippy::panic, reason = "allow in tests")]
    #![expect(clippy::unwrap_used, reason = "allow in tests")]

    use crate::{ChromaSubsampling, FormatFamily, Rational, SampleType, resolve_format_alias};

    use super::{
        ClipFormat, ClipMedia, ClipResolution, FilterChangeSet, FilterCompatibility, FrameCount,
        FrameRate, GraphBuilder, NodeId, NodeKind, is_y4m_compatible_format,
    };

    fn fixed_media(alias: &str) -> ClipMedia {
        ClipMedia::fixed(
            resolve_format_alias(alias).expect("format alias should resolve"),
            1920,
            1080,
            24,
            Rational {
                numerator: 24_000,
                denominator: 1_001,
            },
        )
    }

    #[test]
    fn graph_builder_source_uses_phase1_source_capability_defaults() {
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", fixed_media("yuv420p10"));
        let graph = builder.build();

        let node = graph.node(source.node_id()).expect("source exists");
        let NodeKind::Source { capabilities, .. } = node.kind() else {
            panic!("expected source node");
        };

        assert!(capabilities.supports_random_access());
        assert!(capabilities.indexing_required());
        assert!(capabilities.known_frame_count());
        assert_eq!(capabilities.concurrency_limit(), Some(1));
    }

    #[test]
    fn graph_builder_source_stores_request_path_and_options() {
        let mut builder = GraphBuilder::new();
        let request = crate::SourceRequest::new("relative/input.mkv")
            .with_option(
                "fps",
                crate::SourceOptionValue::Rational(Rational {
                    numerator: 30_000,
                    denominator: 1_001,
                }),
            )
            .with_option(
                "vfr",
                crate::SourceOptionValue::String("normalize".to_owned()),
            );
        let clip = builder.source_with_request(request.clone(), fixed_media("yuv420p8"));
        let graph = builder.build();

        let node = graph.node(clip.node_id()).expect("source exists");
        let NodeKind::Source {
            request: stored, ..
        } = node.kind()
        else {
            panic!("expected source node");
        };

        assert_eq!(stored, &request);
    }

    #[test]
    fn graph_with_source_media_replaces_only_source_media() {
        let mut builder = GraphBuilder::new();
        let source = builder.source(
            "input.mkv",
            ClipMedia::new(
                ClipFormat::Fixed(resolve_format_alias("yuv420p8").unwrap()),
                ClipResolution::Fixed {
                    width: 1,
                    height: 1,
                },
                FrameCount::Unknown,
                FrameRate::Unknown,
            ),
        );
        let graph = builder.build();
        let media = fixed_media("yuv420p10");

        let graph = graph
            .with_source_media(source.node_id(), media.clone())
            .expect("source media updates");

        assert_eq!(graph.node(source.node_id()).unwrap().media(), &media);
    }

    #[test]
    fn graph_builder_filter_records_dependency_and_concurrency_defaults() {
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", fixed_media("yuv420p10"));
        let filtered = builder
            .filter(
                "identity",
                &[source],
                fixed_media("yuv420p10"),
                FilterCompatibility::Preserve,
            )
            .expect("filter should be added");
        let graph = builder.build();

        let node = graph.node(filtered.node_id()).expect("filter exists");
        let NodeKind::Filter {
            dependencies,
            concurrency,
            ..
        } = node.kind()
        else {
            panic!("expected filter node");
        };

        assert_eq!(dependencies, &crate::DependencyPattern::same_frame());
        assert_eq!(*concurrency, crate::ConcurrencyClass::Stateless);
    }

    #[test]
    fn graph_builder_filter_with_schedule_records_explicit_contracts() {
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", fixed_media("yuv420p10"));
        let filtered = builder
            .filter_with_schedule(
                "temporal",
                &[source],
                fixed_media("yuv420p10"),
                FilterCompatibility::Preserve,
                crate::DependencyPattern::window(2, 0),
                crate::ConcurrencyClass::OrderedStateful,
            )
            .expect("filter should be added");
        let graph = builder.build();

        let node = graph.node(filtered.node_id()).expect("filter exists");
        let NodeKind::Filter {
            dependencies,
            concurrency,
            ..
        } = node.kind()
        else {
            panic!("expected filter node");
        };

        assert_eq!(dependencies, &crate::DependencyPattern::window(2, 0));
        assert_eq!(*concurrency, crate::ConcurrencyClass::OrderedStateful);
    }

    #[test]
    fn graph_builder_filter_with_options_preserves_options() {
        let media = fixed_media("yuv420p8");
        let mut builder = GraphBuilder::new();
        let source = builder.source("input", media.clone());
        let mut options = crate::FilterOptions::new();
        options.insert("width".to_owned(), crate::FilterOptionValue::Int(320));
        options.insert("height".to_owned(), crate::FilterOptionValue::Int(180));

        let filtered = builder
            .filter_with_schedule_and_options(
                "resize",
                &[source],
                media,
                FilterCompatibility::Preserve,
                crate::DependencyPattern::same_frame(),
                crate::ConcurrencyClass::Stateless,
                options.clone(),
            )
            .expect("filter should build");
        builder.set_output(filtered);
        let graph = builder.build();

        let NodeKind::Filter {
            options: stored, ..
        } = graph
            .node(filtered.node_id())
            .expect("filter exists")
            .kind()
        else {
            panic!("node should be filter");
        };
        assert_eq!(stored, &options);
    }

    #[test]
    fn custom_filter_compatibility_allows_planner_validated_heterogeneous_inputs() {
        let gray8 = resolve_format_alias("gray8").expect("format should resolve");
        let yuv420 = resolve_format_alias("yuv420p8").expect("format should resolve");
        let y_media = ClipMedia::fixed(
            gray8.clone(),
            4,
            4,
            1,
            Rational {
                numerator: 24,
                denominator: 1,
            },
        );
        let u_media = ClipMedia::fixed(
            gray8.clone(),
            2,
            2,
            1,
            Rational {
                numerator: 24,
                denominator: 1,
            },
        );
        let v_media = ClipMedia::fixed(
            gray8,
            2,
            2,
            1,
            Rational {
                numerator: 24,
                denominator: 1,
            },
        );
        let output_media = ClipMedia::fixed(
            yuv420,
            4,
            4,
            1,
            Rational {
                numerator: 24,
                denominator: 1,
            },
        );

        let mut builder = GraphBuilder::new();
        let y = builder.source("y", y_media);
        let u = builder.source("u", u_media);
        let v = builder.source("v", v_media);
        let merged = builder
            .filter(
                "merge_planes",
                &[y, u, v],
                output_media,
                FilterCompatibility::Custom,
            )
            .expect("custom filter node should build");
        builder.set_output(merged);
        let graph = builder.build();

        graph
            .validate()
            .expect("custom compatibility should skip generic input equality");
    }

    #[test]
    fn custom_filter_compatibility_still_rejects_missing_inputs() {
        let yuv420 = resolve_format_alias("yuv420p8").expect("format should resolve");
        let output_media = ClipMedia::fixed(
            yuv420,
            4,
            4,
            1,
            Rational {
                numerator: 24,
                denominator: 1,
            },
        );
        let mut builder = GraphBuilder::new();
        let merged = builder
            .filter(
                "merge_planes",
                &[],
                output_media,
                FilterCompatibility::Custom,
            )
            .expect("graph builder records empty custom input list");
        builder.set_output(merged);

        let error = builder
            .build()
            .validate()
            .expect_err("empty custom inputs should fail");
        assert_eq!(error.category(), crate::ErrorCategory::Graph);
        assert_eq!(
            error.code(),
            crate::ErrorCode::new("graph.missing_filter_input")
        );
    }

    #[test]
    fn y4m_compatibility_accepts_phase1_integer_yuv_and_gray() {
        let yuv420 = resolve_format_alias("yuv420p10").expect("format should resolve");
        let yuv422 = resolve_format_alias("yuv422p16").expect("format should resolve");
        let yuv444 = resolve_format_alias("yuv444p8").expect("format should resolve");
        let gray = resolve_format_alias("gray12").expect("format should resolve");

        assert!(is_y4m_compatible_format(&yuv420));
        assert!(is_y4m_compatible_format(&yuv422));
        assert!(is_y4m_compatible_format(&yuv444));
        assert!(is_y4m_compatible_format(&gray));
    }

    #[test]
    fn y4m_compatibility_rejects_rgb_and_float_outputs() {
        let rgb = resolve_format_alias("rgbp10").expect("format should resolve");
        let gray_float = resolve_format_alias("grayf32").expect("format should resolve");
        let yuv_float = resolve_format_alias("yuv420pf32").expect("format should resolve");

        assert!(!is_y4m_compatible_format(&rgb));
        assert!(!is_y4m_compatible_format(&gray_float));
        assert!(!is_y4m_compatible_format(&yuv_float));
    }

    #[test]
    fn fixed_media_records_phase1_constraints() {
        let media = fixed_media("yuv420p10");

        assert!(
            matches!(media.format(), ClipFormat::Fixed(format) if format.family() == FormatFamily::Yuv)
        );
        assert!(
            matches!(media.format(), ClipFormat::Fixed(format) if format.subsampling() == Some(ChromaSubsampling::Cs420))
        );
        assert!(
            matches!(media.format(), ClipFormat::Fixed(format) if format.sample_type() == SampleType::U16)
        );
        assert_eq!(
            media.resolution(),
            &ClipResolution::Fixed {
                width: 1920,
                height: 1080,
            }
        );
        assert_eq!(media.frame_count(), FrameCount::Finite(24));
        assert_eq!(
            media.frame_rate(),
            FrameRate::Cfr(Rational {
                numerator: 24_000,
                denominator: 1_001,
            })
        );
    }

    #[test]
    fn filter_call_creates_new_clip_without_mutating_input_clip() {
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", fixed_media("yuv420p10"));
        let filtered = builder
            .filter(
                "identity",
                &[source],
                fixed_media("yuv420p10"),
                FilterCompatibility::Preserve,
            )
            .expect("filter should be added");

        assert_ne!(source.node_id(), filtered.node_id());
        assert_eq!(source.node_id().index(), 0);
        assert_eq!(filtered.node_id().index(), 1);
    }

    #[test]
    fn missing_final_output_fails_with_structured_diagnostic() {
        let mut builder = GraphBuilder::new();
        builder.source("source", fixed_media("yuv420p10"));
        let graph = builder.build();

        let error = graph
            .validation_plan()
            .expect_err("missing output should fail");

        assert_eq!(error.category(), crate::ErrorCategory::Graph);
        assert_eq!(error.code(), crate::ErrorCode::new("graph.missing_output"));
    }

    #[test]
    fn multiple_final_outputs_fail_phase1_validation() {
        let mut builder = GraphBuilder::new();
        let first = builder.source("first", fixed_media("yuv420p10"));
        let second = builder.source("second", fixed_media("yuv420p10"));
        builder.add_output(first);
        builder.add_output(second);
        let graph = builder.build();

        let error = graph
            .validation_plan()
            .expect_err("multiple outputs should fail");

        assert_eq!(error.category(), crate::ErrorCategory::Graph);
        assert_eq!(
            error.code(),
            crate::ErrorCode::new("graph.multiple_outputs")
        );
        assert_eq!(error.message(), "graph has multiple final outputs");
    }

    #[test]
    fn unreachable_source_and_filter_nodes_are_skipped_by_validation_plan() {
        let mut builder = GraphBuilder::new();
        let unused_source = builder.source("unused", fixed_media("yuv420p10"));
        let used_source = builder.source("used", fixed_media("yuv420p10"));
        let used_filter = builder
            .filter(
                "identity",
                &[used_source],
                fixed_media("yuv420p10"),
                FilterCompatibility::Preserve,
            )
            .expect("reachable filter should be added");
        let unused_filter = builder
            .filter(
                "identity",
                &[unused_source],
                fixed_media("yuv420p10"),
                FilterCompatibility::Preserve,
            )
            .expect("unreachable filter should be added");
        builder.set_output(used_filter);
        let graph = builder.build();

        let plan = graph
            .validation_plan()
            .expect("graph should have validation plan");

        assert_eq!(
            plan.reachable_nodes(),
            &[used_source.node_id(), used_filter.node_id()]
        );
        assert_eq!(plan.reachable_sources(), &[used_source.node_id()]);
        assert!(!plan.reachable_nodes().contains(&unused_source.node_id()));
        assert!(!plan.reachable_nodes().contains(&unused_filter.node_id()));
        assert!(matches!(
            graph.node(used_source.node_id()).unwrap().kind(),
            NodeKind::Source { .. }
        ));
    }

    #[test]
    fn cyclic_graph_fails_before_render_validation() {
        let media = fixed_media("yuv420p10");
        let graph = super::Graph::from_parts_for_tests(
            vec![
                super::GraphNode::filter_for_tests(
                    NodeId::new(0),
                    "a",
                    &[super::Clip::new(NodeId::new(1))],
                    media.clone(),
                    FilterCompatibility::Preserve,
                ),
                super::GraphNode::filter_for_tests(
                    NodeId::new(1),
                    "b",
                    &[super::Clip::new(NodeId::new(0))],
                    media,
                    FilterCompatibility::Preserve,
                ),
            ],
            vec![super::Clip::new(NodeId::new(0))],
        );

        let error = graph.validate().expect_err("cycle should fail validation");

        assert_eq!(error.category(), crate::ErrorCategory::Graph);
        assert_eq!(error.code(), crate::ErrorCode::new("graph.cycle"));
    }

    fn invalid_media(
        format: ClipFormat,
        resolution: ClipResolution,
        frame_count: FrameCount,
        frame_rate: FrameRate,
    ) -> ClipMedia {
        ClipMedia::new(format, resolution, frame_count, frame_rate)
    }

    #[test]
    fn variable_format_resolution_unknown_count_and_unknown_rate_fail_before_render() {
        let cases = [
            (
                invalid_media(
                    ClipFormat::Variable,
                    ClipResolution::Fixed {
                        width: 1920,
                        height: 1080,
                    },
                    FrameCount::Finite(24),
                    FrameRate::Cfr(Rational {
                        numerator: 24,
                        denominator: 1,
                    }),
                ),
                crate::ErrorCode::new("graph.variable_format"),
            ),
            (
                invalid_media(
                    ClipFormat::Fixed(resolve_format_alias("yuv420p10").unwrap()),
                    ClipResolution::Variable,
                    FrameCount::Finite(24),
                    FrameRate::Cfr(Rational {
                        numerator: 24,
                        denominator: 1,
                    }),
                ),
                crate::ErrorCode::new("graph.variable_resolution"),
            ),
            (
                invalid_media(
                    ClipFormat::Fixed(resolve_format_alias("yuv420p10").unwrap()),
                    ClipResolution::Fixed {
                        width: 1920,
                        height: 1080,
                    },
                    FrameCount::Unknown,
                    FrameRate::Cfr(Rational {
                        numerator: 24,
                        denominator: 1,
                    }),
                ),
                crate::ErrorCode::new("graph.unknown_frame_count"),
            ),
            (
                invalid_media(
                    ClipFormat::Fixed(resolve_format_alias("yuv420p10").unwrap()),
                    ClipResolution::Fixed {
                        width: 1920,
                        height: 1080,
                    },
                    FrameCount::Finite(24),
                    FrameRate::Unknown,
                ),
                crate::ErrorCode::new("graph.unknown_frame_rate"),
            ),
        ];

        for (media, expected_code) in cases {
            let mut builder = GraphBuilder::new();
            let source = builder.source("source", media);
            builder.set_output(source);
            let graph = builder.build();

            let error = graph.validate().expect_err("invalid media should fail");

            assert_eq!(error.category(), crate::ErrorCategory::Graph);
            assert_eq!(error.code(), expected_code);
        }
    }

    #[test]
    fn unsupported_final_y4m_format_fails_before_render() {
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", fixed_media("rgbp10"));
        builder.set_output(source);
        let graph = builder.build();

        let error = graph
            .validate()
            .expect_err("rgb output should fail y4m validation");

        assert_eq!(error.category(), crate::ErrorCategory::Graph);
        assert_eq!(
            error.code(),
            crate::ErrorCode::new("graph.unsupported_output_format")
        );
    }

    #[test]
    fn preserve_policy_rejects_implicit_format_conversion() {
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", fixed_media("rgbp10"));
        let converted = builder
            .filter(
                "identity",
                &[source],
                fixed_media("yuv420p10"),
                FilterCompatibility::Preserve,
            )
            .expect("filter should be added");
        builder.set_output(converted);
        let graph = builder.build();

        let error = graph
            .validate()
            .expect_err("implicit conversion should fail");

        assert_eq!(error.category(), crate::ErrorCategory::Graph);
        assert_eq!(
            error.code(),
            crate::ErrorCode::new("graph.incompatible_format")
        );
    }

    #[test]
    fn explicit_conversion_filter_can_change_format_without_implicit_insertion() {
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", fixed_media("rgbp10"));
        let converted = builder
            .filter(
                "convert_format",
                &[source],
                fixed_media("yuv420p10"),
                FilterCompatibility::AllowChanges(FilterChangeSet {
                    format: true,
                    resolution: false,
                    frame_count: false,
                    frame_rate: false,
                }),
            )
            .expect("filter should be added");
        builder.set_output(converted);
        let graph = builder.build();

        let validated = graph.validate().expect("explicit conversion should pass");

        assert_eq!(
            validated.plan().reachable_nodes(),
            &[source.node_id(), converted.node_id()]
        );
    }

    #[test]
    fn allow_changes_policy_accepts_format_and_resolution_change() {
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", fixed_media("yuv420p10"));
        let changed_media = ClipMedia::fixed(
            resolve_format_alias("yuv444p8").unwrap(),
            1280,
            720,
            24,
            Rational {
                numerator: 24_000,
                denominator: 1_001,
            },
        );
        let filtered = builder
            .filter(
                "scale_convert",
                &[source],
                changed_media,
                FilterCompatibility::AllowChanges(FilterChangeSet {
                    format: true,
                    resolution: true,
                    frame_count: false,
                    frame_rate: false,
                }),
            )
            .expect("filter should be added");
        builder.set_output(filtered);

        builder
            .build()
            .validate()
            .expect("declared format and resolution change should pass");
    }

    #[test]
    fn allow_changes_policy_rejects_undeclared_frame_count_change() {
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", fixed_media("yuv420p10"));
        let changed_media = ClipMedia::fixed(
            resolve_format_alias("yuv444p8").unwrap(),
            1280,
            720,
            12,
            Rational {
                numerator: 24_000,
                denominator: 1_001,
            },
        );
        let filtered = builder
            .filter(
                "scale_convert",
                &[source],
                changed_media,
                FilterCompatibility::AllowChanges(FilterChangeSet {
                    format: true,
                    resolution: true,
                    frame_count: false,
                    frame_rate: false,
                }),
            )
            .expect("filter should be added");
        builder.set_output(filtered);

        let error = builder
            .build()
            .validate()
            .expect_err("undeclared frame-count change should fail");

        assert_eq!(error.category(), crate::ErrorCategory::Graph);
        assert_eq!(
            error.code(),
            crate::ErrorCode::new("graph.incompatible_frame_count")
        );
    }

    #[test]
    fn allow_changes_policy_accepts_frame_rate_change_when_declared() {
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", fixed_media("yuv420p10"));
        let changed_media = ClipMedia::fixed(
            resolve_format_alias("yuv420p10").unwrap(),
            1920,
            1080,
            24,
            Rational {
                numerator: 30,
                denominator: 1,
            },
        );
        let filtered = builder
            .filter(
                "retime",
                &[source],
                changed_media,
                FilterCompatibility::AllowChanges(FilterChangeSet {
                    format: false,
                    resolution: false,
                    frame_count: false,
                    frame_rate: true,
                }),
            )
            .expect("filter should be added");
        builder.set_output(filtered);

        builder
            .build()
            .validate()
            .expect("declared frame-rate change should pass");
    }

    #[test]
    fn preserve_policy_rejects_frame_count_and_rate_mismatch() {
        let source_media = fixed_media("yuv420p10");
        let count_changed = ClipMedia::fixed(
            resolve_format_alias("yuv420p10").unwrap(),
            1920,
            1080,
            12,
            Rational {
                numerator: 24_000,
                denominator: 1_001,
            },
        );
        let rate_changed = ClipMedia::fixed(
            resolve_format_alias("yuv420p10").unwrap(),
            1920,
            1080,
            24,
            Rational {
                numerator: 30,
                denominator: 1,
            },
        );

        for (media, code) in [
            (
                count_changed,
                crate::ErrorCode::new("graph.incompatible_frame_count"),
            ),
            (
                rate_changed,
                crate::ErrorCode::new("graph.incompatible_frame_rate"),
            ),
        ] {
            let mut builder = GraphBuilder::new();
            let source = builder.source("source", source_media.clone());
            let filtered = builder
                .filter("identity", &[source], media, FilterCompatibility::Preserve)
                .expect("filter should be added");
            builder.set_output(filtered);
            let graph = builder.build();

            let error = graph.validate().expect_err("preserve mismatch should fail");

            assert_eq!(error.category(), crate::ErrorCategory::Graph);
            assert_eq!(error.code(), code);
        }
    }
}