pixelflow-filters 0.1.0

//! Plane split and merge filter validation and executors.

use pixelflow_core::{
    Clip, ClipFormat, ClipMedia, ClipResolution, ErrorCategory, ErrorCode, FilterChangeSet,
    FilterCompatibility, FilterPlan, FilterPlanRequest, FormatDescriptor, FormatFamily, Frame,
    FrameExecutor, FrameRequest, GraphBuilder, PixelFlowError, Result,
};

use crate::{
    OfficialFilterContract, SupportedFormats,
    contracts::{ALL_PLANAR_FAMILIES, ALL_SAMPLE_TYPES},
    options::{required_i64, required_string, single_input_media, single_input_slice},
};

/// Official single-plane split filter script name.
pub const FILTER_SPLIT_PLANE: &str = "split_plane";

/// Contract for `split_plane`.
pub const SPLIT_PLANE_CONTRACT: OfficialFilterContract = OfficialFilterContract::new(
    FILTER_SPLIT_PLANE,
    SupportedFormats::new(ALL_PLANAR_FAMILIES, ALL_SAMPLE_TYPES),
    FilterCompatibility::AllowChanges(FilterChangeSet {
        format: true,
        resolution: true,
        frame_count: false,
        frame_rate: false,
    }),
);

/// Official plane merge filter script name.
pub const FILTER_MERGE_PLANES: &str = "merge_planes";

/// Contract for `merge_planes`.
pub const MERGE_PLANES_CONTRACT: OfficialFilterContract = OfficialFilterContract::new(
    FILTER_MERGE_PLANES,
    SupportedFormats::new(ALL_PLANAR_FAMILIES, ALL_SAMPLE_TYPES),
    FilterCompatibility::Custom,
);

/// User-supplied zero-based split plane index before validation.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct SplitPlaneOptions {
    plane: i64,
}

impl SplitPlaneOptions {
    /// Creates split options from script or embedding plane index.
    #[must_use]
    pub const fn new(plane: i64) -> Self {
        Self { plane }
    }

    /// Returns requested zero-based plane index before validation.
    #[must_use]
    pub const fn plane(self) -> i64 {
        self.plane
    }
}

/// Split plane selection validated against input media.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ValidatedSplitPlane {
    plane_index: usize,
    output_format: FormatDescriptor,
}

impl ValidatedSplitPlane {
    /// Returns selected source plane index.
    #[must_use]
    pub const fn plane_index(&self) -> usize {
        self.plane_index
    }

    /// Returns output Gray format.
    #[must_use]
    pub const fn output_format(&self) -> &FormatDescriptor {
        &self.output_format
    }
}

/// User-supplied target format for plane merge.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct MergePlanesOptions {
    format: String,
}

impl MergePlanesOptions {
    /// Creates merge options from target format alias.
    #[must_use]
    pub fn new(format: impl Into<String>) -> Self {
        Self {
            format: format.into(),
        }
    }

    /// Returns requested target format alias.
    #[must_use]
    pub fn format(&self) -> &str {
        &self.format
    }
}

/// Merge settings validated against all input media.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ValidatedMergePlanes {
    output_format: FormatDescriptor,
    output_width: usize,
    output_height: usize,
}

impl ValidatedMergePlanes {
    /// Returns target output format.
    #[must_use]
    pub const fn output_format(&self) -> &FormatDescriptor {
        &self.output_format
    }

    /// Returns target output width.
    #[must_use]
    pub const fn output_width(&self) -> usize {
        self.output_width
    }

    /// Returns target output height.
    #[must_use]
    pub const fn output_height(&self) -> usize {
        self.output_height
    }
}

/// Validates split-plane options and derives output media.
pub fn split_plane_output_media(
    input: &ClipMedia,
    options: SplitPlaneOptions,
) -> Result<(ValidatedSplitPlane, ClipMedia)> {
    let format = SPLIT_PLANE_CONTRACT.validate_input_media(input)?;
    let (input_width, input_height) = fixed_resolution(input, FILTER_SPLIT_PLANE)?;
    let plane_index = validate_plane_index(options.plane())?;
    let descriptor = format.planes().get(plane_index).ok_or_else(|| {
        plane_error(
            "filter.unsupported_plane",
            format!(
                "filter '{FILTER_SPLIT_PLANE}' format '{}' has no plane index {plane_index}",
                format.name()
            ),
        )
    })?;
    let width = input_width.div_ceil(descriptor.width_divisor);
    let height = input_height.div_ceil(descriptor.height_divisor);
    let output_format = gray_format_for(format)?;
    let output_media = ClipMedia::new(
        ClipFormat::Fixed(output_format.clone()),
        ClipResolution::Fixed { width, height },
        input.frame_count(),
        input.frame_rate(),
    );

    Ok((
        ValidatedSplitPlane {
            plane_index,
            output_format,
        },
        output_media,
    ))
}

/// Adds validated split-plane filter node and returns matching render executor.
pub fn add_split_plane_filter(
    builder: &mut GraphBuilder,
    input: Clip,
    input_media: &ClipMedia,
    options: SplitPlaneOptions,
) -> Result<(Clip, SplitPlaneExecutor)> {
    let (split, output_media) = split_plane_output_media(input_media, options)?;
    let output = builder.filter(
        FILTER_SPLIT_PLANE,
        &[input],
        output_media,
        SPLIT_PLANE_CONTRACT.compatibility(),
    )?;

    Ok((output, SplitPlaneExecutor::new(split)))
}

/// Validates merge-plane options and derives output media.
pub fn merge_planes_output_media(
    inputs: &[ClipMedia],
    options: MergePlanesOptions,
) -> Result<(ValidatedMergePlanes, ClipMedia)> {
    let MergePlanesOptions { format } = options;
    let output_format = pixelflow_core::resolve_format_alias(&format)?;
    if !SupportedFormats::new(ALL_PLANAR_FAMILIES, ALL_SAMPLE_TYPES).contains(&output_format) {
        return Err(plane_error(
            "filter.unsupported_format",
            format!(
                "filter '{FILTER_MERGE_PLANES}' does not support target format '{}'",
                output_format.name()
            ),
        ));
    }
    validate_plane_count(inputs, &output_format)?;
    let first = inputs
        .first()
        .expect("validated plane count must be non-zero");
    let (output_width, output_height) = fixed_resolution(first, FILTER_MERGE_PLANES)?;
    let frame_count = first.frame_count();
    let frame_rate = first.frame_rate();

    for (index, (input, descriptor)) in inputs.iter().zip(output_format.planes()).enumerate() {
        let input_format = fixed_gray_format(input, index)?;
        if input_format.sample_type() != output_format.sample_type()
            || input_format.bits_per_sample() != output_format.bits_per_sample()
        {
            return Err(plane_error(
                "filter.incompatible_planes",
                format!(
                    "filter '{FILTER_MERGE_PLANES}' input plane {index} format '{}' is incompatible with target '{}'",
                    input_format.name(),
                    output_format.name()
                ),
            ));
        }

        let (width, height) = fixed_resolution(input, FILTER_MERGE_PLANES)?;
        let expected_width = output_width.div_ceil(descriptor.width_divisor);
        let expected_height = output_height.div_ceil(descriptor.height_divisor);
        if width != expected_width || height != expected_height {
            return Err(plane_error(
                "filter.incompatible_planes",
                format!(
                    "filter '{FILTER_MERGE_PLANES}' input plane {index} must be {expected_width}x{expected_height}, got {width}x{height}"
                ),
            ));
        }
        if input.frame_count() != frame_count || input.frame_rate() != frame_rate {
            return Err(plane_error(
                "filter.incompatible_planes",
                format!(
                    "filter '{FILTER_MERGE_PLANES}' input plane {index} timing does not match plane 0"
                ),
            ));
        }
    }

    let output_media = ClipMedia::new(
        ClipFormat::Fixed(output_format.clone()),
        ClipResolution::Fixed {
            width: output_width,
            height: output_height,
        },
        frame_count,
        frame_rate,
    );

    Ok((
        ValidatedMergePlanes {
            output_format,
            output_width,
            output_height,
        },
        output_media,
    ))
}

/// Adds validated merge-planes filter node and returns matching render executor.
pub fn add_merge_planes_filter(
    builder: &mut GraphBuilder,
    inputs: &[Clip],
    input_media: &[ClipMedia],
    options: MergePlanesOptions,
) -> Result<(Clip, MergePlanesExecutor)> {
    let (merge, output_media) = merge_planes_output_media(input_media, options)?;
    let output = builder.filter(
        FILTER_MERGE_PLANES,
        inputs,
        output_media,
        MERGE_PLANES_CONTRACT.compatibility(),
    )?;

    Ok((output, MergePlanesExecutor::new(merge)))
}

fn parse_split_plane_options(options: &pixelflow_core::FilterOptions) -> Result<SplitPlaneOptions> {
    let plane = required_i64(
        options,
        FILTER_SPLIT_PLANE,
        "plane",
        "filter.invalid_split_plane",
    )?;

    Ok(SplitPlaneOptions::new(plane))
}

fn parse_merge_planes_options(
    options: &pixelflow_core::FilterOptions,
) -> Result<MergePlanesOptions> {
    let format = required_string(
        options,
        FILTER_MERGE_PLANES,
        "format",
        "filter.invalid_merge_planes",
    )?;

    Ok(MergePlanesOptions::new(format))
}

pub(crate) fn plan_split_plane(request: FilterPlanRequest<'_>) -> Result<FilterPlan> {
    let input = single_input_media(request, FILTER_SPLIT_PLANE)?;
    let output_media =
        split_plane_output_media(input, parse_split_plane_options(request.options())?)?.1;

    Ok(FilterPlan::new(
        output_media,
        SPLIT_PLANE_CONTRACT.compatibility(),
    ))
}

pub(crate) fn split_plane_executor_from_options(
    input_media: &[ClipMedia],
    options: &pixelflow_core::FilterOptions,
) -> Result<SplitPlaneExecutor> {
    let input = single_input_slice(input_media, FILTER_SPLIT_PLANE)?;
    let (split, _media) = split_plane_output_media(input, parse_split_plane_options(options)?)?;

    Ok(SplitPlaneExecutor::new(split))
}

pub(crate) fn plan_merge_planes(request: FilterPlanRequest<'_>) -> Result<FilterPlan> {
    let output_media = merge_planes_output_media(
        request.input_media(),
        parse_merge_planes_options(request.options())?,
    )?
    .1;

    Ok(FilterPlan::new(
        output_media,
        MERGE_PLANES_CONTRACT.compatibility(),
    ))
}

pub(crate) fn merge_planes_executor_from_options(
    input_media: &[ClipMedia],
    options: &pixelflow_core::FilterOptions,
) -> Result<MergePlanesExecutor> {
    let (merge, _media) =
        merge_planes_output_media(input_media, parse_merge_planes_options(options)?)?;

    Ok(MergePlanesExecutor::new(merge))
}

/// Render executor for validated zero-copy plane splits.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SplitPlaneExecutor {
    split: ValidatedSplitPlane,
}

impl SplitPlaneExecutor {
    /// Creates split executor from validated plane selection.
    #[must_use]
    pub const fn new(split: ValidatedSplitPlane) -> Self {
        Self { split }
    }

    /// Returns validated split selection.
    #[must_use]
    pub const fn split(&self) -> &ValidatedSplitPlane {
        &self.split
    }
}

impl FrameExecutor for SplitPlaneExecutor {
    fn prepare(&self, request: FrameRequest<'_>) -> Result<Frame> {
        let input = request.input_frame(0, request.frame_number())?;
        input.single_plane_view(self.split.plane_index, self.split.output_format.clone())
    }
}

/// Render executor for validated zero-copy plane merge operations.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct MergePlanesExecutor {
    merge: ValidatedMergePlanes,
}

impl MergePlanesExecutor {
    /// Creates merge executor from validated settings.
    #[must_use]
    pub const fn new(merge: ValidatedMergePlanes) -> Self {
        Self { merge }
    }

    /// Returns validated merge settings.
    #[must_use]
    pub const fn merge(&self) -> &ValidatedMergePlanes {
        &self.merge
    }
}

impl FrameExecutor for MergePlanesExecutor {
    fn prepare(&self, request: FrameRequest<'_>) -> Result<Frame> {
        let mut frames = Vec::with_capacity(self.merge.output_format.planes().len());
        for input_index in 0..self.merge.output_format.planes().len() {
            frames.push(request.input_frame(input_index, request.frame_number())?);
        }
        let sources: Vec<_> = frames.iter().map(|frame| (frame, 0_usize)).collect();
        let metadata = frames
            .first()
            .expect("validated merge requires at least one input frame")
            .metadata()
            .clone();

        Frame::from_plane_sources(
            self.merge.output_format.clone(),
            self.merge.output_width,
            self.merge.output_height,
            &sources,
            metadata,
        )
    }
}

fn fixed_resolution(input: &ClipMedia, filter_name: &str) -> Result<(usize, usize)> {
    match input.resolution() {
        ClipResolution::Fixed { width, height } => Ok((*width, *height)),
        ClipResolution::Variable => Err(plane_error(
            "filter.variable_resolution",
            format!("filter '{filter_name}' requires fixed input resolution"),
        )),
    }
}

fn validate_plane_index(plane: i64) -> Result<usize> {
    usize::try_from(plane).map_err(|_| {
        plane_error(
            "filter.invalid_split_plane",
            format!("filter '{FILTER_SPLIT_PLANE}' option 'plane' must be a non-negative integer"),
        )
    })
}

fn gray_format_for(format: &FormatDescriptor) -> Result<FormatDescriptor> {
    let alias = match format.bits_per_sample() {
        8 => "gray8",
        10 => "gray10",
        12 => "gray12",
        16 => "gray16",
        32 => "grayf32",
        bits => {
            return Err(plane_error(
                "filter.unsupported_plane_format",
                format!(
                    "filter '{FILTER_SPLIT_PLANE}' cannot derive gray format for bit depth {bits}"
                ),
            ));
        }
    };

    pixelflow_core::resolve_format_alias(alias)
}

fn validate_plane_count(inputs: &[ClipMedia], output_format: &FormatDescriptor) -> Result<()> {
    let expected = output_format.planes().len();
    if inputs.len() != expected {
        return Err(plane_error(
            "filter.invalid_merge_planes",
            format!(
                "filter '{FILTER_MERGE_PLANES}' target format '{}' requires {expected} input planes, got {}",
                output_format.name(),
                inputs.len()
            ),
        ));
    }

    Ok(())
}

fn fixed_gray_format(input: &ClipMedia, index: usize) -> Result<&FormatDescriptor> {
    let ClipFormat::Fixed(format) = input.format() else {
        return Err(plane_error(
            "filter.variable_format",
            format!("filter '{FILTER_MERGE_PLANES}' input plane {index} requires fixed format"),
        ));
    };
    if format.family() != FormatFamily::Gray {
        return Err(plane_error(
            "filter.incompatible_planes",
            format!(
                "filter '{FILTER_MERGE_PLANES}' input plane {index} must be Gray, got '{}'",
                format.name()
            ),
        ));
    }

    Ok(format)
}

fn plane_error(code: &'static str, message: impl Into<String>) -> PixelFlowError {
    PixelFlowError::new(ErrorCategory::Format, ErrorCode::new(code), message)
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};

    use pixelflow_core::{
        ClipFormat, ClipMedia, ClipResolution, ErrorCategory, ErrorCode, Frame, FrameCount,
        FrameExecutor, FrameRate, FrameRequest, GraphBuilder, MetadataValue, Rational,
        RenderEngine, RenderExecutorMap, RenderOptions, WorkerPoolConfig,
    };

    use crate::EXACT_GOLDEN_TOLERANCE;
    use crate::testkit::{
        assert_plane_u8_near, fixed_media, synthetic_u8_frame, with_frame_number_metadata,
    };

    use super::{
        MergePlanesOptions, SplitPlaneOptions, add_merge_planes_filter, add_split_plane_filter,
        merge_planes_output_media, split_plane_output_media,
    };

    #[test]
    fn split_plane_validates_yuv420_plane_dimensions_and_gray_format() {
        let input = fixed_media("yuv420p8", 5, 3);

        let (split, output) = split_plane_output_media(&input, SplitPlaneOptions::new(1))
            .expect("plane 1 split should validate");

        assert_eq!(split.plane_index(), 1);
        assert!(matches!(output.format(), ClipFormat::Fixed(format) if format.name() == "gray8"));
        assert_eq!(
            output.resolution(),
            &ClipResolution::Fixed {
                width: 3,
                height: 2
            }
        );
        assert_eq!(output.frame_count(), FrameCount::Finite(24));
        assert_eq!(
            output.frame_rate(),
            FrameRate::Cfr(Rational {
                numerator: 24_000,
                denominator: 1_001,
            })
        );
    }

    #[test]
    fn split_plane_accepts_gray_and_planar_rgb_indices() {
        let gray = fixed_media("gray10", 4, 2);
        let rgb = fixed_media("rgbpf32", 4, 2);

        let (_, gray_output) = split_plane_output_media(&gray, SplitPlaneOptions::new(0))
            .expect("gray split should validate");
        let (blue, blue_output) = split_plane_output_media(&rgb, SplitPlaneOptions::new(2))
            .expect("plane 2 split should validate");

        assert!(
            matches!(gray_output.format(), ClipFormat::Fixed(format) if format.name() == "gray10")
        );
        assert_eq!(blue.plane_index(), 2);
        assert!(
            matches!(blue_output.format(), ClipFormat::Fixed(format) if format.name() == "grayf32")
        );
        assert_eq!(
            blue_output.resolution(),
            &ClipResolution::Fixed {
                width: 4,
                height: 2
            }
        );
    }

    #[test]
    fn split_plane_rejects_negative_and_out_of_range_indices() {
        let input = fixed_media("yuv444p8", 4, 2);

        let negative = split_plane_output_media(&input, SplitPlaneOptions::new(-1))
            .expect_err("negative plane index should fail");
        assert_eq!(negative.category(), ErrorCategory::Format);
        assert_eq!(
            negative.code(),
            ErrorCode::new("filter.invalid_split_plane")
        );

        let invalid = split_plane_output_media(&input, SplitPlaneOptions::new(3))
            .expect_err("out-of-range plane index should fail");
        assert_eq!(invalid.category(), ErrorCategory::Format);
        assert_eq!(invalid.code(), ErrorCode::new("filter.unsupported_plane"));
    }

    #[test]
    fn split_plane_executor_returns_zero_copy_plane_with_metadata() {
        struct RecordingSource {
            last_frame: Arc<Mutex<Option<Frame>>>,
        }

        impl FrameExecutor for RecordingSource {
            fn prepare(&self, request: FrameRequest<'_>) -> pixelflow_core::Result<Frame> {
                let frame = synthetic_u8_frame("yuv420p8", 4, 4, |plane, x, y| {
                    u8::try_from(plane * 40 + x + y * 10).expect("fixture sample fits u8")
                })?;
                let frame = with_frame_number_metadata(&frame, request.frame_number())?;
                *self.last_frame.lock().expect("test mutex should lock") = Some(frame.clone());
                Ok(frame)
            }
        }

        let input_media = fixed_media("yuv420p8", 4, 4);
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", input_media.clone());
        let (split_clip, executor) = add_split_plane_filter(
            &mut builder,
            source,
            &input_media,
            SplitPlaneOptions::new(2),
        )
        .expect("split node should build");
        builder.set_output(split_clip);
        let graph = builder.build();

        let last_frame = Arc::new(Mutex::new(None));
        let mut executors = RenderExecutorMap::new();
        executors.insert(
            source.node_id(),
            Arc::new(RecordingSource {
                last_frame: last_frame.clone(),
            }),
        );
        executors.insert(split_clip.node_id(), Arc::new(executor));

        let output = RenderEngine::new(WorkerPoolConfig::new(1))
            .render_ordered(graph, executors, RenderOptions::new(2, Some(3)))
            .expect("render should start")
            .next()
            .expect("one frame should render")
            .expect("split should succeed");
        let source_frame = last_frame
            .lock()
            .expect("test mutex should lock")
            .clone()
            .expect("source frame recorded");

        assert_eq!(output.width(), 2);
        assert_eq!(output.height(), 2);
        assert!(source_frame.shares_plane_storage_at(2, &output, 0));
        assert_eq!(
            output.metadata().get("core:frame_number"),
            Some(&MetadataValue::Int(2))
        );
        assert_plane_u8_near(&output, 0, &[&[80, 81], &[90, 91]], EXACT_GOLDEN_TOLERANCE);
    }

    #[test]
    fn merge_planes_validates_yuv420_plane_shapes_and_timing() {
        let y = fixed_media("gray8", 5, 3);
        let u = fixed_media("gray8", 3, 2);
        let v = fixed_media("gray8", 3, 2);

        let (merge, output) =
            merge_planes_output_media(&[y.clone(), u, v], MergePlanesOptions::new("yuv420p8"))
                .expect("compatible YUV420 planes should validate");

        assert_eq!(merge.output_width(), 5);
        assert_eq!(merge.output_height(), 3);
        assert!(
            matches!(output.format(), ClipFormat::Fixed(format) if format.name() == "yuv420p8")
        );
        assert_eq!(
            output.resolution(),
            &ClipResolution::Fixed {
                width: 5,
                height: 3
            }
        );
        assert_eq!(output.frame_count(), y.frame_count());
        assert_eq!(output.frame_rate(), y.frame_rate());
    }

    #[test]
    fn merge_planes_validates_gray_and_rgb_targets() {
        let gray = fixed_media("gray10", 2, 2);
        let r = fixed_media("grayf32", 4, 2);
        let g = fixed_media("grayf32", 4, 2);
        let b = fixed_media("grayf32", 4, 2);

        let (_, gray_output) = merge_planes_output_media(
            std::slice::from_ref(&gray),
            MergePlanesOptions::new("gray10"),
        )
        .expect("gray target should validate with one plane");
        let (_, rgb_output) =
            merge_planes_output_media(&[r, g, b], MergePlanesOptions::new("rgbpf32"))
                .expect("RGB target should validate with three planes");

        assert_eq!(
            gray_output.resolution(),
            &ClipResolution::Fixed {
                width: 2,
                height: 2
            }
        );
        assert!(
            matches!(rgb_output.format(), ClipFormat::Fixed(format) if format.name() == "rgbpf32")
        );
        assert_eq!(
            rgb_output.resolution(),
            &ClipResolution::Fixed {
                width: 4,
                height: 2
            }
        );
    }

    #[test]
    fn merge_planes_rejects_wrong_plane_count_shape_sample_and_timing() {
        let y = fixed_media("gray8", 4, 4);
        let u = fixed_media("gray8", 2, 2);
        let bad_shape = fixed_media("gray8", 3, 2);
        let bad_sample = fixed_media("gray10", 2, 2);
        let bad_rate = ClipMedia::new(
            bad_shape.format().clone(),
            bad_shape.resolution().clone(),
            bad_shape.frame_count(),
            FrameRate::Cfr(Rational {
                numerator: 30,
                denominator: 1,
            }),
        );

        let wrong_count =
            merge_planes_output_media(&[y.clone(), u.clone()], MergePlanesOptions::new("yuv420p8"))
                .expect_err("YUV target requires three inputs");
        assert_eq!(wrong_count.category(), ErrorCategory::Format);
        assert_eq!(
            wrong_count.code(),
            ErrorCode::new("filter.invalid_merge_planes")
        );

        let wrong_shape = merge_planes_output_media(
            &[y.clone(), bad_shape, u.clone()],
            MergePlanesOptions::new("yuv420p8"),
        )
        .expect_err("bad chroma shape should fail");
        assert_eq!(wrong_shape.category(), ErrorCategory::Format);
        assert_eq!(
            wrong_shape.code(),
            ErrorCode::new("filter.incompatible_planes")
        );

        let wrong_sample = merge_planes_output_media(
            &[y.clone(), bad_sample.clone(), bad_sample],
            MergePlanesOptions::new("yuv420p8"),
        )
        .expect_err("bad sample type should fail");
        assert_eq!(wrong_sample.category(), ErrorCategory::Format);
        assert_eq!(
            wrong_sample.code(),
            ErrorCode::new("filter.incompatible_planes")
        );

        let wrong_rate =
            merge_planes_output_media(&[y, u, bad_rate], MergePlanesOptions::new("yuv420p8"))
                .expect_err("bad frame rate should fail");
        assert_eq!(wrong_rate.category(), ErrorCategory::Format);
        assert_eq!(
            wrong_rate.code(),
            ErrorCode::new("filter.incompatible_planes")
        );
    }

    #[test]
    fn merge_planes_executor_combines_plane_storage_without_copying() {
        struct PlaneSource {
            frame: Frame,
            last_frame: Arc<Mutex<Option<Frame>>>,
        }

        impl FrameExecutor for PlaneSource {
            fn prepare(&self, _request: FrameRequest<'_>) -> pixelflow_core::Result<Frame> {
                *self.last_frame.lock().expect("test mutex should lock") = Some(self.frame.clone());
                Ok(self.frame.clone())
            }
        }

        let y_frame = synthetic_u8_frame("gray8", 4, 4, |_plane, x, y| {
            u8::try_from(x + y * 10).expect("fixture sample fits u8")
        })
        .expect("Y frame should build");
        let u_frame = synthetic_u8_frame("gray8", 2, 2, |_plane, x, y| {
            u8::try_from(40 + x + y * 10).expect("fixture sample fits u8")
        })
        .expect("U frame should build");
        let v_frame = synthetic_u8_frame("gray8", 2, 2, |_plane, x, y| {
            u8::try_from(80 + x + y * 10).expect("fixture sample fits u8")
        })
        .expect("V frame should build");
        let y_frame = with_frame_number_metadata(&y_frame, 7).expect("metadata should set");

        let y_media = fixed_media("gray8", 4, 4);
        let u_media = fixed_media("gray8", 2, 2);
        let v_media = fixed_media("gray8", 2, 2);
        let mut builder = GraphBuilder::new();
        let y = builder.source("y", y_media.clone());
        let u = builder.source("u", u_media.clone());
        let v = builder.source("v", v_media.clone());
        let (merged, executor) = add_merge_planes_filter(
            &mut builder,
            &[y, u, v],
            &[y_media, u_media, v_media],
            MergePlanesOptions::new("yuv420p8"),
        )
        .expect("merge node should build");
        builder.set_output(merged);
        let graph = builder.build();

        let y_seen = Arc::new(Mutex::new(None));
        let u_seen = Arc::new(Mutex::new(None));
        let v_seen = Arc::new(Mutex::new(None));
        let mut executors = RenderExecutorMap::new();
        executors.insert(
            y.node_id(),
            Arc::new(PlaneSource {
                frame: y_frame,
                last_frame: y_seen.clone(),
            }),
        );
        executors.insert(
            u.node_id(),
            Arc::new(PlaneSource {
                frame: u_frame,
                last_frame: u_seen.clone(),
            }),
        );
        executors.insert(
            v.node_id(),
            Arc::new(PlaneSource {
                frame: v_frame,
                last_frame: v_seen.clone(),
            }),
        );
        executors.insert(merged.node_id(), Arc::new(executor));

        let output = RenderEngine::new(WorkerPoolConfig::new(1))
            .render_ordered(graph, executors, RenderOptions::new(0, Some(1)))
            .expect("render should start")
            .next()
            .expect("one frame should render")
            .expect("merge should succeed");

        let y_source = y_seen
            .lock()
            .expect("lock")
            .clone()
            .expect("Y source recorded");
        let u_source = u_seen
            .lock()
            .expect("lock")
            .clone()
            .expect("U source recorded");
        let v_source = v_seen
            .lock()
            .expect("lock")
            .clone()
            .expect("V source recorded");
        assert!(y_source.shares_plane_storage_at(0, &output, 0));
        assert!(u_source.shares_plane_storage_at(0, &output, 1));
        assert!(v_source.shares_plane_storage_at(0, &output, 2));
        assert_eq!(
            output.metadata().get("core:frame_number"),
            Some(&MetadataValue::Int(7))
        );
        assert_plane_u8_near(
            &output,
            0,
            &[
                &[0, 1, 2, 3],
                &[10, 11, 12, 13],
                &[20, 21, 22, 23],
                &[30, 31, 32, 33],
            ],
            EXACT_GOLDEN_TOLERANCE,
        );
        assert_plane_u8_near(&output, 1, &[&[40, 41], &[50, 51]], EXACT_GOLDEN_TOLERANCE);
        assert_plane_u8_near(&output, 2, &[&[80, 81], &[90, 91]], EXACT_GOLDEN_TOLERANCE);
    }

    #[test]
    fn split_then_merge_yuv420_roundtrips_plane_data_with_shared_storage() {
        struct Source;

        impl FrameExecutor for Source {
            fn prepare(&self, request: FrameRequest<'_>) -> pixelflow_core::Result<Frame> {
                let frame = synthetic_u8_frame("yuv420p8", 4, 4, |plane, x, y| {
                    u8::try_from(plane * 40 + x + y * 10).expect("fixture sample fits u8")
                })?;
                with_frame_number_metadata(&frame, request.frame_number())
            }
        }

        let input_media = fixed_media("yuv420p8", 4, 4);
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", input_media.clone());
        let (y, y_executor) = add_split_plane_filter(
            &mut builder,
            source,
            &input_media,
            SplitPlaneOptions::new(0),
        )
        .expect("Y split should build");
        let (u, u_executor) = add_split_plane_filter(
            &mut builder,
            source,
            &input_media,
            SplitPlaneOptions::new(1),
        )
        .expect("U split should build");
        let (v, v_executor) = add_split_plane_filter(
            &mut builder,
            source,
            &input_media,
            SplitPlaneOptions::new(2),
        )
        .expect("V split should build");
        let split_media = [
            fixed_media("gray8", 4, 4),
            fixed_media("gray8", 2, 2),
            fixed_media("gray8", 2, 2),
        ];
        let (merged, merge_executor) = add_merge_planes_filter(
            &mut builder,
            &[y, u, v],
            &split_media,
            MergePlanesOptions::new("yuv420p8"),
        )
        .expect("merge should build");
        builder.set_output(merged);
        let graph = builder.build();

        let mut executors = RenderExecutorMap::new();
        executors.insert(source.node_id(), Arc::new(Source));
        executors.insert(y.node_id(), Arc::new(y_executor));
        executors.insert(u.node_id(), Arc::new(u_executor));
        executors.insert(v.node_id(), Arc::new(v_executor));
        executors.insert(merged.node_id(), Arc::new(merge_executor));

        let output = RenderEngine::new(WorkerPoolConfig::new(1))
            .render_ordered(graph, executors, RenderOptions::new(0, Some(1)))
            .expect("render should start")
            .next()
            .expect("one frame should render")
            .expect("roundtrip should render");

        assert_plane_u8_near(
            &output,
            0,
            &[
                &[0, 1, 2, 3],
                &[10, 11, 12, 13],
                &[20, 21, 22, 23],
                &[30, 31, 32, 33],
            ],
            EXACT_GOLDEN_TOLERANCE,
        );
        assert_plane_u8_near(&output, 1, &[&[40, 41], &[50, 51]], EXACT_GOLDEN_TOLERANCE);
        assert_plane_u8_near(&output, 2, &[&[80, 81], &[90, 91]], EXACT_GOLDEN_TOLERANCE);
        assert_eq!(
            output.metadata().get("core:frame_number"),
            Some(&MetadataValue::Int(0))
        );
    }
}