pixelflow-filters 0.1.0

//! Official property helper filters.

use pixelflow_core::{
    Clip, ClipMedia, ErrorCategory, ErrorCode, FilterCompatibility, FilterOptionValue,
    FilterOptions, FilterPlan, FilterPlanRequest, Frame, FrameExecutor, FrameRequest, GraphBuilder,
    MetadataKind, MetadataSchema, MetadataValue, PixelFlowError, Result,
};

use crate::{
    OfficialFilterContract, SupportedFormats,
    contracts::{ALL_PLANAR_FAMILIES, ALL_SAMPLE_TYPES},
    options::{filter_option_error, optional_string, required_string, single_input_media},
};

/// Official property set filter script name.
pub const FILTER_SET_PROP: &str = "set_prop";

/// Official property clear filter script name.
pub const FILTER_CLEAR_PROP: &str = "clear_prop";

/// Official property-copy filter script name.
pub const FILTER_COPY_PROPS: &str = "copy_props";

/// Official property validation filter script name.
pub const FILTER_REQUIRE_PROP: &str = "require_prop";

/// Contract for `set_prop`.
pub const SET_PROP_CONTRACT: OfficialFilterContract = prop_contract(FILTER_SET_PROP);

/// Contract for `clear_prop`.
pub const CLEAR_PROP_CONTRACT: OfficialFilterContract = prop_contract(FILTER_CLEAR_PROP);

/// Contract for `copy_props`.
pub const COPY_PROPS_CONTRACT: OfficialFilterContract = OfficialFilterContract::new(
    FILTER_COPY_PROPS,
    SupportedFormats::new(ALL_PLANAR_FAMILIES, ALL_SAMPLE_TYPES),
    FilterCompatibility::Custom,
);

/// Contract for `require_prop`.
pub const REQUIRE_PROP_CONTRACT: OfficialFilterContract = prop_contract(FILTER_REQUIRE_PROP);

const fn prop_contract(name: &'static str) -> OfficialFilterContract {
    OfficialFilterContract::new(
        name,
        SupportedFormats::new(ALL_PLANAR_FAMILIES, ALL_SAMPLE_TYPES),
        FilterCompatibility::Preserve,
    )
}

/// Options for `set_prop`.
#[derive(Clone, Debug, PartialEq)]
pub struct SetPropOptions {
    key: String,
    value: MetadataValue,
}

impl SetPropOptions {
    /// Creates typed set-prop options.
    #[must_use]
    pub const fn new(key: String, value: MetadataValue) -> Self {
        Self { key, value }
    }

    /// Returns metadata key.
    #[must_use]
    pub fn key(&self) -> &str {
        &self.key
    }

    /// Returns metadata value to write.
    #[must_use]
    pub const fn value(&self) -> &MetadataValue {
        &self.value
    }
}

/// Options for `clear_prop`.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ClearPropOptions {
    key: String,
}

impl ClearPropOptions {
    /// Creates typed clear-prop options.
    #[must_use]
    pub const fn new(key: String) -> Self {
        Self { key }
    }

    /// Returns metadata key.
    #[must_use]
    pub fn key(&self) -> &str {
        &self.key
    }
}

/// Options for `require_prop`.
#[derive(Clone, Debug, PartialEq)]
pub struct RequirePropOptions {
    key: String,
    kind: Option<MetadataKind>,
    value: Option<MetadataValue>,
}

impl RequirePropOptions {
    /// Creates typed require-prop options.
    #[must_use]
    pub const fn new(
        key: String,
        kind: Option<MetadataKind>,
        value: Option<MetadataValue>,
    ) -> Self {
        Self { key, kind, value }
    }

    /// Returns metadata key.
    #[must_use]
    pub fn key(&self) -> &str {
        &self.key
    }

    /// Returns optional required metadata kind.
    #[must_use]
    pub const fn kind(&self) -> Option<MetadataKind> {
        self.kind
    }

    /// Returns optional required metadata value.
    #[must_use]
    pub const fn value(&self) -> Option<&MetadataValue> {
        self.value.as_ref()
    }
}

/// Converts script/planner option values into metadata values.
#[must_use]
pub(crate) fn metadata_value_from_option(value: &FilterOptionValue) -> MetadataValue {
    match value {
        FilterOptionValue::None => MetadataValue::None,
        FilterOptionValue::String(value) => MetadataValue::String(value.clone()),
        FilterOptionValue::Bool(value) => MetadataValue::Bool(*value),
        FilterOptionValue::Int(value) => MetadataValue::Int(*value),
        FilterOptionValue::Float(value) => MetadataValue::Float(*value),
        FilterOptionValue::Array(values) => {
            MetadataValue::Array(values.iter().map(metadata_value_from_option).collect())
        }
        FilterOptionValue::Rational(value) => MetadataValue::Rational(*value),
        FilterOptionValue::Blob(value) => MetadataValue::Blob(value.clone()),
    }
}

/// Validates `set_prop` and returns preserved output media.
pub fn set_prop_output_media(
    input: &ClipMedia,
    schema: &MetadataSchema,
    options: &SetPropOptions,
) -> Result<ClipMedia> {
    SET_PROP_CONTRACT.validate_input_media(input)?;
    schema.validate_value(options.key(), options.value())?;
    Ok(input.clone())
}

/// Validates `clear_prop` and returns preserved output media.
pub fn clear_prop_output_media(
    input: &ClipMedia,
    schema: &MetadataSchema,
    options: &ClearPropOptions,
) -> Result<ClipMedia> {
    CLEAR_PROP_CONTRACT.validate_input_media(input)?;
    schema.validate_value(options.key(), &MetadataValue::None)?;
    Ok(input.clone())
}

/// Validates `copy_props` and returns target-preserved output media.
pub fn copy_props_output_media(inputs: &[ClipMedia]) -> Result<ClipMedia> {
    let (prop_src, target) = copy_props_input_media(inputs)?;
    COPY_PROPS_CONTRACT.validate_input_media(target)?;

    if target.frame_count() != prop_src.frame_count() {
        return Err(PixelFlowError::new(
            ErrorCategory::Format,
            ErrorCode::new("filter.incompatible_props"),
            format!(
                "filter '{FILTER_COPY_PROPS}' metadata source frame count must match target frame count"
            ),
        ));
    }

    Ok(target.clone())
}

/// Validates `require_prop` and returns preserved output media.
pub fn require_prop_output_media(
    input: &ClipMedia,
    schema: &MetadataSchema,
    options: &RequirePropOptions,
) -> Result<ClipMedia> {
    REQUIRE_PROP_CONTRACT.validate_input_media(input)?;
    validate_require_options(schema, options)?;

    Ok(input.clone())
}

/// Adds validated `set_prop` filter node and executor.
pub fn add_set_prop_filter(
    builder: &mut GraphBuilder,
    input: Clip,
    input_media: &ClipMedia,
    schema: &MetadataSchema,
    options: SetPropOptions,
) -> Result<(Clip, SetPropExecutor)> {
    let output_media = set_prop_output_media(input_media, schema, &options)?;
    let output = builder.filter(
        FILTER_SET_PROP,
        &[input],
        output_media,
        SET_PROP_CONTRACT.compatibility(),
    )?;

    Ok((output, SetPropExecutor::new(schema.clone(), options)))
}

/// Adds validated `clear_prop` filter node and executor.
pub fn add_clear_prop_filter(
    builder: &mut GraphBuilder,
    input: Clip,
    input_media: &ClipMedia,
    schema: &MetadataSchema,
    options: ClearPropOptions,
) -> Result<(Clip, ClearPropExecutor)> {
    let output_media = clear_prop_output_media(input_media, schema, &options)?;
    let output = builder.filter(
        FILTER_CLEAR_PROP,
        &[input],
        output_media,
        CLEAR_PROP_CONTRACT.compatibility(),
    )?;

    Ok((output, ClearPropExecutor::new(schema.clone(), options)))
}

/// Adds validated `copy_props` filter node and executor.
pub fn add_copy_props_filter(
    builder: &mut GraphBuilder,
    prop_src: Clip,
    target: Clip,
    input_media: &[ClipMedia],
) -> Result<(Clip, CopyPropsExecutor)> {
    let output_media = copy_props_output_media(input_media)?;
    let output = builder.filter(
        FILTER_COPY_PROPS,
        &[prop_src, target],
        output_media,
        COPY_PROPS_CONTRACT.compatibility(),
    )?;

    Ok((output, CopyPropsExecutor::new()))
}

/// Adds validated `require_prop` filter node and executor.
pub fn add_require_prop_filter(
    builder: &mut GraphBuilder,
    input: Clip,
    input_media: &ClipMedia,
    schema: &MetadataSchema,
    options: RequirePropOptions,
) -> Result<(Clip, RequirePropExecutor)> {
    let output_media = require_prop_output_media(input_media, schema, &options)?;
    let output = builder.filter(
        FILTER_REQUIRE_PROP,
        &[input],
        output_media,
        REQUIRE_PROP_CONTRACT.compatibility(),
    )?;

    Ok((output, RequirePropExecutor::new(schema.clone(), options)?))
}

fn parse_set_prop_options(options: &pixelflow_core::FilterOptions) -> Result<SetPropOptions> {
    let key = required_string(options, FILTER_SET_PROP, "key", "filter.invalid_prop")?;
    let value = options.get("value").ok_or_else(|| {
        filter_option_error(
            "filter.invalid_prop",
            format!("filter '{FILTER_SET_PROP}' requires option 'value'"),
        )
    })?;

    Ok(SetPropOptions::new(
        key.to_owned(),
        metadata_value_from_option(value),
    ))
}

fn parse_clear_prop_options(options: &pixelflow_core::FilterOptions) -> Result<ClearPropOptions> {
    let key = required_string(options, FILTER_CLEAR_PROP, "key", "filter.invalid_prop")?;
    Ok(ClearPropOptions::new(key.to_owned()))
}

fn parse_copy_props_options(options: &FilterOptions) -> Result<()> {
    if options.is_empty() {
        return Ok(());
    }

    Err(filter_option_error(
        "filter.invalid_copy_props",
        format!("filter '{FILTER_COPY_PROPS}' accepts no options"),
    ))
}

fn parse_require_prop_options(
    options: &pixelflow_core::FilterOptions,
) -> Result<RequirePropOptions> {
    let key = required_string(options, FILTER_REQUIRE_PROP, "key", "filter.invalid_prop")?;
    let kind = optional_string(options, FILTER_REQUIRE_PROP, "kind", "filter.invalid_prop")?
        .map(|value| metadata_kind_from_name(FILTER_REQUIRE_PROP, value))
        .transpose()?;
    let value = options.get("value").map(metadata_value_from_option);

    Ok(RequirePropOptions::new(key.to_owned(), kind, value))
}

pub(crate) fn plan_set_prop(request: FilterPlanRequest<'_>) -> Result<FilterPlan> {
    let input = single_input_media(request, FILTER_SET_PROP)?;
    let options = parse_set_prop_options(request.options())?;
    let output_media = set_prop_output_media(input, request.metadata_schema(), &options)?;

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

pub(crate) fn set_prop_executor_from_options(
    input_media: &[ClipMedia],
    schema: &MetadataSchema,
    options: &pixelflow_core::FilterOptions,
) -> Result<SetPropExecutor> {
    let request = FilterPlanRequest::new(input_media, options, schema);
    let input = single_input_media(request, FILTER_SET_PROP)?;
    let parsed = parse_set_prop_options(options)?;
    set_prop_output_media(input, schema, &parsed)?;

    Ok(SetPropExecutor::new(schema.clone(), parsed))
}

pub(crate) fn plan_clear_prop(request: FilterPlanRequest<'_>) -> Result<FilterPlan> {
    let input = single_input_media(request, FILTER_CLEAR_PROP)?;
    let options = parse_clear_prop_options(request.options())?;
    let output_media = clear_prop_output_media(input, request.metadata_schema(), &options)?;

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

pub(crate) fn clear_prop_executor_from_options(
    input_media: &[ClipMedia],
    schema: &MetadataSchema,
    options: &pixelflow_core::FilterOptions,
) -> Result<ClearPropExecutor> {
    let request = FilterPlanRequest::new(input_media, options, schema);
    let input = single_input_media(request, FILTER_CLEAR_PROP)?;
    let parsed = parse_clear_prop_options(options)?;
    clear_prop_output_media(input, schema, &parsed)?;

    Ok(ClearPropExecutor::new(schema.clone(), parsed))
}

pub(crate) fn plan_copy_props(request: FilterPlanRequest<'_>) -> Result<FilterPlan> {
    parse_copy_props_options(request.options())?;
    let output_media = copy_props_output_media(request.input_media())?;

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

pub(crate) fn copy_props_executor_from_options(
    input_media: &[ClipMedia],
    options: &FilterOptions,
) -> Result<CopyPropsExecutor> {
    parse_copy_props_options(options)?;
    copy_props_output_media(input_media)?;

    Ok(CopyPropsExecutor::new())
}

pub(crate) fn plan_require_prop(request: FilterPlanRequest<'_>) -> Result<FilterPlan> {
    let input = single_input_media(request, FILTER_REQUIRE_PROP)?;
    let options = parse_require_prop_options(request.options())?;
    let output_media = require_prop_output_media(input, request.metadata_schema(), &options)?;

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

pub(crate) fn require_prop_executor_from_options(
    input_media: &[ClipMedia],
    schema: &MetadataSchema,
    options: &pixelflow_core::FilterOptions,
) -> Result<RequirePropExecutor> {
    let request = FilterPlanRequest::new(input_media, options, schema);
    let input = single_input_media(request, FILTER_REQUIRE_PROP)?;
    let parsed = parse_require_prop_options(options)?;
    require_prop_output_media(input, schema, &parsed)?;

    RequirePropExecutor::new(schema.clone(), parsed)
}

/// Render executor for metadata-only `set_prop` updates.
#[derive(Clone, Debug, PartialEq)]
pub struct SetPropExecutor {
    schema: MetadataSchema,
    options: SetPropOptions,
}

impl SetPropExecutor {
    /// Creates `set_prop` executor from validated schema and options.
    #[must_use]
    pub const fn new(schema: MetadataSchema, options: SetPropOptions) -> Self {
        Self { schema, options }
    }

    fn apply(&self, input: &Frame) -> Result<Frame> {
        let mut metadata = input.metadata().clone();
        metadata.set(
            &self.schema,
            self.options.key(),
            self.options.value().clone(),
        )?;
        Ok(input.with_metadata(metadata))
    }

    #[cfg(test)]
    pub(crate) fn apply_for_tests(&self, input: &Frame) -> Result<Frame> {
        self.apply(input)
    }
}

impl FrameExecutor for SetPropExecutor {
    fn prepare(&self, request: FrameRequest<'_>) -> Result<Frame> {
        let input = request.input_frame(0, request.frame_number())?;
        self.apply(&input)
    }
}

/// Render executor for metadata-only `clear_prop` updates.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ClearPropExecutor {
    schema: MetadataSchema,
    options: ClearPropOptions,
}

impl ClearPropExecutor {
    /// Creates `clear_prop` executor from validated schema and options.
    #[must_use]
    pub const fn new(schema: MetadataSchema, options: ClearPropOptions) -> Self {
        Self { schema, options }
    }

    fn apply(&self, input: &Frame) -> Result<Frame> {
        let mut metadata = input.metadata().clone();
        metadata.clear(&self.schema, self.options.key())?;
        Ok(input.with_metadata(metadata))
    }

    #[cfg(test)]
    pub(crate) fn apply_for_tests(&self, input: &Frame) -> Result<Frame> {
        self.apply(input)
    }
}

impl FrameExecutor for ClearPropExecutor {
    fn prepare(&self, request: FrameRequest<'_>) -> Result<Frame> {
        let input = request.input_frame(0, request.frame_number())?;
        self.apply(&input)
    }
}

/// Render executor for copying all frame metadata from input 0 onto input 1.
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct CopyPropsExecutor;

impl CopyPropsExecutor {
    /// Creates `copy_props` executor.
    #[must_use]
    pub const fn new() -> Self {
        Self
    }

    fn apply(&self, prop_src: &Frame, target: &Frame) -> Frame {
        target.with_metadata(prop_src.metadata().clone())
    }

    #[cfg(test)]
    pub(crate) fn apply_for_tests(&self, prop_src: &Frame, target: &Frame) -> Frame {
        self.apply(prop_src, target)
    }
}

impl FrameExecutor for CopyPropsExecutor {
    fn prepare(&self, request: FrameRequest<'_>) -> Result<Frame> {
        let prop_src = request.input_frame(0, request.frame_number())?;
        let target = request.input_frame(1, request.frame_number())?;
        Ok(self.apply(&prop_src, &target))
    }
}

/// Render executor for read-side `require_prop` validation.
#[derive(Clone, Debug, PartialEq)]
pub struct RequirePropExecutor {
    schema: MetadataSchema,
    validated: ValidatedRequireProp,
}

#[derive(Clone, Debug, PartialEq)]
struct ValidatedRequireProp {
    key: String,
    expected_kind: MetadataKind,
    required_value: Option<MetadataValue>,
}

impl RequirePropExecutor {
    /// Creates `require_prop` executor from validated schema and options.
    pub fn new(schema: MetadataSchema, options: RequirePropOptions) -> Result<Self> {
        Ok(Self {
            validated: validate_require_options_owned(&schema, options)?,
            schema,
        })
    }

    fn apply(&self, input: Frame) -> Result<Frame> {
        require_value(
            &self.schema,
            &self.validated,
            input.metadata().get(&self.validated.key),
        )?;
        Ok(input)
    }

    #[cfg(test)]
    pub(crate) fn apply_for_tests(&self, input: Frame) -> Result<Frame> {
        self.apply(input)
    }
}

impl FrameExecutor for RequirePropExecutor {
    fn prepare(&self, request: FrameRequest<'_>) -> Result<Frame> {
        self.apply(request.input_frame(0, request.frame_number())?)
    }
}

fn metadata_kind_from_name(filter_name: &str, value: &str) -> Result<MetadataKind> {
    match value.to_ascii_lowercase().as_str() {
        "bool" => Ok(MetadataKind::Bool),
        "int" => Ok(MetadataKind::Int),
        "float" => Ok(MetadataKind::Float),
        "string" => Ok(MetadataKind::String),
        "array" => Ok(MetadataKind::Array),
        "rational" => Ok(MetadataKind::Rational),
        "blob" | "binary" => Ok(MetadataKind::Blob),
        _ => Err(filter_option_error(
            "filter.invalid_prop",
            format!("filter '{filter_name}' option 'kind' has unsupported metadata kind '{value}'"),
        )),
    }
}

fn copy_props_input_media(inputs: &[ClipMedia]) -> Result<(&ClipMedia, &ClipMedia)> {
    match inputs {
        [prop_src, target] => Ok((prop_src, target)),
        inputs => Err(PixelFlowError::new(
            ErrorCategory::Graph,
            ErrorCode::new("graph.invalid_filter_inputs"),
            format!(
                "filter '{FILTER_COPY_PROPS}' requires exactly two input clips, got {}",
                inputs.len()
            ),
        )),
    }
}

fn registered_key_kind(schema: &MetadataSchema, key: &str) -> Result<MetadataKind> {
    schema.kind(key).ok_or_else(|| {
        PixelFlowError::new(
            ErrorCategory::Plugin,
            ErrorCode::new("metadata.unregistered_key"),
            format!("metadata key '{key}' is not registered"),
        )
    })
}

fn require_value(
    schema: &MetadataSchema,
    options: &ValidatedRequireProp,
    actual: Option<&MetadataValue>,
) -> Result<()> {
    let actual = actual.ok_or_else(|| metadata_missing_error(schema, &options.key))?;

    if matches!(actual, MetadataValue::None) {
        return Err(metadata_missing_error(schema, &options.key));
    }

    if let Some(actual_kind) = actual.kind()
        && actual_kind != options.expected_kind
    {
        return Err(metadata_type_error(
            schema,
            &options.key,
            options.expected_kind,
            actual_kind,
        ));
    }
    if let Some(required_value) = &options.required_value
        && actual != required_value
    {
        return Err(metadata_value_mismatch_error(schema, &options.key));
    }

    Ok(())
}

fn validate_require_options(
    schema: &MetadataSchema,
    options: &RequirePropOptions,
) -> Result<ValidatedRequireProp> {
    validate_require_parts(schema, options.key(), options.kind(), options.value())?;

    Ok(ValidatedRequireProp {
        key: options.key().to_owned(),
        expected_kind: registered_key_kind(schema, options.key())?,
        required_value: options.value().cloned(),
    })
}

fn validate_require_options_owned(
    schema: &MetadataSchema,
    options: RequirePropOptions,
) -> Result<ValidatedRequireProp> {
    let RequirePropOptions { key, kind, value } = options;
    validate_require_parts(schema, &key, kind, value.as_ref())?;
    let expected_kind = registered_key_kind(schema, &key)?;

    Ok(ValidatedRequireProp {
        key,
        expected_kind,
        required_value: value,
    })
}

fn validate_require_parts(
    schema: &MetadataSchema,
    key: &str,
    kind: Option<MetadataKind>,
    value: Option<&MetadataValue>,
) -> Result<()> {
    let expected_kind = registered_key_kind(schema, key)?;

    if let Some(kind) = kind
        && kind != expected_kind
    {
        return Err(metadata_type_error(schema, key, expected_kind, kind));
    }
    if let Some(value) = value {
        if matches!(value, MetadataValue::None) {
            return Err(filter_option_error(
                "filter.invalid_prop",
                format!("filter '{FILTER_REQUIRE_PROP}' option 'value' cannot be none"),
            ));
        }
        schema.validate_value(key, value)?;
    }

    Ok(())
}

fn metadata_type_error(
    schema: &MetadataSchema,
    key: &str,
    expected: MetadataKind,
    actual: MetadataKind,
) -> PixelFlowError {
    PixelFlowError::new(
        metadata_namespace_category(schema, key),
        ErrorCode::new("metadata.type_mismatch"),
        format!(
            "metadata key '{key}' expects {:?}, got {:?}",
            expected, actual
        ),
    )
}

fn metadata_missing_error(schema: &MetadataSchema, key: &str) -> PixelFlowError {
    PixelFlowError::new(
        metadata_namespace_category(schema, key),
        ErrorCode::new("metadata.missing_value"),
        format!("metadata key '{key}' is missing required value"),
    )
}

fn metadata_value_mismatch_error(schema: &MetadataSchema, key: &str) -> PixelFlowError {
    PixelFlowError::new(
        metadata_namespace_category(schema, key),
        ErrorCode::new("metadata.value_mismatch"),
        format!("metadata key '{key}' does not match required value"),
    )
}

fn metadata_namespace_category(schema: &MetadataSchema, key: &str) -> ErrorCategory {
    if schema.is_core_key(key) {
        ErrorCategory::Core
    } else {
        ErrorCategory::Plugin
    }
}

#[cfg(test)]
mod tests {
    use pixelflow_core::{
        ClipMedia, ErrorCategory, ErrorCode, FilterOptionValue, FilterOptions, FilterPlanRequest,
        Frame, FrameCount, GraphBuilder, MetadataKind, MetadataSchema, MetadataValue, Rational,
    };

    use crate::testkit::{fixed_media, synthetic_u8_frame};

    use super::{
        ClearPropOptions, RequirePropOptions, SetPropOptions, add_clear_prop_filter,
        add_copy_props_filter, add_require_prop_filter, add_set_prop_filter,
        clear_prop_output_media, copy_props_output_media, metadata_value_from_option,
        require_prop_output_media, set_prop_output_media,
    };

    fn prop_schema() -> MetadataSchema {
        let mut schema = MetadataSchema::core();
        schema
            .register_plugin_key("acme/filter:enabled", MetadataKind::Bool)
            .expect("bool key should register");
        schema
            .register_plugin_key("acme/filter:payload", MetadataKind::Blob)
            .expect("blob key should register");
        schema
            .register_plugin_key("acme/filter:ratios", MetadataKind::Array)
            .expect("array key should register");
        schema
    }

    fn gray_frame() -> Frame {
        synthetic_u8_frame("gray8", 4, 2, |_plane, x, y| {
            u8::try_from(x + y * 10).expect("fixture sample fits u8")
        })
        .expect("gray frame should build")
    }

    #[test]
    fn metadata_value_from_option_converts_all_supported_variants() {
        let value = metadata_value_from_option(&pixelflow_core::FilterOptionValue::Array(vec![
            pixelflow_core::FilterOptionValue::None,
            pixelflow_core::FilterOptionValue::Bool(true),
            pixelflow_core::FilterOptionValue::Int(7),
            pixelflow_core::FilterOptionValue::Float(1.5),
            pixelflow_core::FilterOptionValue::String("x".to_owned()),
            pixelflow_core::FilterOptionValue::Rational(Rational {
                numerator: 1,
                denominator: 2,
            }),
            pixelflow_core::FilterOptionValue::Blob(vec![1_u8, 2, 3].into()),
        ]));

        assert_eq!(
            value,
            MetadataValue::Array(vec![
                MetadataValue::None,
                MetadataValue::Bool(true),
                MetadataValue::Int(7),
                MetadataValue::Float(1.5),
                MetadataValue::String("x".to_owned()),
                MetadataValue::Rational(Rational {
                    numerator: 1,
                    denominator: 2,
                }),
                MetadataValue::Blob(vec![1_u8, 2, 3].into()),
            ])
        );
    }

    #[test]
    fn require_prop_plan_and_executor_parse_same_options() {
        let input = fixed_media("gray8", 4, 2);
        let schema = prop_schema();
        let options = FilterOptions::from([
            (
                "key".to_owned(),
                FilterOptionValue::String("acme/filter:enabled".to_owned()),
            ),
            (
                "kind".to_owned(),
                FilterOptionValue::String("bool".to_owned()),
            ),
            ("value".to_owned(), FilterOptionValue::Bool(true)),
        ]);
        let request = FilterPlanRequest::new(std::slice::from_ref(&input), &options, &schema);

        let plan = super::plan_require_prop(request).expect("planner parses options");
        let executor = super::require_prop_executor_from_options(
            std::slice::from_ref(&input),
            &schema,
            &options,
        )
        .expect("executor parses same options");

        assert_eq!(plan.output_media(), &input);

        let frame = gray_frame();
        let mut metadata = frame.metadata().clone();
        metadata
            .set(&schema, "acme/filter:enabled", MetadataValue::Bool(true))
            .expect("bool metadata should set");
        let frame = frame.with_metadata(metadata);

        executor
            .apply_for_tests(frame)
            .expect("matching metadata should satisfy executor");
    }

    #[test]
    fn copy_props_plan_and_executor_parse_same_empty_options() {
        let prop_src = fixed_media("gray8", 2, 2);
        let target = fixed_media("yuv420p8", 8, 6);
        let schema = prop_schema();
        let options = FilterOptions::new();
        let inputs = [prop_src, target.clone()];
        let request = FilterPlanRequest::new(&inputs, &options, &schema);

        let plan = super::plan_copy_props(request).expect("copy_props planner should succeed");
        let executor = super::copy_props_executor_from_options(&inputs, &options)
            .expect("copy_props executor should parse the same options");

        assert_eq!(plan.output_media(), &target);
        assert_eq!(
            plan.compatibility(),
            super::COPY_PROPS_CONTRACT.compatibility()
        );
        assert_eq!(executor, super::CopyPropsExecutor::new());
    }

    #[test]
    fn copy_props_output_media_preserves_target_and_allows_different_prop_source_shape() {
        let prop_src = fixed_media("gray8", 2, 2);
        let target = fixed_media("yuv420p8", 8, 6);

        let output = copy_props_output_media(&[prop_src, target.clone()])
            .expect("copy_props should preserve target media");

        assert_eq!(output, target);
    }

    #[test]
    fn copy_props_output_media_requires_exactly_two_inputs() {
        let target = fixed_media("gray8", 8, 6);

        let one_input = copy_props_output_media(std::slice::from_ref(&target))
            .expect_err("copy_props requires metadata source and target inputs");
        assert_eq!(one_input.category(), ErrorCategory::Graph);
        assert_eq!(
            one_input.code(),
            ErrorCode::new("graph.invalid_filter_inputs")
        );

        let three_inputs = copy_props_output_media(&[target.clone(), target.clone(), target])
            .expect_err("copy_props accepts only metadata source and target inputs");
        assert_eq!(three_inputs.category(), ErrorCategory::Graph);
        assert_eq!(
            three_inputs.code(),
            ErrorCode::new("graph.invalid_filter_inputs")
        );
    }

    #[test]
    fn copy_props_output_media_rejects_frame_count_mismatch() {
        let target = fixed_media("gray8", 8, 6);
        let prop_src = ClipMedia::new(
            target.format().clone(),
            target.resolution().clone(),
            FrameCount::Finite(12),
            target.frame_rate(),
        );

        let error = copy_props_output_media(&[prop_src, target])
            .expect_err("metadata source must cover the target frame range");

        assert_eq!(error.category(), ErrorCategory::Format);
        assert_eq!(error.code(), ErrorCode::new("filter.incompatible_props"));
    }

    #[test]
    fn copy_props_rejects_options() {
        let input = fixed_media("gray8", 8, 6);
        let schema = prop_schema();
        let options =
            FilterOptions::from([("key".to_owned(), FilterOptionValue::String("x".to_owned()))]);
        let inputs = [input.clone(), input];
        let request = FilterPlanRequest::new(&inputs, &options, &schema);

        let error = super::plan_copy_props(request).expect_err("copy_props accepts no options");

        assert_eq!(error.category(), ErrorCategory::Format);
        assert_eq!(error.code(), ErrorCode::new("filter.invalid_copy_props"));
    }

    #[test]
    fn set_prop_output_media_preserves_media_and_validates_registered_key() {
        let input = fixed_media("gray8", 8, 6);
        let schema = prop_schema();

        let output = set_prop_output_media(
            &input,
            &schema,
            &SetPropOptions::new("acme/filter:enabled".to_owned(), MetadataValue::Bool(true)),
        )
        .expect("registered key should validate");

        assert_eq!(output, input);
    }

    #[test]
    fn set_prop_output_media_splits_type_mismatch_category_by_namespace() {
        let input = fixed_media("gray8", 8, 6);
        let schema = prop_schema();

        let core_error = set_prop_output_media(
            &input,
            &schema,
            &SetPropOptions::new(
                "core:frame_number".to_owned(),
                MetadataValue::String("bad".to_owned()),
            ),
        )
        .expect_err("core type mismatch should fail");
        assert_eq!(core_error.category(), ErrorCategory::Core);
        assert_eq!(core_error.code(), ErrorCode::new("metadata.type_mismatch"));

        let plugin_error = set_prop_output_media(
            &input,
            &schema,
            &SetPropOptions::new(
                "acme/filter:enabled".to_owned(),
                MetadataValue::String("bad".to_owned()),
            ),
        )
        .expect_err("plugin type mismatch should fail");
        assert_eq!(plugin_error.category(), ErrorCategory::Plugin);
        assert_eq!(
            plugin_error.code(),
            ErrorCode::new("metadata.type_mismatch")
        );
    }

    #[test]
    fn clear_prop_output_media_requires_registered_key() {
        let input = fixed_media("gray8", 8, 6);
        let schema = prop_schema();

        let output = clear_prop_output_media(
            &input,
            &schema,
            &ClearPropOptions::new("acme/filter:enabled".to_owned()),
        )
        .expect("registered key should clear");
        assert_eq!(output, input);

        let error = clear_prop_output_media(
            &input,
            &schema,
            &ClearPropOptions::new("acme/filter:missing".to_owned()),
        )
        .expect_err("unregistered key should fail");
        assert_eq!(error.category(), ErrorCategory::Plugin);
        assert_eq!(error.code(), ErrorCode::new("metadata.unregistered_key"));
    }

    #[test]
    fn require_prop_output_media_rejects_unknown_key_kind_and_required_value_type_mismatch() {
        let input = fixed_media("gray8", 8, 6);
        let schema = prop_schema();

        let unknown_key = require_prop_output_media(
            &input,
            &schema,
            &RequirePropOptions::new("acme/filter:missing".to_owned(), None, None),
        )
        .expect_err("unregistered key should fail");
        assert_eq!(unknown_key.category(), ErrorCategory::Plugin);
        assert_eq!(
            unknown_key.code(),
            ErrorCode::new("metadata.unregistered_key")
        );

        let kind_mismatch = require_prop_output_media(
            &input,
            &schema,
            &RequirePropOptions::new(
                "core:frame_number".to_owned(),
                Some(MetadataKind::Bool),
                None,
            ),
        )
        .expect_err("kind mismatch should fail");
        assert_eq!(kind_mismatch.category(), ErrorCategory::Core);
        assert_eq!(
            kind_mismatch.code(),
            ErrorCode::new("metadata.type_mismatch")
        );

        let value_mismatch = require_prop_output_media(
            &input,
            &schema,
            &RequirePropOptions::new(
                "acme/filter:enabled".to_owned(),
                None,
                Some(MetadataValue::String("bad".to_owned())),
            ),
        )
        .expect_err("required value with wrong type should fail");
        assert_eq!(value_mismatch.category(), ErrorCategory::Plugin);
        assert_eq!(
            value_mismatch.code(),
            ErrorCode::new("metadata.type_mismatch")
        );

        let none_value = require_prop_output_media(
            &input,
            &schema,
            &RequirePropOptions::new(
                "acme/filter:enabled".to_owned(),
                None,
                Some(MetadataValue::None),
            ),
        )
        .expect_err("required none value should fail during planning");
        assert_eq!(none_value.category(), ErrorCategory::Format);
        assert_eq!(none_value.code(), ErrorCode::new("filter.invalid_prop"));
    }

    #[test]
    fn add_prop_filters_create_filter_nodes_with_preserved_media() {
        let input_media = fixed_media("gray8", 8, 6);
        let schema = prop_schema();
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", input_media.clone());

        let (set_clip, _set_executor) = add_set_prop_filter(
            &mut builder,
            source,
            &input_media,
            &schema,
            SetPropOptions::new("acme/filter:enabled".to_owned(), MetadataValue::Bool(true)),
        )
        .expect("set_prop node should build");
        let (copy_clip, _copy_executor) = add_copy_props_filter(
            &mut builder,
            source,
            set_clip,
            &[input_media.clone(), input_media.clone()],
        )
        .expect("copy_props node should build");
        let (clear_clip, _clear_executor) = add_clear_prop_filter(
            &mut builder,
            copy_clip,
            &input_media,
            &schema,
            ClearPropOptions::new("acme/filter:enabled".to_owned()),
        )
        .expect("clear_prop node should build");
        let (require_clip, _require_executor) = add_require_prop_filter(
            &mut builder,
            clear_clip,
            &input_media,
            &schema,
            RequirePropOptions::new("acme/filter:enabled".to_owned(), None, None),
        )
        .expect("require_prop node should build");
        let graph = builder.build();

        for clip in [set_clip, copy_clip, clear_clip, require_clip] {
            let node = graph
                .node(clip.node_id())
                .expect("filter node should exist");
            assert_eq!(node.media(), &input_media);
        }
    }

    #[test]
    fn copy_props_executor_replaces_metadata_without_copying_target_planes() {
        let schema = prop_schema();
        let prop_src = gray_frame();
        let mut prop_metadata = prop_src.metadata().clone();
        prop_metadata
            .set(
                &schema,
                "core:range",
                MetadataValue::String("full".to_owned()),
            )
            .expect("source range metadata should set");
        prop_metadata
            .set(&schema, "core:frame_number", MetadataValue::Int(42))
            .expect("source frame number metadata should set");
        prop_metadata
            .set(&schema, "acme/filter:enabled", MetadataValue::Bool(true))
            .expect("plugin metadata should set");
        let prop_src = prop_src.with_metadata(prop_metadata);

        let target = gray_frame();
        let mut target_metadata = target.metadata().clone();
        target_metadata
            .set(
                &schema,
                "core:range",
                MetadataValue::String("limited".to_owned()),
            )
            .expect("target range metadata should set");
        let target = target.with_metadata(target_metadata);
        let executor = super::CopyPropsExecutor::new();

        let output = executor.apply_for_tests(&prop_src, &target);

        assert!(target.shares_plane_storage(&output, 0));
        assert_eq!(output.metadata(), prop_src.metadata());
        assert_eq!(
            output.metadata().get("core:range"),
            Some(&MetadataValue::String("full".to_owned()))
        );
        assert_eq!(
            output.metadata().get("acme/filter:enabled"),
            Some(&MetadataValue::Bool(true))
        );
    }

    #[test]
    fn set_prop_executor_updates_metadata_without_copying_planes() {
        let schema = prop_schema();
        let input = gray_frame();
        let executor = super::SetPropExecutor::new(
            schema,
            SetPropOptions::new("acme/filter:enabled".to_owned(), MetadataValue::Bool(true)),
        );

        let output = executor
            .apply_for_tests(&input)
            .expect("set_prop should succeed");

        assert!(input.shares_plane_storage(&output, 0));
        assert_eq!(
            output.metadata().get("acme/filter:enabled"),
            Some(&MetadataValue::Bool(true))
        );
    }

    #[test]
    fn clear_prop_executor_writes_none_without_copying_planes() {
        let schema = prop_schema();
        let mut metadata = gray_frame().metadata().clone();
        metadata
            .set(&schema, "acme/filter:enabled", MetadataValue::Bool(true))
            .expect("registered key should set");
        let input = gray_frame().with_metadata(metadata);
        let executor = super::ClearPropExecutor::new(
            schema,
            ClearPropOptions::new("acme/filter:enabled".to_owned()),
        );

        let output = executor
            .apply_for_tests(&input)
            .expect("clear_prop should succeed");

        assert!(input.shares_plane_storage(&output, 0));
        assert_eq!(
            output.metadata().get("acme/filter:enabled"),
            Some(&MetadataValue::None)
        );
    }

    #[test]
    fn require_prop_executor_returns_original_frame_on_success() {
        let schema = prop_schema();
        let mut metadata = gray_frame().metadata().clone();
        metadata
            .set(&schema, "acme/filter:enabled", MetadataValue::Bool(true))
            .expect("registered key should set");
        let input = gray_frame().with_metadata(metadata);
        let executor = super::RequirePropExecutor::new(
            schema,
            RequirePropOptions::new(
                "acme/filter:enabled".to_owned(),
                Some(MetadataKind::Bool),
                Some(MetadataValue::Bool(true)),
            ),
        )
        .expect("validated require_prop executor should build");

        let output = executor
            .apply_for_tests(input.clone())
            .expect("require_prop should succeed");

        assert_eq!(output.metadata(), input.metadata());
        assert!(input.shares_plane_storage(&output, 0));
    }

    #[test]
    fn require_prop_executor_reports_missing_type_and_value_mismatches_by_namespace() {
        let schema = prop_schema();
        let input = gray_frame();

        let missing_core = super::RequirePropExecutor::new(
            schema.clone(),
            RequirePropOptions::new("core:frame_number".to_owned(), None, None),
        )
        .expect("validated require_prop executor should build")
        .apply_for_tests(input.clone())
        .err()
        .expect("missing core value should fail");
        assert_eq!(missing_core.category(), ErrorCategory::Core);
        assert_eq!(
            missing_core.code(),
            ErrorCode::new("metadata.missing_value")
        );

        let wrong_type = super::RequirePropExecutor::new(
            schema.clone(),
            RequirePropOptions::new(
                "acme/filter:enabled".to_owned(),
                Some(MetadataKind::Int),
                None,
            ),
        )
        .expect_err("requested kind mismatch should fail at construction");
        assert_eq!(wrong_type.category(), ErrorCategory::Plugin);
        assert_eq!(wrong_type.code(), ErrorCode::new("metadata.type_mismatch"));

        let mut metadata = input.metadata().clone();
        metadata
            .set(&schema, "acme/filter:enabled", MetadataValue::Bool(true))
            .expect("registered key should set");
        let with_value = input.with_metadata(metadata);
        let value_mismatch = super::RequirePropExecutor::new(
            schema,
            RequirePropOptions::new(
                "acme/filter:enabled".to_owned(),
                Some(MetadataKind::Bool),
                Some(MetadataValue::Bool(false)),
            ),
        )
        .expect("validated require_prop executor should build")
        .apply_for_tests(with_value)
        .err()
        .expect("value mismatch should fail");
        assert_eq!(value_mismatch.category(), ErrorCategory::Plugin);
        assert_eq!(
            value_mismatch.code(),
            ErrorCode::new("metadata.value_mismatch")
        );
    }
}