dora_core/descriptor/

validate.rs

use crate::{
    adjust_shared_library_path,
    descriptor::{self, source_is_url},
    get_python_path,
};

use dora_message::{
    config::{Input, InputMapping, UserInputMapping, parse_byte_count},
    descriptor::{CoreNodeKind, DYNAMIC_SOURCE, OperatorSource, ResolvedNode, SHELL_SOURCE},
    id::{DataId, NodeId, OperatorId},
};
use eyre::{Context, bail, eyre};
use std::{
    collections::{BTreeMap, BTreeSet},
    path::Path,
    process::Command,
};
use tracing::info;

use super::{Descriptor, DescriptorExt, resolve_path};
const VERSION: &str = env!("CARGO_PKG_VERSION");

/// Validate input/output wiring without checking path existence.
///
/// This is safe to run before building nodes. It verifies every input
/// reference points to a declared output on the source node.
pub fn check_wiring(dataflow: &Descriptor) -> eyre::Result<()> {
    let nodes = dataflow.resolve_aliases_and_set_defaults()?;
    check_wiring_resolved(&nodes)
}

/// [`check_wiring`] on an already-resolved node map, so callers that have
/// resolved the descriptor can avoid re-resolving it.
fn check_wiring_resolved(nodes: &BTreeMap<NodeId, ResolvedNode>) -> eyre::Result<()> {
    for node in nodes.values() {
        match &node.kind {
            descriptor::CoreNodeKind::Custom(custom_node) => {
                for (input_id, input) in &custom_node.run_config.inputs {
                    check_input(input, nodes, &format!("{}/{input_id}", node.id))?;
                }
            }
            descriptor::CoreNodeKind::Runtime(runtime_node) => {
                for operator_definition in &runtime_node.operators {
                    for (input_id, input) in &operator_definition.config.inputs {
                        check_input(
                            input,
                            nodes,
                            &format!("{}/{}/{input_id}", node.id, operator_definition.id),
                        )?;
                    }
                }
            }
        };
    }

    Ok(())
}

/// Static (filesystem-independent) descriptor checks: ROS2 bridge configs,
/// timing fields, input/output wiring, and log-output configuration.
///
/// This is the subset of [`check_dataflow`] that `dora validate` can run
/// before any build has happened. [`check_dataflow`] runs it as its first
/// step and adds source-path existence and Python runtime checks on top.
pub fn check_dataflow_static(dataflow: &Descriptor) -> eyre::Result<()> {
    // validate ROS2 bridge configs before resolution
    validate_ros2_configs(dataflow)?;

    let nodes = dataflow.resolve_aliases_and_set_defaults()?;
    check_dataflow_static_resolved(dataflow, &nodes)
}

/// Validate all ROS2 bridge configs on the *unresolved* descriptor.
///
/// Run before [`Descriptor::resolve_aliases_and_set_defaults`] so a ROS2
/// misconfiguration is reported ahead of any alias-resolution error.
fn validate_ros2_configs(dataflow: &Descriptor) -> eyre::Result<()> {
    for node in &dataflow.nodes {
        if let Some(ros2) = &node.ros2 {
            validate_ros2_config(&node.id, ros2, &node.inputs, &node.outputs)?;
        }
    }
    Ok(())
}

/// The resolution-dependent part of [`check_dataflow_static`], operating on an
/// already-resolved node map so callers don't re-resolve the descriptor.
fn check_dataflow_static_resolved(
    dataflow: &Descriptor,
    nodes: &BTreeMap<NodeId, ResolvedNode>,
) -> eyre::Result<()> {
    for node in nodes.values() {
        check_node_timing(node)?;
    }
    // dataflow-level `health_check_interval` reaches `Duration::from_secs_f64`
    // in the same way (`binaries/daemon/src/lib.rs`). A zero interval must also
    // be rejected: it is fed to `tokio::time::interval`, which panics on a zero
    // period.
    check_seconds_field(
        "dataflow",
        "health_check_interval",
        dataflow.health_check_interval,
        false,
    )?;

    // check that all inputs mappings point to an existing output
    check_wiring_resolved(nodes)?;

    // Check that nodes can resolve `send_stdout_as`, `send_logs_as`, `min_log_level`
    for node in nodes.values() {
        node.send_stdout_as()
            .context("Could not resolve `send_stdout_as` configuration")?;
        node.send_logs_as()
            .context("Could not resolve `send_logs_as` configuration")?;
        node.min_log_level()
            .context("Could not resolve `min_log_level` configuration")?;
        node.max_log_size()
            .context("Could not resolve `max_log_size` configuration")?;
        node.max_rotated_files()
            .context("Could not resolve `max_rotated_files` configuration")?;
    }

    Ok(())
}

pub fn check_dataflow(dataflow: &Descriptor, working_dir: &Path) -> eyre::Result<()> {
    // Resolve the descriptor once and share the result across every check
    // (static validation + path/runtime existence) instead of re-resolving it
    // for each, which clones the whole node topology on every call.
    validate_ros2_configs(dataflow)?;
    let nodes = dataflow.resolve_aliases_and_set_defaults()?;
    check_dataflow_static_resolved(dataflow, &nodes)?;

    let mut has_python_operator = false;

    // check that nodes and operators exist
    for node in nodes.values() {
        match &node.kind {
            descriptor::CoreNodeKind::Custom(custom) => match &custom.source {
                dora_message::descriptor::NodeSource::Local => match custom.path.as_str() {
                    SHELL_SOURCE => (),
                    DYNAMIC_SOURCE => (),
                    source => {
                        if source_is_url(source) {
                            info!("{source} is a URL."); // TODO: Implement url check.
                        } else if custom.build.is_some() {
                            info!("skipping path check for node with build command");
                        } else {
                            resolve_path(source, working_dir).wrap_err_with(|| {
                                format!("Could not find source path `{source}`")
                            })?;
                        };
                    }
                },
                dora_message::descriptor::NodeSource::GitBranch { .. } => {
                    info!("skipping check for node with git source");
                }
            },
            descriptor::CoreNodeKind::Runtime(node) => {
                for operator_definition in &node.operators {
                    match &operator_definition.config.source {
                        OperatorSource::SharedLibrary(path) => {
                            if source_is_url(path) {
                                info!("{path} is a URL."); // TODO: Implement url check.
                            } else if operator_definition.config.build.is_some() {
                                info!("skipping path check for operator with build command");
                            } else {
                                let path = adjust_shared_library_path(Path::new(&path))?;
                                if !working_dir.join(&path).exists() {
                                    bail!("no shared library at `{}`", path.display());
                                }
                            }
                        }
                        OperatorSource::Python(python_source) => {
                            has_python_operator = true;
                            let path = &python_source.source;
                            if source_is_url(path) {
                                info!("{path} is a URL."); // TODO: Implement url check.
                            } else if operator_definition.config.build.is_some() {
                                info!("skipping path check for operator with build command");
                            } else if !working_dir.join(path).exists() {
                                bail!("no Python library at `{path}`");
                            }
                        }
                        OperatorSource::Wasm(path) => {
                            if source_is_url(path) {
                                info!("{path} is a URL."); // TODO: Implement url check.
                            } else if operator_definition.config.build.is_some() {
                                info!("skipping path check for operator with build command");
                            } else if !working_dir.join(path).exists() {
                                bail!("no WASM library at `{path}`");
                            }
                        }
                    }
                }
            }
        }
    }

    if has_python_operator {
        check_python_runtime()?;
    }

    Ok(())
}

/// Validate the second-valued timing fields of one resolved node: the custom
/// node's own timeouts and restart delays, plus every input's
/// `input_timeout`.
///
/// The daemon feeds these to `Duration::from_secs_f64`, which panics on
/// negative, non-finite, or overflowing values. [`check_dataflow_static`]
/// runs this for every node; call it directly for a node that joins a running
/// dataflow (`dora node add` / `dora node replace`), which never goes through
/// whole-dataflow validation.
///
/// This also enforces the `startup_timeout` rule for dynamic nodes, which is
/// a descriptor-policy check rather than a panic guard. The spawn path of an
/// existing dataflow (`dora start`) must not apply that rule to a descriptor
/// that started fine before, so it calls [`check_node_timing_values`] instead.
pub fn check_node_timing(node: &ResolvedNode) -> eyre::Result<()> {
    check_node_timing_values(node)?;
    if let descriptor::CoreNodeKind::Custom(custom) = &node.kind
        && custom.path.as_str() == DYNAMIC_SOURCE
        && custom.startup_timeout.is_some()
    {
        bail!(
            "dynamic node `{}` cannot specify `startup_timeout` (dynamic nodes connect out-of-band and are not managed by the startup watchdog)",
            node.id
        );
    }
    Ok(())
}

/// The panic-guard half of [`check_node_timing`]: reject only the values the
/// daemon would panic on in `Duration::from_secs_f64`, with no
/// descriptor-policy rules on top.
///
/// `dora start` never ran whole-dataflow validation before, so applying a
/// policy rule there would reject descriptors that start fine on 1.0.x and
/// never panicked -- a dynamic node carrying an (ignored) `startup_timeout`,
/// for one. The spawn path therefore guards against the panic only, and
/// leaves the policy rules to `dora run` / `dora validate`
/// ([`check_dataflow_static`]) and to the node-join path
/// ([`check_node_timing`]).
pub fn check_node_timing_values(node: &ResolvedNode) -> eyre::Result<()> {
    if let descriptor::CoreNodeKind::Custom(custom) = &node.kind {
        check_timing_fields(&node.id, custom)?;
    }
    // `input_timeout` is a second-valued `f64` that the daemon also feeds
    // to `Duration::from_secs_f64`, on both the initial-spawn and the
    // reconnect paths.
    for (input_id, input) in node_inputs(node) {
        check_seconds_field(
            &format!("input `{input_id}` of node `{}`", node.id),
            "input_timeout",
            input.input_timeout,
            true,
        )?;
    }
    Ok(())
}

/// Reject negative / non-finite / overflowing second-valued timing fields on a
/// custom node.
///
/// The daemon converts these to `Duration` via `Duration::from_secs_f64`, which
/// panics on negative, NaN, infinite, or overflowing (> `Duration::MAX`) input.
/// Catching them here surfaces a clean descriptor error instead of crashing the
/// daemon on node spawn.
fn check_timing_fields(
    node_id: &NodeId,
    custom: &dora_message::descriptor::CustomNode,
) -> eyre::Result<()> {
    let owner = format!("node `{node_id}`");
    for (field, value) in [
        ("finish_grace_secs", custom.finish_grace_secs),
        ("health_check_timeout", custom.health_check_timeout),
        ("startup_timeout", custom.startup_timeout),
        ("restart_delay", custom.restart_delay),
        ("max_restart_delay", custom.max_restart_delay),
        ("restart_window", custom.restart_window),
    ] {
        check_seconds_field(&owner, field, value, true)?;
    }
    Ok(())
}

/// Reject a negative / non-finite / overflowing second-valued field before it
/// reaches the daemon, where `Duration::from_secs_f64` would panic on such
/// input.
///
/// `Duration::from_secs_f64` panics on negative, NaN, infinite, *and* finite
/// values too large to fit in a `Duration` (> `Duration::MAX`, ≈ 1.8e19 s). We
/// probe the exact same boundary with its non-panicking twin
/// `try_from_secs_f64`, so a value accepted here can never panic the daemon.
///
/// When `allow_zero` is `false`, any value that produces a zero-length
/// `Duration` is also rejected. This is required for fields that reach
/// `tokio::time::interval` (e.g. `health_check_interval`), which panics on a
/// zero period. Checking the resulting `Duration` -- not just the literal
/// `0.0` -- also rejects a tiny-but-positive value such as `1e-10`, which
/// `Duration::from_secs_f64` rounds down to `Duration::ZERO`. This mirrors the
/// timer parser's `interval.is_zero()` guard in `dora-message`.
fn check_seconds_field(
    owner: &str,
    field: &str,
    value: Option<f64>,
    allow_zero: bool,
) -> eyre::Result<()> {
    if let Some(value) = value {
        // A negative / non-finite / overflowing value fails to convert; a
        // tiny-but-positive value (e.g. `1e-10`) converts to `Duration::ZERO`,
        // which must also be rejected for interval fields (`allow_zero ==
        // false`). Inspect the resulting `Duration`, not the literal `0.0`.
        let duration = std::time::Duration::try_from_secs_f64(value);
        let is_zero = duration.as_ref().is_ok_and(|d| d.is_zero());
        if !allow_zero && is_zero {
            // Distinct message: a value like `1e-10` *is* positive, so calling it
            // "not positive" would misdirect the user -- the real reason is that
            // it rounds down to a zero-length duration.
            bail!(
                "{owner} has invalid `{field}`: {value} \
                 (must be a positive number of seconds; this value is zero or \
                 rounds down to a zero-length duration)"
            );
        }
        if duration.is_err() {
            let requirement = if allow_zero {
                "non-negative"
            } else {
                "positive"
            };
            bail!(
                "{owner} has invalid `{field}`: {value} \
                 (must be a finite, {requirement} number of seconds smaller than {})",
                std::time::Duration::MAX.as_secs_f64()
            );
        }
    }
    Ok(())
}

/// All `(input_id, input)` pairs declared on a resolved node: a custom node's
/// own inputs, or the merged inputs of every operator of a runtime node.
fn node_inputs(node: &ResolvedNode) -> Vec<(&DataId, &Input)> {
    match &node.kind {
        CoreNodeKind::Custom(custom) => custom.run_config.inputs.iter().collect(),
        CoreNodeKind::Runtime(runtime) => runtime
            .operators
            .iter()
            .flat_map(|op| op.config.inputs.iter())
            .collect(),
    }
}

pub trait ResolvedNodeExt {
    fn send_stdout_as(&self) -> eyre::Result<Option<String>>;
    fn send_logs_as(&self) -> eyre::Result<Option<String>>;
    fn min_log_level(&self) -> eyre::Result<Option<dora_message::common::LogLevelOrStdout>>;
    fn max_log_size(&self) -> eyre::Result<Option<u64>>;
    fn max_rotated_files(&self) -> eyre::Result<Option<u32>>;
}

/// Check that a resolved `send_stdout_as` / `send_logs_as` name is a valid
/// [`DataId`]. The daemon's per-node log task turns it into one for every log
/// line, and an invalid name (empty, a space, `op/` from an empty operator
/// entry, ...) would otherwise panic that task on the node's first line and
/// silently drop all of the node's later logs.
fn checked_output_name(field: &str, name: String) -> eyre::Result<String> {
    name.parse::<DataId>()
        .map_err(|err| eyre!("`{field}: {name}` is not a valid output id: {err}"))?;
    Ok(name)
}

impl ResolvedNodeExt for ResolvedNode {
    fn send_stdout_as(&self) -> eyre::Result<Option<String>> {
        let name = match &self.kind {
            // TODO: Split stdout between operators
            CoreNodeKind::Runtime(n) => {
                let count = n
                    .operators
                    .iter()
                    .filter(|op| op.config.send_stdout_as.is_some())
                    .count();
                if count == 1 && n.operators.len() > 1 {
                    tracing::warn!(
                        "All stdout from all operators of a runtime are going to be sent in the selected `send_stdout_as` operator."
                    )
                } else if count > 1 {
                    return Err(eyre!(
                        "More than one `send_stdout_as` entries for a runtime node. Please only use one `send_stdout_as` per runtime."
                    ));
                }
                n.operators.iter().find_map(|op| {
                    op.config
                        .send_stdout_as
                        .as_ref()
                        .map(|stdout| format!("{}/{}", op.id, stdout))
                })
            }
            CoreNodeKind::Custom(n) => n.send_stdout_as.clone(),
        };
        name.map(|name| checked_output_name("send_stdout_as", name))
            .transpose()
    }

    fn send_logs_as(&self) -> eyre::Result<Option<String>> {
        let name = match &self.kind {
            CoreNodeKind::Runtime(n) => {
                let count = n
                    .operators
                    .iter()
                    .filter(|op| op.config.send_logs_as.is_some())
                    .count();
                if count > 1 {
                    return Err(eyre!(
                        "More than one `send_logs_as` entries for a runtime node. Please only use one `send_logs_as` per runtime."
                    ));
                }
                n.operators.iter().find_map(|op| {
                    op.config
                        .send_logs_as
                        .as_ref()
                        .map(|logs| format!("{}/{}", op.id, logs))
                })
            }
            CoreNodeKind::Custom(n) => n.send_logs_as.clone(),
        };
        name.map(|name| checked_output_name("send_logs_as", name))
            .transpose()
    }

    fn min_log_level(&self) -> eyre::Result<Option<dora_message::common::LogLevelOrStdout>> {
        let level_str = match &self.kind {
            CoreNodeKind::Runtime(n) => {
                // Use the first operator's min_log_level (only one allowed)
                let levels: Vec<_> = n
                    .operators
                    .iter()
                    .filter_map(|op| op.config.min_log_level.as_deref())
                    .collect();
                if levels.len() > 1 {
                    return Err(eyre!(
                        "More than one `min_log_level` entries for a runtime node. Please only use one `min_log_level` per runtime."
                    ));
                }
                levels.first().map(|s| s.to_string())
            }
            CoreNodeKind::Custom(n) => n.min_log_level.clone(),
        };
        match level_str {
            None => Ok(None),
            Some(s) => {
                let level = parse_log_level(&s)?;
                Ok(Some(level))
            }
        }
    }

    fn max_log_size(&self) -> eyre::Result<Option<u64>> {
        let size_str = match &self.kind {
            CoreNodeKind::Runtime(n) => {
                let sizes: Vec<_> = n
                    .operators
                    .iter()
                    .filter_map(|op| op.config.max_log_size.as_deref())
                    .collect();
                if sizes.len() > 1 {
                    return Err(eyre!(
                        "More than one `max_log_size` entries for a runtime node. Please only use one `max_log_size` per runtime."
                    ));
                }
                sizes.first().map(|s| s.to_string())
            }
            CoreNodeKind::Custom(n) => n.max_log_size.clone(),
        };
        match size_str {
            None => Ok(None),
            Some(s) => {
                let bytes = parse_byte_size(&s)?;
                Ok(Some(bytes))
            }
        }
    }

    fn max_rotated_files(&self) -> eyre::Result<Option<u32>> {
        let value = match &self.kind {
            CoreNodeKind::Runtime(n) => {
                let values: Vec<_> = n
                    .operators
                    .iter()
                    .filter_map(|op| op.config.max_rotated_files)
                    .collect();
                if values.len() > 1 {
                    return Err(eyre!(
                        "More than one `max_rotated_files` entries for a runtime node. Please only use one `max_rotated_files` per runtime."
                    ));
                }
                values.first().copied()
            }
            CoreNodeKind::Custom(n) => n.max_rotated_files,
        };
        if let Some(n) = value {
            // 0 is meaningful: keep the active log only, rotating the previous
            // one away rather than retaining it. That matches the documented
            // disk bound `max_log_size * (1 + max_rotated_files)`, which at 0
            // is one active file.
            if n > 100 {
                bail!("`max_rotated_files` must not exceed 100");
            }
        }
        Ok(value)
    }
}

fn parse_byte_size(s: &str) -> eyre::Result<u64> {
    // Share `dora_message::config::ByteSize`'s grammar, but parse in `u64`:
    // `ByteSize` is `usize`-based, which would reject log sizes of 4 GiB and
    // above on 32-bit targets (#3604).
    parse_byte_count(s).map_err(|e| eyre!("{e}"))
}

fn parse_log_level(s: &str) -> eyre::Result<dora_message::common::LogLevelOrStdout> {
    match s.to_lowercase().as_str() {
        "error" => Ok(dora_message::common::LogLevelOrStdout::LogLevel(
            log::Level::Error,
        )),
        "warn" => Ok(dora_message::common::LogLevelOrStdout::LogLevel(
            log::Level::Warn,
        )),
        "info" => Ok(dora_message::common::LogLevelOrStdout::LogLevel(
            log::Level::Info,
        )),
        "debug" => Ok(dora_message::common::LogLevelOrStdout::LogLevel(
            log::Level::Debug,
        )),
        "trace" => Ok(dora_message::common::LogLevelOrStdout::LogLevel(
            log::Level::Trace,
        )),
        "stdout" => Ok(dora_message::common::LogLevelOrStdout::Stdout),
        _ => bail!(
            "invalid min_log_level: '{s}', expected one of: error, warn, info, debug, trace, stdout"
        ),
    }
}

fn check_input(
    input: &Input,
    nodes: &BTreeMap<NodeId, super::ResolvedNode>,
    input_id_str: &str,
) -> Result<(), eyre::ErrReport> {
    match &input.mapping {
        InputMapping::Timer { interval: _ } | InputMapping::Logs(_) => {}
        InputMapping::User(UserInputMapping { source, output }) => {
            // `nodes` is keyed by each node's own id (see
            // `resolve_aliases_and_set_defaults`), so look the source up
            // directly instead of scanning every node — this runs once per
            // input of every node, so a linear scan makes wiring validation
            // quadratic in the node count.
            let source_node = nodes.get(source).ok_or_else(|| {
                eyre!("source node `{source}` mapped to input `{input_id_str}` does not exist",)
            })?;
            match &source_node.kind {
                CoreNodeKind::Custom(custom_node) => {
                    if !custom_node.run_config.outputs.contains(output) {
                        bail!(
                            "output `{source}/{output}` mapped to \
                            input `{input_id_str}` does not exist",
                        );
                    }
                }
                CoreNodeKind::Runtime(runtime) => {
                    // Outputs of runtime nodes are referenced as
                    // `<node>/<operator>/<output>`. Without the operator
                    // segment, falling back to an empty operator id would
                    // produce a misleading "operator `` does not exist" error.
                    let Some((operator_id, output)) = output.split_once('/') else {
                        bail!(
                            "input `{input_id_str}` references output `{output}` of node \
                            `{source}`, which is a runtime node; runtime node outputs \
                            must include the operator id \
                            (expected format: `{source}/<operator_id>/<output_id>`)"
                        );
                    };
                    let operator_id = OperatorId::from(operator_id.to_owned());
                    let output = DataId::from(output.to_owned());

                    let operator = runtime
                        .operators
                        .iter()
                        .find(|o| o.id == operator_id)
                        .ok_or_else(|| {
                            eyre!(
                                "source operator `{source}/{operator_id}` used \
                                for input `{input_id_str}` does not exist",
                            )
                        })?;

                    if !operator.config.outputs.contains(&output) {
                        bail!(
                            "output `{source}/{operator_id}/{output}` mapped to \
                            input `{input_id_str}` does not exist",
                        );
                    }
                }
            }
        }
    };
    Ok(())
}

fn check_python_runtime() -> eyre::Result<()> {
    // Check if python dora-rs is installed and match cli version
    let reinstall_command =
        format!("Please reinstall it with: `pip install dora-rs=={VERSION} --force`");
    let mut command = Command::new(get_python_path().context("Could not get python binary")?);
    command.args([
        "-c",
        &format!(
            "
import dora;
assert dora.__version__=='{VERSION}',  'Python dora-rs should be {VERSION}, but current version is %s. {reinstall_command}' % (dora.__version__)
        "
        ),
    ]);
    let mut result = command
        .spawn()
        .wrap_err("Could not spawn python dora-rs command.")?;
    let status = result
        .wait()
        .wrap_err("Could not get exit status when checking python dora-rs")?;

    if !status.success() {
        bail!("Something went wrong with Python dora-rs. {reinstall_command}")
    }

    Ok(())
}

fn validate_ros2_config(
    node_id: &NodeId,
    config: &dora_message::descriptor::Ros2BridgeConfig,
    node_inputs: &BTreeMap<DataId, Input>,
    node_outputs: &BTreeSet<DataId>,
) -> eyre::Result<()> {
    use dora_message::descriptor::{Ros2Role, Ros2TransportConfig};

    if let Ros2TransportConfig::Zenoh {
        config_uri: Some(uri),
        ..
    } = &config.transport
        && uri.as_os_str().is_empty()
    {
        bail!("node `{node_id}`: ros2 Zenoh config_uri must not be empty");
    }

    // Exactly one of topic, topics, service, action must be set
    let mode_count = [
        config.topic.is_some(),
        config.topics.is_some(),
        config.service.is_some(),
        config.action.is_some(),
    ]
    .iter()
    .filter(|&&v| v)
    .count();
    if mode_count == 0 {
        bail!(
            "node `{node_id}`: ros2 config requires one of \
             `topic`, `topics`, `service`, or `action`"
        );
    }
    if mode_count > 1 {
        bail!(
            "node `{node_id}`: ros2 config has multiple of \
             `topic`, `topics`, `service`, `action` - only one is allowed"
        );
    }

    if let Some(topic) = &config.topic {
        validate_ros2_name(node_id, "topic", topic)?;
        let message_type = config.message_type.as_ref().ok_or_else(|| {
            eyre!("node `{node_id}`: ros2 config with `topic` requires `message_type`")
        })?;
        validate_ros2_type_format(node_id, topic, message_type)?;

        // Single-topic mode has no explicit `output:`/`input:` field, so the
        // bridge binds to the node's declared port. Resolution is what picks
        // that port and what rejects a node whose declarations leave the choice
        // ambiguous, so call it here rather than restating the rule: the two
        // must not be able to disagree.
        super::resolve_ros2_single_topic(node_id, config, node_inputs, node_outputs)?;
    } else if let Some(topics) = &config.topics {
        if topics.is_empty() {
            bail!("node `{node_id}`: ros2 `topics` list must not be empty");
        }
        if topics.len() > 64 {
            bail!(
                "node `{node_id}`: ros2 `topics` list has {} entries, maximum is 64",
                topics.len()
            );
        }
        for t in topics {
            validate_ros2_name(node_id, "topic", &t.topic)?;
            validate_ros2_type_format(node_id, &t.topic, &t.message_type)?;
        }
        // The per-topic port-existence check lives on the resolution path
        // (`validate_ros2_topic_ports`) so the coordinator's `dora start` path,
        // which resolves without validating, enforces it too. Delegate here
        // rather than restating the rule, keeping validation and `dora start` in
        // lockstep by construction (dora-rs/dora#3484).
        super::validate_ros2_topic_ports(node_id, topics, node_inputs, node_outputs)?;
    } else if let Some(service) = &config.service {
        validate_ros2_name(node_id, "service", service)?;
        let service_type = config.service_type.as_ref().ok_or_else(|| {
            eyre!("node `{node_id}`: ros2 config with `service` requires `service_type`")
        })?;
        validate_ros2_type_format(node_id, service, service_type)?;
        let role = config.role.as_ref().ok_or_else(|| {
            eyre!("node `{node_id}`: ros2 service bridge requires `role` (client or server)")
        })?;
        match role {
            Ros2Role::Client => {
                if node_inputs.is_empty() {
                    bail!(
                        "node `{node_id}`: ros2 service client requires at least one input (request)"
                    );
                }
                if node_outputs.is_empty() {
                    bail!(
                        "node `{node_id}`: ros2 service client requires at least one output (response)"
                    );
                }
            }
            Ros2Role::Server => {
                if node_inputs.is_empty() {
                    bail!(
                        "node `{node_id}`: ros2 service server requires at least one input (response)"
                    );
                }
                if node_outputs.is_empty() {
                    bail!(
                        "node `{node_id}`: ros2 service server requires at least one output (request)"
                    );
                }
            }
        }
    } else if let Some(action) = &config.action {
        validate_ros2_name(node_id, "action", action)?;
        let action_type = config.action_type.as_ref().ok_or_else(|| {
            eyre!("node `{node_id}`: ros2 config with `action` requires `action_type`")
        })?;
        validate_ros2_type_format(node_id, action, action_type)?;
        let role = config
            .role
            .as_ref()
            .ok_or_else(|| eyre!("node `{node_id}`: ros2 action bridge requires `role`"))?;
        match role {
            Ros2Role::Client => {
                if node_inputs.is_empty() {
                    bail!(
                        "node `{node_id}`: ros2 action client requires at least one input (goal)"
                    );
                }
                if node_outputs.is_empty() {
                    bail!(
                        "node `{node_id}`: ros2 action client requires at least one output \
                         (feedback/result)"
                    );
                }
            }
            Ros2Role::Server => {
                if node_inputs.is_empty() {
                    bail!(
                        "node `{node_id}`: ros2 action server requires at least one input \
                         (feedback/result)"
                    );
                }
                if node_outputs.is_empty() {
                    bail!(
                        "node `{node_id}`: ros2 action server requires at least one output (goal)"
                    );
                }
            }
        }
    }

    // Validate QoS config
    validate_ros2_qos(node_id, &config.qos)?;
    if let Some(topics) = &config.topics {
        for t in topics {
            if let Some(qos) = &t.qos {
                validate_ros2_qos(node_id, qos)?;
            }
        }
    }

    Ok(())
}

fn validate_ros2_qos(
    node_id: &NodeId,
    qos: &dora_message::descriptor::Ros2QosConfig,
) -> eyre::Result<()> {
    if let Some(d) = &qos.durability {
        match d.as_str() {
            "volatile" | "transient_local" => {}
            _ => bail!(
                "node `{node_id}`: invalid QoS durability `{d}`, \
                 expected \"volatile\" or \"transient_local\""
            ),
        }
    }
    if let Some(l) = &qos.liveliness {
        match l.as_str() {
            "automatic" | "manual_by_participant" | "manual_by_topic" => {}
            _ => bail!(
                "node `{node_id}`: invalid QoS liveliness `{l}`, \
                 expected \"automatic\", \"manual_by_participant\", or \"manual_by_topic\""
            ),
        }
    }
    if let Some(depth) = qos.keep_last
        && !(1..=10_000).contains(&depth)
    {
        bail!(
            "node `{node_id}`: QoS keep_last depth {depth} out of range, \
                 must be between 1 and 10000"
        );
    }
    // Both fields reach `Duration::from_secs_f64` in the ROS2 bridge QoS
    // conversion and panic on a finite-but-overflowing value (e.g. `1e300`),
    // not just on NaN/infinite/negative. `check_seconds_field` probes that
    // exact boundary with `try_from_secs_f64`, matching every other
    // second-valued field here, and keeps validating both fields whenever set
    // (as the prior check did). `allow_zero = true`: `max_blocking_time`
    // defaults to `0.0` and a zero lease is harmless.
    let owner = format!("node `{node_id}`");
    check_seconds_field(&owner, "QoS max_blocking_time", qos.max_blocking_time, true)?;
    check_seconds_field(&owner, "QoS lease_duration", qos.lease_duration, true)?;
    Ok(())
}

/// Validate a ROS2 name (topic, service, or action) follows ROS2 naming rules.
///
/// ROS2 names allow ASCII alphanumeric, underscore, and forward slash characters,
/// and must additionally be structurally valid: no consecutive slashes, no trailing
/// slash, not a bare `/`, and each `/`-delimited token must not start with a digit.
/// Tokens may begin with an underscore (hidden topic/service convention in ROS2).
/// See <https://design.ros2.org/articles/topic_and_service_names.html>.
fn validate_ros2_name(node_id: &NodeId, field: &str, name: &str) -> eyre::Result<()> {
    if name.is_empty() {
        bail!("node `{node_id}`: `{field}` must not be empty");
    }
    if !name
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '/')
    {
        bail!(
            "node `{node_id}`: invalid `{field}` name `{name}`, \
             only ASCII alphanumeric, underscore, and '/' characters allowed"
        );
    }
    if name == "/" {
        bail!("node `{node_id}`: invalid `{field}` name, must not be a bare '/'");
    }
    if name.contains("//") {
        bail!(
            "node `{node_id}`: invalid `{field}` name `{name}`, \
             consecutive slashes ('//') are not allowed"
        );
    }
    if name.ends_with('/') {
        bail!(
            "node `{node_id}`: invalid `{field}` name `{name}`, \
             name must not end with '/'"
        );
    }
    // Each '/'-delimited token must not start with a digit. A leading '/'
    // (absolute name) produces an empty first token, which is allowed and skipped.
    // Leading underscores are valid — they denote hidden topics/services in ROS2.
    for (i, token) in name.split('/').enumerate() {
        if i == 0 && token.is_empty() {
            continue;
        }
        if token.starts_with(|c: char| c.is_ascii_digit()) {
            bail!(
                "node `{node_id}`: invalid `{field}` name `{name}`, \
                 a token must not start with a digit (offending token `{token}`)"
            );
        }
    }
    Ok(())
}

/// A non-fatal warning about type annotations in the dataflow.
#[derive(Debug)]
pub struct TypeWarning {
    /// Node that produced the warning.
    pub node_id: String,
    /// Human-readable warning message.
    pub message: String,
}

impl std::fmt::Display for TypeWarning {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "node \"{}\": {}", self.node_id, self.message)
    }
}

/// A type inferred from an upstream annotated output (Phase 3).
#[derive(Debug)]
pub struct TypeInference {
    /// Node that received the inferred type.
    pub node_id: String,
    /// Port that received the inferred type.
    pub port_id: String,
    /// The inferred type URN.
    pub inferred_urn: String,
    /// Source of the inference (e.g. "sensor/reading").
    pub source: String,
}

impl std::fmt::Display for TypeInference {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "inferred {} on {}/{} (from {})",
            self.inferred_urn, self.node_id, self.port_id, self.source
        )
    }
}

/// Result of type checking a dataflow.
pub struct TypeCheckResult {
    /// Type warnings found.
    pub warnings: Vec<TypeWarning>,
    /// Types inferred from upstream annotations (Phase 3).
    pub inferences: Vec<TypeInference>,
}

/// Timer nodes auto-inject this type.
const TIMER_TYPE: &str = "std/core/v1/UInt64";

/// Register one declared output type under `key` (a `(node_id, output_ref)`
/// pair). The port is *always* recorded in `annotated` so a typo'd or
/// otherwise unresolvable URN still suppresses the strict-mode "upstream has
/// no type annotation" warning; it is additionally recorded in
/// `output_type_map` (the map used for cross-edge type matching) only when its
/// URN resolves. Keeping both writes behind one call makes the two structures
/// impossible to desync — the invariant the strict-mode check relies on.
fn register_output_type(
    output_type_map: &mut BTreeMap<(String, String), String>,
    annotated: &mut BTreeSet<(String, String)>,
    key: (String, String),
    urn: &str,
    registry: &crate::types::TypeRegistry,
) {
    annotated.insert(key.clone());
    if registry.resolve(urn).is_some() {
        output_type_map.insert(key, urn.to_string());
    }
}

/// Check type annotations in a dataflow, with strict mode support.
///
/// Returns the collected warnings plus any inferred edge types.
///
/// Checks:
/// 1. `output_types` keys exist in `outputs`
/// 2. `input_types` keys exist in `inputs`
/// 3. All type URNs resolve in the registry
/// 4. Connected edges have compatible types (using compatibility graph)
/// 5. Type inference across edges (Phase 3)
/// 6. Metadata annotation validation (Phase 5)
/// 7. Arrow schema validation (Phase 6)
/// 8. Strict mode: warn on unannotated ports connected to annotated ports
pub fn check_type_annotations_full(
    dataflow: &super::Descriptor,
    registry: &crate::types::TypeRegistry,
    strict: bool,
) -> TypeCheckResult {
    use crate::types::{CompatibilityGraph, TypeRule};

    let mut warnings = Vec::new();
    let mut inferences = Vec::new();

    // Build compatibility graph from user rules
    let user_rules: Vec<TypeRule> = dataflow
        .type_rules
        .iter()
        .map(|r| TypeRule {
            from: r.from.clone(),
            to: r.to.clone(),
        })
        .collect();
    let compat = CompatibilityGraph::new(&user_rules);

    // Map of (source_node_id, output_ref) -> type_urn for cross-edge checking.
    // Only outputs whose declared URN *resolves* are registered here.
    let mut output_type_map: BTreeMap<(String, String), String> = BTreeMap::new();

    // Every (source_node_id, output_ref) that carries *some* output-type
    // annotation, including one whose URN does not resolve (a typo). Used to
    // suppress the strict-mode "upstream has no type annotation" warning on a
    // port that *is* annotated but with an unknown URN — that URN is already
    // reported by `check_port_types`, so the extra "no annotation" warning
    // would be factually wrong and misdirect the user.
    let mut annotated_output_ports: BTreeSet<(String, String)> = BTreeSet::new();

    for node in &dataflow.nodes {
        let nid = node.id.to_string();

        // Check node-level output_types
        check_port_types(
            &nid,
            &node.output_types,
            |id| node.outputs.contains(id),
            "output",
            registry,
            &mut warnings,
        );
        // Register node outputs for cross-edge checking
        for (output_id, urn) in &node.output_types {
            register_output_type(
                &mut output_type_map,
                &mut annotated_output_ports,
                (nid.clone(), output_id.to_string()),
                urn,
                registry,
            );
        }

        // Check node-level input_types
        check_port_types(
            &nid,
            &node.input_types,
            |id| node.inputs.contains_key(id),
            "input",
            registry,
            &mut warnings,
        );

        // Check metadata annotations (Phase 5)
        check_metadata_annotations(
            &nid,
            &node.output_metadata,
            &node.pattern,
            &node.outputs,
            &mut warnings,
        );

        // Check operator output/input types
        if let Some(op) = &node.operator {
            let op_id = op
                .id
                .as_ref()
                .map(|id| id.to_string())
                .unwrap_or_else(|| super::SINGLE_OPERATOR_DEFAULT_ID.to_string());
            check_port_types(
                &nid,
                &op.config.output_types,
                |id| op.config.outputs.contains(id),
                "output",
                registry,
                &mut warnings,
            );
            for (output_id, urn) in &op.config.output_types {
                // A consumer of a single-`operator:` node references its output
                // in short form (`node/output`) in the source YAML; the `op/`
                // prefix is only injected later by
                // `resolve_aliases_and_set_defaults`, which this pre-build check
                // never runs. Register both the prefixed key and the bare
                // `output_id` so short-form edges resolve here too — otherwise
                // the upstream type is invisible, silently dropping genuine
                // mismatches and, under `strict`, emitting a false "upstream has
                // no type annotation" warning.
                register_output_type(
                    &mut output_type_map,
                    &mut annotated_output_ports,
                    (nid.clone(), format!("{op_id}/{output_id}")),
                    urn,
                    registry,
                );
                register_output_type(
                    &mut output_type_map,
                    &mut annotated_output_ports,
                    (nid.clone(), output_id.to_string()),
                    urn,
                    registry,
                );
            }
            check_port_types(
                &nid,
                &op.config.input_types,
                |id| op.config.inputs.contains_key(id),
                "input",
                registry,
                &mut warnings,
            );
            check_metadata_annotations(
                &nid,
                &op.config.output_metadata,
                &op.config.pattern,
                &op.config.outputs,
                &mut warnings,
            );
        }
        if let Some(runtime) = &node.operators {
            for op in &runtime.operators {
                let label = format!("{nid}/{}", op.id);
                check_port_types(
                    &label,
                    &op.config.output_types,
                    |id| op.config.outputs.contains(id),
                    "output",
                    registry,
                    &mut warnings,
                );
                for (output_id, urn) in &op.config.output_types {
                    register_output_type(
                        &mut output_type_map,
                        &mut annotated_output_ports,
                        (nid.clone(), format!("{}/{output_id}", op.id)),
                        urn,
                        registry,
                    );
                }
                check_port_types(
                    &label,
                    &op.config.input_types,
                    |id| op.config.inputs.contains_key(id),
                    "input",
                    registry,
                    &mut warnings,
                );
                check_metadata_annotations(
                    &label,
                    &op.config.output_metadata,
                    &op.config.pattern,
                    &op.config.outputs,
                    &mut warnings,
                );
            }
        }
    }

    // Cross-edge type checking + inference (Phase 3 + 4)
    // Also check timer inputs against input_types annotations.
    for node in &dataflow.nodes {
        let nid = node.id.to_string();
        let timer_types = timer_input_types(&node.inputs);
        check_edge_mismatches_with_compat(
            &nid,
            &node.input_types,
            &node.inputs,
            &output_type_map,
            &annotated_output_ports,
            &timer_types,
            &compat,
            registry,
            strict,
            &mut warnings,
            &mut inferences,
        );

        if let Some(op) = &node.operator {
            let op_timer = timer_input_types(&op.config.inputs);
            check_edge_mismatches_with_compat(
                &nid,
                &op.config.input_types,
                &op.config.inputs,
                &output_type_map,
                &annotated_output_ports,
                &op_timer,
                &compat,
                registry,
                strict,
                &mut warnings,
                &mut inferences,
            );
        }
        if let Some(runtime) = &node.operators {
            for op in &runtime.operators {
                let label = format!("{nid}/{}", op.id);
                let op_timer = timer_input_types(&op.config.inputs);
                check_edge_mismatches_with_compat(
                    &label,
                    &op.config.input_types,
                    &op.config.inputs,
                    &output_type_map,
                    &annotated_output_ports,
                    &op_timer,
                    &compat,
                    registry,
                    strict,
                    &mut warnings,
                    &mut inferences,
                );
            }
        }
    }

    TypeCheckResult {
        warnings,
        inferences,
    }
}

/// Build a map of (input_id -> timer_type) for Timer-mapped inputs.
fn timer_input_types(inputs: &BTreeMap<DataId, Input>) -> BTreeMap<DataId, String> {
    // Quick check: skip allocation if no timer inputs
    if !inputs
        .values()
        .any(|i| matches!(i.mapping, InputMapping::Timer { .. }))
    {
        return BTreeMap::new();
    }
    let mut result = BTreeMap::new();
    for (input_id, input) in inputs {
        if matches!(input.mapping, InputMapping::Timer { .. }) {
            result.insert(input_id.clone(), TIMER_TYPE.to_string());
        }
    }
    result
}

/// Validate that port type keys exist in the port list and URNs resolve.
fn check_port_types(
    node_id: &str,
    type_map: &BTreeMap<DataId, String>,
    contains: impl Fn(&DataId) -> bool,
    port_kind: &str,
    registry: &crate::types::TypeRegistry,
    warnings: &mut Vec<TypeWarning>,
) {
    for (port_id, urn) in type_map {
        if !contains(port_id) {
            warnings.push(TypeWarning {
                node_id: node_id.to_string(),
                message: format!(
                    "{port_kind}_types key \"{port_id}\" not found in {port_kind}s list"
                ),
            });
        }
        if registry.resolve(urn).is_none() {
            let hint = registry
                .suggest(urn)
                .map(|s| format!(" (did you mean \"{s}\"?)"))
                .unwrap_or_default();
            warnings.push(TypeWarning {
                node_id: node_id.to_string(),
                message: format!("unknown type \"{urn}\" on {port_kind} \"{port_id}\"{hint}"),
            });
        }
    }
}

/// Check metadata annotations on outputs (Phase 5).
fn check_metadata_annotations(
    node_id: &str,
    output_metadata: &BTreeMap<DataId, Vec<String>>,
    pattern: &Option<String>,
    outputs: &BTreeSet<DataId>,
    warnings: &mut Vec<TypeWarning>,
) {
    // Check explicit output_metadata keys exist in outputs
    for output_id in output_metadata.keys() {
        if !outputs.contains(output_id) {
            warnings.push(TypeWarning {
                node_id: node_id.to_string(),
                message: format!("output_metadata key \"{output_id}\" not found in outputs list"),
            });
        }
    }

    // Validate pattern shorthand
    if let Some(pat) = pattern
        && crate::types::pattern_metadata_keys(pat).is_none()
    {
        warnings.push(TypeWarning {
            node_id: node_id.to_string(),
            message: format!(
                "unknown pattern \"{pat}\", expected one of: \
                     service-server, service-client, action-server, action-client"
            ),
        });
    }
}

/// Check cross-edge type mismatches using compatibility graph.
///
/// Also performs type inference (Phase 3) and strict-mode unannotated checks.
/// `timer_types` maps input IDs to auto-injected timer type URNs.
#[allow(clippy::too_many_arguments)]
fn check_edge_mismatches_with_compat(
    node_id: &str,
    input_types: &BTreeMap<DataId, String>,
    inputs: &BTreeMap<DataId, Input>,
    output_type_map: &BTreeMap<(String, String), String>,
    annotated_output_ports: &BTreeSet<(String, String)>,
    timer_types: &BTreeMap<DataId, String>,
    compat: &crate::types::CompatibilityGraph,
    registry: &crate::types::TypeRegistry,
    strict: bool,
    warnings: &mut Vec<TypeWarning>,
    inferences: &mut Vec<TypeInference>,
) {
    for (input_id, input) in inputs {
        match &input.mapping {
            InputMapping::User(mapping) => {
                let key = (mapping.source.to_string(), mapping.output.to_string());
                let upstream_urn = output_type_map.get(&key);
                let downstream_urn = input_types.get(input_id);

                match (upstream_urn, downstream_urn) {
                    (Some(out_urn), Some(in_urn)) if !compat.is_compatible(out_urn, in_urn) => {
                        let schema_detail = check_schema_compat(out_urn, in_urn, registry);
                        let detail = schema_detail.map(|d| format!(" ({d})")).unwrap_or_default();
                        warnings.push(TypeWarning {
                            node_id: node_id.to_string(),
                            message: format!(
                                "type mismatch on input \"{input_id}\": \
                                 upstream {}/{} declares \"{out_urn}\", \
                                 but expected \"{in_urn}\"{detail}",
                                mapping.source, mapping.output,
                            ),
                        });
                    }
                    (Some(out_urn), None) => {
                        inferences.push(TypeInference {
                            node_id: node_id.to_string(),
                            port_id: input_id.to_string(),
                            inferred_urn: out_urn.clone(),
                            source: format!("{}/{}", mapping.source, mapping.output),
                        });
                    }
                    // Strict mode, and the upstream has no *resolvable* type.
                    // The guard also excludes an upstream that *is* annotated
                    // but with an unresolvable URN (a typo): such a port is
                    // absent from `output_type_map` yet present in
                    // `annotated_output_ports`, so it falls through to `_` with
                    // no warning here. `check_port_types` already reports the
                    // unknown URN, so a second "has no type annotation" warning
                    // would be factually wrong and send the user chasing a
                    // missing annotation instead of the typo.
                    (None, Some(in_urn)) if strict && !annotated_output_ports.contains(&key) => {
                        warnings.push(TypeWarning {
                            node_id: node_id.to_string(),
                            message: format!(
                                "input \"{input_id}\" expects type \"{in_urn}\" but upstream \
                                 {}/{} has no type annotation",
                                mapping.source, mapping.output,
                            ),
                        });
                    }
                    _ => {}
                }
            }
            InputMapping::Timer { .. } => {
                // Timer inputs auto-inject std/core/v1/UInt64
                if let Some(expected_urn) = input_types.get(input_id) {
                    let timer_urn = timer_types
                        .get(input_id)
                        .map(|s| s.as_str())
                        .unwrap_or(TIMER_TYPE);
                    if !compat.is_compatible(timer_urn, expected_urn) {
                        warnings.push(TypeWarning {
                            node_id: node_id.to_string(),
                            message: format!(
                                "type mismatch on input \"{input_id}\": \
                                 timer provides \"{timer_urn}\", \
                                 but expected \"{expected_urn}\"",
                            ),
                        });
                    }
                }
            }
            // Log subscriptions deliver JSON strings — skip type checking.
            InputMapping::Logs(_) => {}
        }
    }
}

/// Check schema-level compatibility for struct types (Phase 6).
/// Returns an optional detail string for the error message.
fn check_schema_compat(
    out_urn: &str,
    in_urn: &str,
    registry: &crate::types::TypeRegistry,
) -> Option<String> {
    let out_def = registry.resolve(out_urn)?;
    let in_def = registry.resolve(in_urn)?;
    let out_schema = out_def.to_arrow_schema_with_registry(registry)?;
    let in_schema = in_def.to_arrow_schema_with_registry(registry)?;
    // in_schema = expected (consumer), out_schema = actual (producer)
    match crate::types::schema_compatible(&in_schema, &out_schema) {
        Ok(()) => None,
        Err(e) => Some(e.to_string()),
    }
}

fn validate_ros2_type_format(node_id: &NodeId, name: &str, type_str: &str) -> eyre::Result<()> {
    // type must be "package/TypeName" format with alphanumeric+underscore parts
    let parts: Vec<&str> = type_str.split('/').collect();
    if parts.len() != 2 || parts[0].is_empty() || parts[1].is_empty() {
        bail!(
            "node `{node_id}`: invalid type `{type_str}` for `{name}`, \
             expected format `package/TypeName` (e.g. `sensor_msgs/Image`)"
        );
    }
    for (label, part) in [("package", parts[0]), ("type name", parts[1])] {
        if !part.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') {
            bail!(
                "node `{node_id}`: invalid {label} `{part}` in type `{type_str}` for `{name}`, \
                 only ASCII alphanumeric and underscore characters allowed"
            );
        }
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::TypeRegistry;
    use dora_message::config::{Input, InputMapping};
    use dora_message::descriptor::{
        Descriptor, RmwZenohCompatibility, Ros2BridgeConfig, Ros2Direction, Ros2Role,
        Ros2TopicConfig, Ros2TransportConfig,
    };
    use std::{path::PathBuf, time::Duration};

    fn dummy_input() -> Input {
        Input {
            mapping: InputMapping::Timer {
                interval: Duration::from_secs(1),
            },
            queue_size: None,
            input_timeout: None,
            queue_policy: None,
        }
    }

    fn service_config(role: Ros2Role) -> Ros2BridgeConfig {
        Ros2BridgeConfig {
            service: Some("/add_two_ints".into()),
            service_type: Some("example_interfaces/AddTwoInts".into()),
            role: Some(role),
            ..Default::default()
        }
    }

    fn action_config(role: Ros2Role) -> Ros2BridgeConfig {
        Ros2BridgeConfig {
            action: Some("/navigate".into()),
            action_type: Some("nav2_msgs/NavigateToPose".into()),
            role: Some(role),
            ..Default::default()
        }
    }

    fn runtime_node() -> ResolvedNode {
        serde_yaml::from_str(
            r#"
id: runtime-node
operators:
  - id: op1
    python: op.py
    outputs:
      - out
"#,
        )
        .unwrap()
    }

    fn custom_node() -> dora_message::descriptor::CustomNode {
        dora_message::descriptor::CustomNode::new("node".to_string())
    }

    #[test]
    fn timing_fields_accept_finite_non_negative_and_none() {
        let id = NodeId::from("n".to_owned());
        let mut node = custom_node();
        // unset is fine
        check_timing_fields(&id, &node).unwrap();
        // a large finite grace is fine
        node.finish_grace_secs = Some(3600.0);
        node.health_check_timeout = Some(0.0);
        node.startup_timeout = Some(5.0);
        check_timing_fields(&id, &node).unwrap();
    }

    #[test]
    fn timing_fields_reject_negative_finish_grace_secs() {
        let id = NodeId::from("n".to_owned());
        let mut node = custom_node();
        node.finish_grace_secs = Some(-1.0);
        let err = check_timing_fields(&id, &node).unwrap_err().to_string();
        assert!(
            err.contains("finish_grace_secs") && err.contains("non-negative"),
            "error should name the field and the constraint, got: {err}"
        );
    }

    #[test]
    fn timing_fields_reject_negative_startup_timeout() {
        let id = NodeId::from("n".to_owned());
        let mut node = custom_node();
        node.startup_timeout = Some(-0.5);
        let err = check_timing_fields(&id, &node).unwrap_err().to_string();
        assert!(
            err.contains("startup_timeout") && err.contains("non-negative"),
            "error should name the field and the constraint, got: {err}"
        );
    }

    #[test]
    fn dynamic_node_rejects_startup_timeout() {
        let descriptor: Descriptor = serde_yaml::from_str(
            r#"
nodes:
  - id: dyn
    path: dynamic
    startup_timeout: 5.0
"#,
        )
        .unwrap();
        let err = check_dataflow_static(&descriptor).unwrap_err().to_string();
        assert!(
            err.contains("dynamic node `dyn` cannot specify `startup_timeout`"),
            "error should explain dynamic node startup_timeout rejection, got: {err}"
        );
    }

    /// `check_node_timing_values` is the panic guard the `dora start` path
    /// uses, so it must stop at the values `Duration::from_secs_f64` panics
    /// on and leave the dynamic-node `startup_timeout` policy rule to
    /// `check_node_timing` (`dora node add` / `replace`) and
    /// `check_dataflow_static` (`dora run` / `validate`). Were the policy
    /// rule in both, `dora start` would reject a descriptor that starts fine
    /// on 1.0.x.
    #[test]
    fn timing_values_check_skips_the_dynamic_node_policy_rule() {
        let descriptor: Descriptor = serde_yaml::from_str(
            r#"
nodes:
  - id: dyn
    path: dynamic
    startup_timeout: 5.0
"#,
        )
        .unwrap();
        let nodes = descriptor.resolve_aliases_and_set_defaults().unwrap();
        let node = nodes.values().next().unwrap();

        check_node_timing_values(node)
            .expect("an ignored startup_timeout is not a panic risk, so the spawn path allows it");
        let err = check_node_timing(node).unwrap_err().to_string();
        assert!(
            err.contains("dynamic node `dyn` cannot specify `startup_timeout`"),
            "the node-join path keeps the policy rule, got: {err}"
        );

        // The panic guard itself is unchanged by the split.
        let bad: Descriptor = serde_yaml::from_str(
            r#"
nodes:
  - id: dyn
    path: dynamic
    restart_delay: -1.0
"#,
        )
        .unwrap();
        let bad_nodes = bad.resolve_aliases_and_set_defaults().unwrap();
        let err = check_node_timing_values(bad_nodes.values().next().unwrap())
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("restart_delay"),
            "the spawn path still rejects a panicking value, got: {err}"
        );
    }

    #[test]
    fn timing_fields_reject_non_finite_values() {
        let id = NodeId::from("n".to_owned());
        for bad in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
            let mut node = custom_node();
            node.health_check_timeout = Some(bad);
            let err = check_timing_fields(&id, &node).unwrap_err().to_string();
            assert!(
                err.contains("health_check_timeout"),
                "non-finite {bad} should be rejected, got: {err}"
            );
        }
    }

    #[test]
    fn seconds_field_accepts_none_zero_and_positive() {
        check_seconds_field("owner", "field", None, true).unwrap();
        check_seconds_field("owner", "field", Some(0.0), true).unwrap();
        check_seconds_field("owner", "field", Some(3600.0), true).unwrap();
    }

    #[test]
    fn seconds_field_rejects_negative_and_non_finite() {
        for bad in [-1.0, f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
            let err = check_seconds_field("owner", "field", Some(bad), true)
                .unwrap_err()
                .to_string();
            assert!(
                err.contains("field") && err.contains("non-negative"),
                "{bad} should be rejected with a field/constraint message, got: {err}"
            );
        }
    }

    // Finite, non-negative, but too large for `Duration::from_secs_f64`, which
    // would otherwise panic on daemon node spawn (see #2470). `Duration::MAX`'s
    // own `as_secs_f64()` rounds *up* past the representable max, so the exact
    // boundary value must be rejected too — a `value > MAX` guard would wrongly
    // let it through.
    #[test]
    fn seconds_field_rejects_values_that_overflow_duration() {
        for bad in [1e20, Duration::MAX.as_secs_f64()] {
            // Confirms the value genuinely trips the daemon's conversion.
            assert!(Duration::try_from_secs_f64(bad).is_err());
            let err = check_seconds_field("owner", "field", Some(bad), true)
                .unwrap_err()
                .to_string();
            assert!(
                err.contains("field") && err.contains("smaller than"),
                "overflowing {bad} should be rejected with a field/bound message, got: {err}"
            );
        }
    }

    // A large but representable value (well under `Duration::MAX`) stays valid.
    #[test]
    fn seconds_field_accepts_large_representable_value() {
        assert!(Duration::try_from_secs_f64(1e18).is_ok());
        check_seconds_field("owner", "field", Some(1e18), true).unwrap();
    }

    // With `allow_zero: false`, `0.0` must be rejected (in addition to the
    // negative / non-finite cases) because such a field reaches
    // `tokio::time::interval`, which panics on a zero period.
    #[test]
    fn seconds_field_rejects_zero_when_positive_required() {
        check_seconds_field("owner", "field", None, false).unwrap();
        check_seconds_field("owner", "field", Some(3600.0), false).unwrap();
        // `1e-10` is finite and positive but `Duration::from_secs_f64` rounds
        // it down to `Duration::ZERO`, which would panic `tokio::time::interval`
        // just like a literal `0.0`, so it must be rejected too.
        for bad in [0.0, 1e-10, -1.0, f64::NAN, f64::INFINITY] {
            let err = check_seconds_field("owner", "field", Some(bad), false)
                .unwrap_err()
                .to_string();
            assert!(
                err.contains("field") && err.contains("positive"),
                "{bad} should be rejected with a field/constraint message, got: {err}"
            );
        }
        // A tiny-but-positive value is fine when zero is allowed (it does not
        // reach `tokio::time::interval`).
        check_seconds_field("owner", "field", Some(1e-10), true).unwrap();
    }

    // `health_check_interval` (dataflow-level) reaches `Duration::from_secs_f64`
    // and then `tokio::time::interval` in the daemon, so `check_dataflow` must
    // reject a non-finite / negative / zero value rather than let the daemon
    // panic. A zero interval additionally panics `tokio::time::interval`, so it
    // is rejected with a `positive` constraint.
    #[test]
    fn check_dataflow_rejects_negative_health_check_interval() {
        let dataflow = parse_dataflow(
            "\
health_check_interval: -1.0
nodes:
  - id: a
    path: node_a
    build: cargo build
    outputs:
      - out
",
        );
        let err = check_dataflow(&dataflow, Path::new("/nonexistent-dora-validate-test"))
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("health_check_interval") && err.contains("positive"),
            "error should name the field and constraint, got: {err}"
        );
    }

    // A zero `health_check_interval` passes `Duration::from_secs_f64` (yielding
    // `Duration::ZERO`) but then panics `tokio::time::interval`, so it must be
    // rejected up front (regression test for #2752).
    #[test]
    fn check_dataflow_rejects_zero_health_check_interval() {
        let dataflow = parse_dataflow(
            "\
health_check_interval: 0.0
nodes:
  - id: a
    path: node_a
    build: cargo build
    outputs:
      - out
",
        );
        let err = check_dataflow(&dataflow, Path::new("/nonexistent-dora-validate-test"))
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("health_check_interval") && err.contains("positive"),
            "error should name the field and constraint, got: {err}"
        );
    }

    // A tiny-but-positive `health_check_interval` (e.g. `1e-10`) is finite and
    // non-zero as an `f64`, so a literal `value == 0.0` check would let it
    // through -- but `Duration::from_secs_f64` rounds it down to
    // `Duration::ZERO`, which panics `tokio::time::interval`. It must be
    // rejected up front just like `0.0`.
    #[test]
    fn check_dataflow_rejects_subnanosecond_health_check_interval() {
        let dataflow = parse_dataflow(
            "\
health_check_interval: 0.0000000001
nodes:
  - id: a
    path: node_a
    build: cargo build
    outputs:
      - out
",
        );
        let err = check_dataflow(&dataflow, Path::new("/nonexistent-dora-validate-test"))
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("health_check_interval") && err.contains("positive"),
            "error should name the field and constraint, got: {err}"
        );
    }

    // Same guarantee for the per-input `input_timeout`, which the daemon feeds
    // to `Duration::from_secs_f64` on both the spawn and reconnect paths.
    #[test]
    fn check_dataflow_rejects_non_finite_input_timeout() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: a
    path: node_a
    build: cargo build
    outputs:
      - out
  - id: b
    path: node_b
    build: cargo build
    inputs:
      x:
        source: a/out
        input_timeout: .inf
",
        );
        let err = check_dataflow(&dataflow, Path::new("/nonexistent-dora-validate-test"))
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("input_timeout") && err.contains('x'),
            "error should name the offending input and field, got: {err}"
        );
    }

    // A finite `input_timeout` too large for `Duration` must be rejected too:
    // `Duration::from_secs_f64` panics on overflow just as it does on `.inf`.
    #[test]
    fn check_dataflow_rejects_overflowing_input_timeout() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: a
    path: node_a
    build: cargo build
    outputs:
      - out
  - id: b
    path: node_b
    build: cargo build
    inputs:
      x:
        source: a/out
        input_timeout: 1e20
",
        );
        let err = check_dataflow(&dataflow, Path::new("/nonexistent-dora-validate-test"))
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("input_timeout") && err.contains('x'),
            "error should name the offending input and field, got: {err}"
        );
    }

    #[test]
    fn check_dataflow_accepts_valid_interval_and_timeout() {
        let dataflow = parse_dataflow(
            "\
health_check_interval: 2.5
nodes:
  - id: a
    path: node_a
    build: cargo build
    outputs:
      - out
  - id: b
    path: node_b
    build: cargo build
    inputs:
      x:
        source: a/out
        input_timeout: 0.5
",
        );
        check_dataflow(&dataflow, Path::new("/nonexistent-dora-validate-test")).unwrap();
    }

    fn user_input(source: &str, output: &str) -> Input {
        Input {
            mapping: InputMapping::User(UserInputMapping {
                source: NodeId::from(source.to_owned()),
                output: DataId::from(output.to_owned()),
            }),
            queue_size: None,
            input_timeout: None,
            queue_policy: None,
        }
    }

    #[test]
    fn runtime_input_with_operator_segment_is_accepted() {
        let node = runtime_node();
        let nodes = BTreeMap::from([(node.id.clone(), node)]);
        check_input(&user_input("runtime-node", "op1/out"), &nodes, "sink/in").unwrap();
    }

    #[test]
    fn runtime_input_without_operator_segment_reports_expected_format() {
        // Mapping `runtime-node/out` (instead of `runtime-node/op1/out`) used
        // to fall back to an empty operator id and report that operator ``
        // does not exist, hiding the actual mistake.
        let node = runtime_node();
        let nodes = BTreeMap::from([(node.id.clone(), node)]);
        let err = check_input(&user_input("runtime-node", "out"), &nodes, "sink/in")
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("runtime-node/<operator_id>/<output_id>"),
            "error should explain the expected format, got: {err}"
        );
    }

    // A runtime operator whose artifact is produced by a `build:` command must
    // not fail the on-disk source check before the build has run — mirroring the
    // custom-node and shared-library behavior. Exercised via the WASM arm so the
    // test does not require a Python `dora-rs` runtime (the Python arm carries an
    // identical guard).
    #[test]
    fn operator_with_build_command_skips_missing_source_check() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: runtime-node
    operators:
      - id: op1
        wasm: does/not/exist.wasm
        build: cargo build
        outputs:
          - out
",
        );
        check_dataflow(&dataflow, Path::new("/nonexistent-dora-validate-test")).unwrap();
    }

    #[test]
    fn operator_without_build_command_rejects_missing_source() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: runtime-node
    operators:
      - id: op1
        wasm: does/not/exist.wasm
        outputs:
          - out
",
        );
        let err = check_dataflow(&dataflow, Path::new("/nonexistent-dora-validate-test"))
            .unwrap_err()
            .to_string();
        assert!(
            err.contains("no WASM library"),
            "missing source without a build command should still be rejected, got: {err}"
        );
    }

    #[test]
    fn validate_no_mode_set() {
        let config = Ros2BridgeConfig::default();
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &BTreeSet::new(),
        )
        .unwrap_err();
        assert!(err.to_string().contains("requires one of"));
    }

    #[test]
    fn validate_rejects_empty_ros2_zenoh_config_uri() {
        let config = Ros2BridgeConfig {
            transport: Ros2TransportConfig::Zenoh {
                compatibility: RmwZenohCompatibility::Humble,
                config_uri: Some(PathBuf::new()),
            },
            topic: Some("/t".into()),
            message_type: Some("a/B".into()),
            ..Default::default()
        };
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &BTreeSet::from([DataId::from("out".to_owned())]),
        )
        .unwrap_err();
        assert!(err.to_string().contains("config_uri must not be empty"));
    }

    fn single_topic_config(topic: &str, direction: Ros2Direction) -> Ros2BridgeConfig {
        Ros2BridgeConfig {
            topic: Some(topic.into()),
            message_type: Some("turtlesim/Pose".into()),
            direction,
            ..Default::default()
        }
    }

    /// Single-topic mode binds to the node's declared port, so a name that has
    /// nothing to do with the topic is correct and must be accepted.
    #[test]
    fn validate_single_topic_accepts_the_sole_declared_port() {
        let config = single_topic_config("/turtle1/pose", Ros2Direction::Subscribe);
        validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &BTreeSet::from([DataId::from("pose".to_owned())]),
        )
        .unwrap();
    }

    /// Several declared outputs leave the binding ambiguous, which the bridge
    /// would resolve by silently dropping every message.
    #[test]
    fn validate_single_topic_rejects_ambiguous_subscribe_output() {
        let config = single_topic_config("/turtle1/pose", Ros2Direction::Subscribe);
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &BTreeSet::from([
                DataId::from("pose".to_owned()),
                DataId::from("log".to_owned()),
            ]),
        )
        .unwrap_err()
        .to_string();
        assert!(
            err.contains("no unambiguous output")
                && err.contains("`pose`")
                && err.contains("`log`")
                && err.contains("turtle1_pose"),
            "error should list the candidates and the topic-derived id, got: {err}"
        );
    }

    #[test]
    fn validate_multi_topic_rejects_undeclared_subscribe_output() {
        // A subscribe topic mapped to an output the node never declares would
        // pass validation and then silently drop every message at runtime
        // (`DoraNode::send_output` ignores unknown output ids).
        let config = Ros2BridgeConfig {
            topics: Some(vec![Ros2TopicConfig {
                topic: "/scan".into(),
                message_type: "sensor_msgs/LaserScan".into(),
                direction: Ros2Direction::Subscribe,
                output: Some("typo_out".into()),
                input: None,
                qos: None,
            }]),
            ..Default::default()
        };
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &BTreeSet::from([DataId::from("scan".to_owned())]),
        )
        .unwrap_err()
        .to_string();
        assert!(
            err.contains("typo_out") && err.contains("not declared"),
            "error should name the undeclared output, got: {err}"
        );
    }

    #[test]
    fn validate_multi_topic_rejects_undeclared_publish_input() {
        let config = Ros2BridgeConfig {
            topics: Some(vec![Ros2TopicConfig {
                topic: "/cmd_vel".into(),
                message_type: "geometry_msgs/Twist".into(),
                direction: Ros2Direction::Publish,
                output: None,
                input: Some("typo_in".into()),
                qos: None,
            }]),
            ..Default::default()
        };
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::from([(DataId::from("cmd".to_owned()), dummy_input())]),
            &BTreeSet::new(),
        )
        .unwrap_err()
        .to_string();
        assert!(
            err.contains("typo_in") && err.contains("not declared"),
            "error should name the undeclared input, got: {err}"
        );
    }

    #[test]
    fn validate_multi_topic_rejects_undeclared_derived_output() {
        // No explicit `output:` — the bridge derives the output id from the
        // topic name (`/scan` -> `scan`). A derived id that is not a declared
        // output silently drops every message, so validation must reject it.
        let config = Ros2BridgeConfig {
            topics: Some(vec![Ros2TopicConfig {
                topic: "/scan".into(),
                message_type: "sensor_msgs/LaserScan".into(),
                direction: Ros2Direction::Subscribe,
                output: None,
                input: None,
                qos: None,
            }]),
            ..Default::default()
        };
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &BTreeSet::from([DataId::from("laser".to_owned())]),
        )
        .unwrap_err()
        .to_string();
        assert!(
            err.contains("scan") && err.contains("not declared"),
            "error should name the derived output, got: {err}"
        );
    }

    #[test]
    fn validate_multi_topic_accepts_derived_output() {
        // The topic-derived id (`/scan` -> `scan`) matches a declared output.
        let config = Ros2BridgeConfig {
            topics: Some(vec![Ros2TopicConfig {
                topic: "/scan".into(),
                message_type: "sensor_msgs/LaserScan".into(),
                direction: Ros2Direction::Subscribe,
                output: None,
                input: None,
                qos: None,
            }]),
            ..Default::default()
        };
        validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &BTreeSet::from([DataId::from("scan".to_owned())]),
        )
        .expect("a topic-derived output matching a declared output should validate");
    }

    #[test]
    fn validate_multi_topic_accepts_declared_ports() {
        let config = Ros2BridgeConfig {
            topics: Some(vec![
                Ros2TopicConfig {
                    topic: "/scan".into(),
                    message_type: "sensor_msgs/LaserScan".into(),
                    direction: Ros2Direction::Subscribe,
                    output: Some("scan".into()),
                    input: None,
                    qos: None,
                },
                Ros2TopicConfig {
                    topic: "/cmd_vel".into(),
                    message_type: "geometry_msgs/Twist".into(),
                    direction: Ros2Direction::Publish,
                    output: None,
                    input: Some("cmd".into()),
                    qos: None,
                },
            ]),
            ..Default::default()
        };
        validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::from([(DataId::from("cmd".to_owned()), dummy_input())]),
            &BTreeSet::from([DataId::from("scan".to_owned())]),
        )
        .expect("multi-topic config mapping declared ports should validate");
    }

    #[test]
    fn validate_multiple_modes() {
        let config = Ros2BridgeConfig {
            topic: Some("/t".into()),
            service: Some("/s".into()),
            message_type: Some("a/B".into()),
            service_type: Some("a/B".into()),
            role: Some(Ros2Role::Client),
            ..Default::default()
        };
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &BTreeSet::new(),
        )
        .unwrap_err();
        assert!(err.to_string().contains("multiple of"));
    }

    #[test]
    fn validate_service_client_ok() {
        let config = service_config(Ros2Role::Client);
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("request".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("response".to_owned()));
        validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs).unwrap();
    }

    #[test]
    fn validate_service_client_missing_service_type() {
        let config = Ros2BridgeConfig {
            service: Some("/svc".into()),
            role: Some(Ros2Role::Client),
            ..Default::default()
        };
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("request".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("response".to_owned()));
        let err = validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs)
            .unwrap_err();
        assert!(err.to_string().contains("service_type"));
    }

    #[test]
    fn validate_service_client_missing_role() {
        let config = Ros2BridgeConfig {
            service: Some("/svc".into()),
            service_type: Some("a/B".into()),
            ..Default::default()
        };
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("request".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("response".to_owned()));
        let err = validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs)
            .unwrap_err();
        assert!(err.to_string().contains("role"));
    }

    #[test]
    fn validate_service_client_no_inputs() {
        let config = service_config(Ros2Role::Client);
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("response".to_owned()));
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &outputs,
        )
        .unwrap_err();
        assert!(err.to_string().contains("input"));
    }

    #[test]
    fn validate_service_client_no_outputs() {
        let config = service_config(Ros2Role::Client);
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("request".to_owned()), dummy_input());
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &inputs,
            &BTreeSet::new(),
        )
        .unwrap_err();
        assert!(err.to_string().contains("output"));
    }

    #[test]
    fn validate_service_server_ok() {
        let config = service_config(Ros2Role::Server);
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("response".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("request".to_owned()));
        validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs).unwrap();
    }

    #[test]
    fn validate_service_server_no_inputs() {
        let config = service_config(Ros2Role::Server);
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("request".to_owned()));
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &outputs,
        )
        .unwrap_err();
        assert!(err.to_string().contains("input"));
    }

    #[test]
    fn validate_service_server_no_outputs() {
        let config = service_config(Ros2Role::Server);
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("response".to_owned()), dummy_input());
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &inputs,
            &BTreeSet::new(),
        )
        .unwrap_err();
        assert!(err.to_string().contains("output"));
    }

    #[test]
    fn validate_action_client_ok() {
        let config = action_config(Ros2Role::Client);
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("goal".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("feedback".to_owned()));
        outputs.insert(DataId::from("result".to_owned()));
        validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs).unwrap();
    }

    #[test]
    fn validate_action_client_missing_action_type() {
        let config = Ros2BridgeConfig {
            action: Some("/nav".into()),
            role: Some(Ros2Role::Client),
            ..Default::default()
        };
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("goal".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("feedback".to_owned()));
        let err = validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs)
            .unwrap_err();
        assert!(err.to_string().contains("action_type"));
    }

    #[test]
    fn validate_action_client_no_inputs() {
        let config = action_config(Ros2Role::Client);
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("feedback".to_owned()));
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &outputs,
        )
        .unwrap_err();
        assert!(err.to_string().contains("input"));
    }

    #[test]
    fn validate_action_client_no_outputs() {
        let config = action_config(Ros2Role::Client);
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("goal".to_owned()), dummy_input());
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &inputs,
            &BTreeSet::new(),
        )
        .unwrap_err();
        assert!(err.to_string().contains("output"));
    }

    #[test]
    fn validate_action_server_ok() {
        let config = action_config(Ros2Role::Server);
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("feedback".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("goal".to_owned()));
        validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs).unwrap();
    }

    #[test]
    fn validate_action_server_no_inputs() {
        let config = action_config(Ros2Role::Server);
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("goal".to_owned()));
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &BTreeMap::new(),
            &outputs,
        )
        .unwrap_err();
        assert!(err.to_string().contains("input"));
    }

    #[test]
    fn validate_action_server_no_outputs() {
        let config = action_config(Ros2Role::Server);
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("feedback".to_owned()), dummy_input());
        let err = validate_ros2_config(
            &NodeId::from("n".to_owned()),
            &config,
            &inputs,
            &BTreeSet::new(),
        )
        .unwrap_err();
        assert!(err.to_string().contains("output"));
    }

    #[test]
    fn validate_bad_type_format() {
        let config = Ros2BridgeConfig {
            service: Some("/svc".into()),
            service_type: Some("invalid_no_slash".into()),
            role: Some(Ros2Role::Client),
            ..Default::default()
        };
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("request".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("response".to_owned()));
        let err = validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs)
            .unwrap_err();
        assert!(err.to_string().contains("package/TypeName"));
    }

    #[test]
    fn validate_type_rejects_special_chars() {
        let config = Ros2BridgeConfig {
            service: Some("/svc".into()),
            service_type: Some("pkg-bad/Evil".into()),
            role: Some(Ros2Role::Client),
            ..Default::default()
        };
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("request".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("response".to_owned()));
        let err = validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs)
            .unwrap_err();
        assert!(err.to_string().contains("alphanumeric"));
    }

    #[test]
    fn ros2_name_accepts_valid() {
        let node = NodeId::from("n".to_owned());
        for name in [
            "topic",
            "/topic",
            "/a/b/c",
            "/add_two_ints",
            "ns/sub_topic",
            "/navigate",
            "_hidden",
            "/_internal/state",
            "/_ros2cli_node",
        ] {
            validate_ros2_name(&node, "topic", name)
                .unwrap_or_else(|e| panic!("`{name}` should be valid: {e}"));
        }
    }

    #[test]
    fn ros2_name_rejects_double_leading_slash() {
        let node = NodeId::from("n".to_owned());
        let err = validate_ros2_name(&node, "topic", "//topic").unwrap_err();
        assert!(err.to_string().contains("consecutive slashes"));
    }

    #[test]
    fn ros2_name_rejects_trailing_slash() {
        let node = NodeId::from("n".to_owned());
        let err = validate_ros2_name(&node, "topic", "topic/").unwrap_err();
        assert!(err.to_string().contains("end with"));
    }

    #[test]
    fn ros2_name_rejects_consecutive_interior_slashes() {
        let node = NodeId::from("n".to_owned());
        let err = validate_ros2_name(&node, "topic", "topic//sub").unwrap_err();
        assert!(err.to_string().contains("consecutive slashes"));
    }

    #[test]
    fn ros2_name_rejects_bare_slash() {
        let node = NodeId::from("n".to_owned());
        let err = validate_ros2_name(&node, "topic", "/").unwrap_err();
        assert!(err.to_string().contains("bare"));
    }

    #[test]
    fn ros2_name_accepts_leading_underscore() {
        let node = NodeId::from("n".to_owned());
        for name in ["_hidden", "/_internal/state", "/_ros2cli_node"] {
            validate_ros2_name(&node, "topic", name)
                .unwrap_or_else(|e| panic!("`{name}` should be valid (hidden topic): {e}"));
        }
    }

    #[test]
    fn ros2_name_rejects_token_starting_with_digit() {
        let node = NodeId::from("n".to_owned());
        let err = validate_ros2_name(&node, "topic", "/2bad").unwrap_err();
        assert!(err.to_string().contains("digit"));
    }

    #[test]
    fn validate_qos_bad_durability() {
        let config = Ros2BridgeConfig {
            service: Some("/svc".into()),
            service_type: Some("a/B".into()),
            role: Some(Ros2Role::Client),
            qos: dora_message::descriptor::Ros2QosConfig {
                durability: Some("persistent".into()),
                ..Default::default()
            },
            ..Default::default()
        };
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("request".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("response".to_owned()));
        let err = validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs)
            .unwrap_err();
        assert!(err.to_string().contains("durability"));
    }

    #[test]
    fn validate_qos_keep_last_out_of_range() {
        let config = Ros2BridgeConfig {
            service: Some("/svc".into()),
            service_type: Some("a/B".into()),
            role: Some(Ros2Role::Client),
            qos: dora_message::descriptor::Ros2QosConfig {
                keep_last: Some(100_000),
                ..Default::default()
            },
            ..Default::default()
        };
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("request".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("response".to_owned()));
        let err = validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs)
            .unwrap_err();
        assert!(err.to_string().contains("keep_last"));
    }

    // --- Type annotation tests ---

    fn parse_dataflow(yaml: &str) -> Descriptor {
        serde_yaml::from_str(yaml).expect("test YAML should parse")
    }

    #[test]
    fn type_check_no_annotations_no_warnings() {
        let dataflow = parse_dataflow("nodes:\n  - id: a\n");
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert!(warnings.is_empty());
    }

    #[test]
    fn type_check_valid_output_type() {
        let dataflow = parse_dataflow(
            "nodes:\n  - id: camera\n    outputs:\n      - image\n    output_types:\n      image: std/media/v1/Image\n",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert!(warnings.is_empty());
    }

    #[test]
    fn type_check_output_types_key_not_in_outputs() {
        let dataflow = parse_dataflow(
            "nodes:\n  - id: camera\n    output_types:\n      image: std/media/v1/Image\n",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert_eq!(warnings.len(), 1);
        assert!(warnings[0].message.contains("not found in outputs"));
    }

    #[test]
    fn type_check_unknown_urn_with_suggestion() {
        let dataflow = parse_dataflow(
            "nodes:\n  - id: camera\n    outputs:\n      - image\n    output_types:\n      image: std/media/v1/Imag\n",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert_eq!(warnings.len(), 1);
        assert!(warnings[0].message.contains("unknown type"));
        assert!(warnings[0].message.contains("did you mean"));
    }

    #[test]
    fn type_check_matching_edge_types() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: sender
    outputs:
      - data
    output_types:
      data: std/core/v1/Float32
  - id: receiver
    inputs:
      data: sender/data
    input_types:
      data: std/core/v1/Float32
",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert!(warnings.is_empty());
    }

    #[test]
    fn type_check_mismatched_edge_types() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: sender
    outputs:
      - data
    output_types:
      data: std/core/v1/Float32
  - id: receiver
    inputs:
      data: sender/data
    input_types:
      data: std/media/v1/Image
",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert_eq!(warnings.len(), 1);
        assert!(warnings[0].message.contains("type mismatch"));
        assert!(warnings[0].message.contains("Float32"));
        assert!(warnings[0].message.contains("Image"));
    }

    // --- Phase 1: strict mode tests ---

    #[test]
    fn strict_types_parses_in_yaml() {
        let dataflow = parse_dataflow("nodes:\n  - id: a\nstrict_types: true\n");
        assert_eq!(dataflow.strict_types, Some(true));
    }

    #[test]
    fn strict_mode_warns_on_unannotated_upstream() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: sender
    outputs:
      - data
  - id: receiver
    inputs:
      data: sender/data
    input_types:
      data: std/core/v1/Float32
",
        );
        let reg = TypeRegistry::new();
        let result = check_type_annotations_full(&dataflow, &reg, true);
        assert!(!result.warnings.is_empty());
        assert!(result.warnings[0].message.contains("no type annotation"));
    }

    /// Regression: an upstream output that *is* annotated but with an
    /// unresolvable URN (a typo) must not trigger the strict-mode "upstream
    /// has no type annotation" warning. The unknown URN is already reported by
    /// the port-type check; a second "no annotation" warning would be
    /// factually wrong and send the user chasing a missing annotation instead
    /// of the typo.
    #[test]
    fn strict_mode_no_missing_annotation_warning_for_typoed_upstream() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: sender
    outputs:
      - image
    output_types:
      image: std/media/v1/Imag
  - id: receiver
    inputs:
      image: sender/image
    input_types:
      image: std/media/v1/Image
",
        );
        let reg = TypeRegistry::new();
        let result = check_type_annotations_full(&dataflow, &reg, true);
        // The typo itself is still reported...
        assert!(
            result
                .warnings
                .iter()
                .any(|w| w.message.contains("unknown type")),
            "expected an unknown-type warning for the typo, got: {:?}",
            result.warnings
        );
        // ...but the misleading "no type annotation" warning must not appear:
        // the upstream *is* annotated, just misspelled.
        assert!(
            !result
                .warnings
                .iter()
                .any(|w| w.message.contains("no type annotation")),
            "must not claim the annotated-but-typoed upstream has no annotation, got: {:?}",
            result.warnings
        );
    }

    #[test]
    fn non_strict_no_warning_on_unannotated_upstream() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: sender
    outputs:
      - data
  - id: receiver
    inputs:
      data: sender/data
    input_types:
      data: std/core/v1/Float32
",
        );
        let reg = TypeRegistry::new();
        let result = check_type_annotations_full(&dataflow, &reg, false);
        assert!(result.warnings.is_empty());
    }

    // --- Phase 3: type inference tests ---

    #[test]
    fn inference_from_annotated_upstream() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: sensor
    outputs:
      - reading
    output_types:
      reading: std/core/v1/Float64
  - id: processor
    inputs:
      reading: sensor/reading
",
        );
        let reg = TypeRegistry::new();
        let result = check_type_annotations_full(&dataflow, &reg, false);
        assert!(result.warnings.is_empty());
        assert_eq!(result.inferences.len(), 1);
        assert_eq!(result.inferences[0].inferred_urn, "std/core/v1/Float64");
        assert_eq!(result.inferences[0].port_id, "reading");
    }

    #[test]
    fn no_inference_when_both_annotated() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: sender
    outputs:
      - data
    output_types:
      data: std/core/v1/Float32
  - id: receiver
    inputs:
      data: sender/data
    input_types:
      data: std/core/v1/Float32
",
        );
        let reg = TypeRegistry::new();
        let result = check_type_annotations_full(&dataflow, &reg, false);
        assert!(result.inferences.is_empty());
    }

    #[test]
    fn no_inference_when_neither_annotated() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: sender
    outputs:
      - data
  - id: receiver
    inputs:
      data: sender/data
",
        );
        let reg = TypeRegistry::new();
        let result = check_type_annotations_full(&dataflow, &reg, false);
        assert!(result.inferences.is_empty());
    }

    // --- Phase 4: compatibility tests ---

    #[test]
    fn compat_uint8_to_uint32_edge() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: sender
    outputs:
      - data
    output_types:
      data: std/core/v1/UInt8
  - id: receiver
    inputs:
      data: sender/data
    input_types:
      data: std/core/v1/UInt32
",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert!(warnings.is_empty(), "UInt8 -> UInt32 should be compatible");
    }

    #[test]
    fn compat_any_to_bytes_edge() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: sender
    outputs:
      - data
    output_types:
      data: std/media/v1/Image
  - id: receiver
    inputs:
      data: sender/data
    input_types:
      data: std/core/v1/Bytes
",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert!(
            warnings.is_empty(),
            "anything -> Bytes should be compatible"
        );
    }

    #[test]
    fn compat_user_defined_rule_in_yaml() {
        let dataflow = parse_dataflow(
            "\
type_rules:
  - from: std/core/v1/UInt8
    to: std/core/v1/String
nodes:
  - id: sender
    outputs:
      - data
    output_types:
      data: std/core/v1/UInt8
  - id: receiver
    inputs:
      data: sender/data
    input_types:
      data: std/core/v1/String
",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert!(
            warnings.is_empty(),
            "user-defined rule should make this compatible"
        );
    }

    // --- Phase 5: metadata tests ---

    #[test]
    fn metadata_pattern_resolves() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: srv
    pattern: service-server
    outputs:
      - response
",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert!(warnings.is_empty());
    }

    #[test]
    fn metadata_unknown_pattern() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: srv
    pattern: unknown-pattern
    outputs:
      - response
",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert_eq!(warnings.len(), 1);
        assert!(warnings[0].message.contains("unknown pattern"));
    }

    #[test]
    fn metadata_output_key_not_in_outputs() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: srv
    output_metadata:
      missing_port: [request_id]
    outputs:
      - response
",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert_eq!(warnings.len(), 1);
        assert!(warnings[0].message.contains("output_metadata key"));
    }

    // --- Timer auto-inject test ---

    #[test]
    fn timer_input_type_mismatch() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: node
    inputs:
      tick: dora/timer/millis/100
    input_types:
      tick: std/media/v1/Image
",
        );
        let reg = TypeRegistry::new();
        let warnings = check_type_annotations_full(&dataflow, &reg, false).warnings;
        assert!(!warnings.is_empty());
        assert!(warnings[0].message.contains("type mismatch"));
    }

    #[test]
    fn validate_qos_negative_lease_duration() {
        let config = Ros2BridgeConfig {
            service: Some("/svc".into()),
            service_type: Some("a/B".into()),
            role: Some(Ros2Role::Client),
            qos: dora_message::descriptor::Ros2QosConfig {
                lease_duration: Some(-1.0),
                ..Default::default()
            },
            ..Default::default()
        };
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("request".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("response".to_owned()));
        let err = validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs)
            .unwrap_err();
        assert!(err.to_string().contains("lease_duration"));
    }

    /// A finite but enormous QoS duration passes the old
    /// `is_finite()` / `>= 0` check yet overflows `Duration`, which panics
    /// `Duration::from_secs_f64` in the ROS2 bridge QoS conversion. Both
    /// fields must be rejected at descriptor validation instead.
    #[test]
    fn validate_qos_overflowing_durations_are_rejected() {
        let base = Ros2BridgeConfig {
            service: Some("/svc".into()),
            service_type: Some("a/B".into()),
            role: Some(Ros2Role::Client),
            ..Default::default()
        };
        let mut inputs = BTreeMap::new();
        inputs.insert(DataId::from("request".to_owned()), dummy_input());
        let mut outputs = BTreeSet::new();
        outputs.insert(DataId::from("response".to_owned()));

        // `1e300` seconds is finite and positive but far beyond `Duration::MAX`.
        for (field, qos) in [
            (
                "lease_duration",
                dora_message::descriptor::Ros2QosConfig {
                    lease_duration: Some(1e300),
                    ..Default::default()
                },
            ),
            (
                "max_blocking_time",
                dora_message::descriptor::Ros2QosConfig {
                    reliable: true,
                    max_blocking_time: Some(1e300),
                    ..Default::default()
                },
            ),
        ] {
            let config = Ros2BridgeConfig {
                qos,
                ..base.clone()
            };
            let err =
                validate_ros2_config(&NodeId::from("n".to_owned()), &config, &inputs, &outputs)
                    .unwrap_err();
            assert!(
                err.to_string().contains(field),
                "expected error naming `{field}`, got: {err}"
            );
        }
    }

    // --- Wiring validation tests ---

    #[test]
    fn wiring_valid_dataflow() {
        let yaml = r#"
nodes:
  - id: source
    path: source.py
    outputs:
      - data
  - id: sink
    path: sink.py
    inputs:
      data: source/data
"#;
        let descriptor: Descriptor = serde_yaml::from_str(yaml).unwrap();
        check_wiring(&descriptor).unwrap();
    }

    /// Every shipped `ros2:` example must satisfy the port-mapping rule the
    /// bridge binds by. The docs present these as working dataflows, and a
    /// mismatch is silent data loss at runtime rather than a startup error, so
    /// parsing them is not enough — they have to be validated.
    #[test]
    fn ros2_example_dataflows_have_valid_port_mappings() {
        let examples = [
            "examples/ros2-bridge/yaml-bridge/dataflow.yml",
            "examples/ros2-bridge/yaml-bridge/dataflow-zenoh.yml",
            "examples/ros2-bridge/yaml-bridge-service/dataflow-client.yml",
            "examples/ros2-bridge/yaml-bridge-service/dataflow-client-zenoh.yml",
            "examples/ros2-bridge/yaml-bridge-service/dataflow-server.yml",
            "examples/ros2-bridge/yaml-bridge-service/dataflow-server-zenoh.yml",
            "examples/ros2-bridge/yaml-bridge-action/dataflow.yml",
            "examples/ros2-bridge/yaml-bridge-action/dataflow-zenoh.yml",
            "examples/ros2-bridge/yaml-bridge-action-server/dataflow.yml",
        ];
        for relative in examples {
            let Some(yaml) = repo_fixture(relative) else {
                continue; // packaged crate: the examples tree is not shipped
            };
            let descriptor: Descriptor = serde_yaml::from_str(&yaml).unwrap();
            if let Err(err) = validate_ros2_configs(&descriptor) {
                panic!("shipped example `{relative}` has an invalid ros2 config: {err}");
            }
        }
    }

    /// Reads a fixture that lives outside this crate, or `None` when this is
    /// not a repository checkout. `include_str!` would be the obvious choice,
    /// but a path that leaves the crate directory is not in the published
    /// `.crate`, so the crate would fail to *compile* its tests for anyone
    /// building from crates.io (#3400).
    ///
    /// The workspace manifest is the marker for "we are in the repo". Only
    /// its absence skips: inside a checkout a missing fixture is a stale
    /// path and panics, so this keeps the one property `include_str!` had
    /// that a plain `.ok()` would throw away.
    fn repo_fixture(relative: &str) -> Option<String> {
        let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
        if !root.join("Cargo.toml").is_file() {
            return None;
        }
        let path = root.join(relative);
        Some(
            std::fs::read_to_string(&path)
                .unwrap_or_else(|e| panic!("fixture {} is missing: {e}", path.display())),
        )
    }

    #[test]
    fn ros2_zenoh_documentation_examples_parse_with_explicit_profiles() {
        let examples = [
            "examples/ros2-bridge/yaml-bridge/dataflow-zenoh.yml",
            "examples/ros2-bridge/yaml-bridge-service/dataflow-client-zenoh.yml",
            "examples/ros2-bridge/yaml-bridge-action/dataflow-zenoh.yml",
        ];
        for relative in examples {
            let Some(yaml) = repo_fixture(relative) else {
                continue; // packaged crate: the examples tree is not shipped
            };
            let descriptor: Descriptor = serde_yaml::from_str(&yaml).unwrap();
            let ros2 = descriptor
                .nodes
                .iter()
                .find_map(|node| node.ros2.as_ref())
                .unwrap();
            assert!(matches!(
                ros2.transport,
                Ros2TransportConfig::Zenoh {
                    compatibility: RmwZenohCompatibility::Humble,
                    ..
                }
            ));
        }
    }

    #[test]
    fn ros2_zenoh_documentation_links_upstream_wire_contract() {
        let Some(guide) = repo_fixture("guide/src/advanced/ros2-bridge.md") else {
            return; // packaged crate: the guide is not shipped
        };
        assert!(guide.contains("https://github.com/ros2/rmw_zenoh/blob/rolling/docs/design.md"));
        assert!(guide.contains("https://www.ros.org/reps/rep-2016.html"));
    }

    #[test]
    fn wiring_rejects_nonexistent_source_node() {
        let yaml = r#"
nodes:
  - id: sink
    path: sink.py
    inputs:
      data: nonexistent/data
"#;
        let descriptor: Descriptor = serde_yaml::from_str(yaml).unwrap();
        let err = check_wiring(&descriptor).unwrap_err();
        assert!(
            err.to_string().contains("nonexistent"),
            "expected error about missing node, got: {err}"
        );
    }

    #[test]
    fn wiring_rejects_nonexistent_output() {
        let yaml = r#"
nodes:
  - id: source
    path: source.py
    outputs:
      - data
  - id: sink
    path: sink.py
    inputs:
      data: source/typo
"#;
        let descriptor: Descriptor = serde_yaml::from_str(yaml).unwrap();
        let err = check_wiring(&descriptor).unwrap_err();
        assert!(
            err.to_string().contains("typo"),
            "expected error about missing output, got: {err}"
        );
    }

    #[test]
    fn wiring_runtime_input_id_order() {
        // A runtime node's operator input that references a missing source
        // must report the input in the conventional
        // `{node_id}/{operator_id}/{input_id}` order (see #2019).
        let yaml = r#"
nodes:
  - id: runtime-node
    operators:
      - id: my-operator
        shared-library: op
        inputs:
          tick: nonexistent/data
        outputs:
          - status
"#;
        let descriptor: Descriptor = serde_yaml::from_str(yaml).unwrap();
        let err = check_wiring(&descriptor).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("runtime-node/my-operator/tick"),
            "expected node/operator/input order, got: {msg}"
        );
        assert!(
            !msg.contains("my-operator/runtime-node/tick"),
            "input id should not use reversed operator/node order, got: {msg}"
        );
    }

    // --- Single-`operator:` producer edges (short-form references) ---
    //
    // A consumer of a single-`operator:` node references its output in short
    // form (`node/output`) in the source YAML. The `op/` prefix is only
    // injected later by `resolve_aliases_and_set_defaults`, which the
    // pre-build type check does not run. These tests guard that the type
    // check still sees the upstream type across such an edge.

    #[test]
    fn infers_type_across_single_operator_edge() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: producer
    operator:
      python: producer.py
      outputs:
        - result
      output_types:
        result: std/core/v1/Float64
  - id: consumer
    inputs:
      reading: producer/result
",
        );
        let reg = TypeRegistry::new();
        let result = check_type_annotations_full(&dataflow, &reg, false);
        assert!(
            result.warnings.is_empty(),
            "unexpected: {:?}",
            result.warnings
        );
        assert_eq!(
            result.inferences.len(),
            1,
            "should infer from the operator output"
        );
        assert_eq!(result.inferences[0].inferred_urn, "std/core/v1/Float64");
        assert_eq!(result.inferences[0].port_id, "reading");
    }

    #[test]
    fn detects_mismatch_across_single_operator_edge() {
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: producer
    operator:
      python: producer.py
      outputs:
        - result
      output_types:
        result: std/core/v1/Float64
  - id: consumer
    inputs:
      reading: producer/result
    input_types:
      reading: std/core/v1/Int32
",
        );
        let reg = TypeRegistry::new();
        let result = check_type_annotations_full(&dataflow, &reg, false);
        assert!(
            result
                .warnings
                .iter()
                .any(|w| w.message.contains("type mismatch")),
            "expected a type mismatch warning, got: {:?}",
            result.warnings
        );
    }

    #[test]
    fn strict_mode_no_false_positive_across_single_operator_edge() {
        // Upstream single-operator output IS annotated, so strict mode must
        // not claim it has "no type annotation".
        let dataflow = parse_dataflow(
            "\
nodes:
  - id: producer
    operator:
      python: producer.py
      outputs:
        - result
      output_types:
        result: std/core/v1/Float64
  - id: consumer
    inputs:
      reading: producer/result
    input_types:
      reading: std/core/v1/Float64
",
        );
        let reg = TypeRegistry::new();
        let result = check_type_annotations_full(&dataflow, &reg, true);
        assert!(
            !result
                .warnings
                .iter()
                .any(|w| w.message.contains("no type annotation")),
            "annotated upstream must not trigger a strict no-annotation warning, got: {:?}",
            result.warnings
        );
    }

    // --- Focused unit tests for parse_byte_size and parse_log_level ---
    //
    // Added 2026-04-08 to close the biggest mutation-score gap in
    // dora-core: 14 escaped mutants in parse_byte_size and 6 in
    // parse_log_level (20 of the 91 missed mutants in validate.rs).
    // See docs/qa-baseline-2026-04-07.md.
    //
    // These tests exercise every match arm and every boundary case by
    // construction, so any mutation that deletes an arm, flips a
    // comparison, or changes a multiplier is caught.

    // parse_byte_size: every unit produces the exact byte count ----

    #[test]
    fn parse_byte_size_bare_number() {
        assert_eq!(parse_byte_size("0").unwrap(), 0);
        assert_eq!(parse_byte_size("1").unwrap(), 1);
        assert_eq!(parse_byte_size("100").unwrap(), 100);
        assert_eq!(parse_byte_size("1000000").unwrap(), 1_000_000);
    }

    #[test]
    fn parse_byte_size_bytes_unit() {
        assert_eq!(parse_byte_size("0B").unwrap(), 0);
        assert_eq!(parse_byte_size("1B").unwrap(), 1);
        assert_eq!(parse_byte_size("42B").unwrap(), 42);
        // case insensitive
        assert_eq!(parse_byte_size("42b").unwrap(), 42);
    }

    #[test]
    fn parse_byte_size_kilobyte_units() {
        assert_eq!(parse_byte_size("1KB").unwrap(), 1024);
        assert_eq!(parse_byte_size("1K").unwrap(), 1024);
        assert_eq!(parse_byte_size("2KB").unwrap(), 2048);
        assert_eq!(parse_byte_size("4K").unwrap(), 4096);
        // case insensitive
        assert_eq!(parse_byte_size("1kb").unwrap(), 1024);
        assert_eq!(parse_byte_size("1k").unwrap(), 1024);
    }

    #[test]
    fn parse_byte_size_megabyte_units() {
        assert_eq!(parse_byte_size("1MB").unwrap(), 1024 * 1024);
        assert_eq!(parse_byte_size("1M").unwrap(), 1024 * 1024);
        assert_eq!(parse_byte_size("10MB").unwrap(), 10 * 1024 * 1024);
        // distinct from KB: pins the multiplier ratio
        assert_eq!(
            parse_byte_size("1MB").unwrap(),
            1024 * parse_byte_size("1KB").unwrap()
        );
    }

    #[test]
    fn parse_byte_size_gigabyte_units() {
        assert_eq!(parse_byte_size("1GB").unwrap(), 1024 * 1024 * 1024);
        assert_eq!(parse_byte_size("1G").unwrap(), 1024 * 1024 * 1024);
        // distinct from MB: pins the multiplier ratio
        assert_eq!(
            parse_byte_size("1GB").unwrap(),
            1024 * parse_byte_size("1MB").unwrap()
        );
    }

    #[test]
    fn parse_byte_size_all_units_are_distinct() {
        // Pins each unit's multiplier to catch mutations that swap arms
        // (e.g., KB → MB would make "1KB" and "1MB" equal).
        let b = parse_byte_size("1B").unwrap();
        let kb = parse_byte_size("1KB").unwrap();
        let mb = parse_byte_size("1MB").unwrap();
        let gb = parse_byte_size("1GB").unwrap();
        assert_eq!(b, 1);
        assert_eq!(kb, 1024);
        assert_eq!(mb, 1024 * kb);
        assert_eq!(gb, 1024 * mb);
        // And strict ordering
        assert!(b < kb);
        assert!(kb < mb);
        assert!(mb < gb);
    }

    #[test]
    fn parse_byte_size_accepts_sizes_above_4gib() {
        // `usize` is 32 bits on i686/armv7; the log size limit must not be
        // bounded by it (#3604).
        assert_eq!(parse_byte_size("8GB").unwrap(), 8 << 30);
        assert_eq!(parse_byte_size("5000MB").unwrap(), 5000 << 20);
        assert_eq!(parse_byte_size("4294967296").unwrap(), 1 << 32);
        assert_eq!(parse_byte_size("4.5GB").unwrap(), 9 << 29);
    }

    #[test]
    fn parse_byte_size_float_path() {
        // When the integer parse fails, fall back to float.
        assert_eq!(parse_byte_size("1.5KB").unwrap(), 1536);
        assert_eq!(parse_byte_size("0.5MB").unwrap(), 512 * 1024);
        assert_eq!(parse_byte_size("2.25KB").unwrap(), 2304);
    }

    #[test]
    fn parse_byte_size_whitespace_tolerated() {
        assert_eq!(parse_byte_size(" 1KB ").unwrap(), 1024);
        assert_eq!(parse_byte_size("1 KB").unwrap(), 1024);
        assert_eq!(parse_byte_size("  1  KB  ").unwrap(), 1024);
    }

    #[test]
    fn parse_byte_size_rejects_negative() {
        // A negative f64 cast to u64 saturates to 0, which would silently
        // disable a limit instead of being rejected (issue #2018).
        assert!(parse_byte_size("-1KB").is_err());
        assert!(parse_byte_size("-0.5MB").is_err());
        assert!(parse_byte_size("-1").is_err());
        assert!(parse_byte_size("-100").is_err());
    }

    #[test]
    fn parse_byte_size_rejects_non_finite() {
        // inf/nan cast to u64 saturate (to u64::MAX / 0) rather than erroring.
        assert!(parse_byte_size("infKB").is_err());
        assert!(parse_byte_size("nanMB").is_err());
    }

    #[test]
    fn parse_byte_size_rejects_unknown_unit() {
        assert!(parse_byte_size("1TB").is_err());
        assert!(parse_byte_size("1XB").is_err());
        assert!(parse_byte_size("1foo").is_err());
    }

    #[test]
    fn parse_byte_size_rejects_integer_overflow() {
        // num fits in u64 but num * multiplier overflows; must not panic
        // (debug) or silently wrap (release).
        assert!(parse_byte_size("20000000000GB").is_err());
        assert!(parse_byte_size("18446744073709551615GB").is_err());
    }

    #[test]
    fn parse_byte_size_rejects_float_overflow() {
        // num does not fit in u64 (so it takes the f64 path) and is finite and
        // positive, but num * multiplier exceeds u64::MAX. Casting the result
        // to u64 saturates to u64::MAX instead of erroring, so it must be
        // rejected up front rather than silently clamped.
        assert!(parse_byte_size("99999999999999999999GB").is_err());
        assert!(parse_byte_size("99999999999999999999.0GB").is_err());
        assert!(parse_byte_size("184467440737095516160B").is_err());
    }

    #[test]
    fn parse_byte_size_rejects_invalid_number() {
        assert!(parse_byte_size("abc").is_err());
        assert!(parse_byte_size("abcKB").is_err());
        assert!(parse_byte_size("1.2.3KB").is_err());
    }

    #[test]
    fn send_as_rejects_names_that_are_not_valid_output_ids() {
        for (field, value) in [
            ("send_stdout_as", "std out"),
            ("send_stdout_as", "out/"),
            ("send_logs_as", "\"\""),
            ("send_logs_as", "a//b"),
        ] {
            let yaml = format!(
                "nodes:\n  - id: talker\n    path: talker\n    {field}: {value}\n    outputs: [out]\n"
            );
            let descriptor: Descriptor = serde_yaml::from_str(&yaml).unwrap();
            let err = check_dataflow_static(&descriptor)
                .expect_err(&format!("`{field}: {value}` must be rejected"));
            assert!(
                format!("{err:?}").contains("is not a valid output id"),
                "unexpected error for `{field}: {value}`: {err:?}"
            );
        }
    }

    #[test]
    fn send_as_rejects_empty_operator_entry() {
        let yaml = r#"
nodes:
  - id: rt
    operators:
      - id: op1
        python: op.py
        send_stdout_as: ""
        send_logs_as: ""
        outputs: [out]
"#;
        let descriptor: Descriptor = serde_yaml::from_str(yaml).unwrap();
        let nodes = descriptor.resolve_aliases_and_set_defaults().unwrap();
        let node = &nodes[&NodeId::from("rt".to_string())];
        assert!(node.send_stdout_as().is_err());
        assert!(node.send_logs_as().is_err());
    }

    #[test]
    fn send_as_accepts_valid_names() {
        let yaml = r#"
nodes:
  - id: rt
    operators:
      - id: op1
        python: op.py
        send_stdout_as: stdout
        send_logs_as: logs
        outputs: [stdout, logs]
  - id: talker
    path: talker
    send_stdout_as: stdout
    send_logs_as: logs
    outputs: [stdout, logs]
"#;
        let descriptor: Descriptor = serde_yaml::from_str(yaml).unwrap();
        let nodes = descriptor.resolve_aliases_and_set_defaults().unwrap();
        let rt = &nodes[&NodeId::from("rt".to_string())];
        assert_eq!(rt.send_stdout_as().unwrap().as_deref(), Some("op1/stdout"));
        assert_eq!(rt.send_logs_as().unwrap().as_deref(), Some("op1/logs"));
        let talker = &nodes[&NodeId::from("talker".to_string())];
        assert_eq!(talker.send_stdout_as().unwrap().as_deref(), Some("stdout"));
        assert_eq!(talker.send_logs_as().unwrap().as_deref(), Some("logs"));
    }

    // parse_log_level: every level variant ----

    #[test]
    fn parse_log_level_all_levels() {
        use dora_message::common::LogLevelOrStdout;

        assert!(matches!(
            parse_log_level("error").unwrap(),
            LogLevelOrStdout::LogLevel(log::Level::Error)
        ));
        assert!(matches!(
            parse_log_level("warn").unwrap(),
            LogLevelOrStdout::LogLevel(log::Level::Warn)
        ));
        assert!(matches!(
            parse_log_level("info").unwrap(),
            LogLevelOrStdout::LogLevel(log::Level::Info)
        ));
        assert!(matches!(
            parse_log_level("debug").unwrap(),
            LogLevelOrStdout::LogLevel(log::Level::Debug)
        ));
        assert!(matches!(
            parse_log_level("trace").unwrap(),
            LogLevelOrStdout::LogLevel(log::Level::Trace)
        ));
        assert!(matches!(
            parse_log_level("stdout").unwrap(),
            LogLevelOrStdout::Stdout
        ));
    }

    #[test]
    fn parse_log_level_case_insensitive() {
        use dora_message::common::LogLevelOrStdout;

        for variant in ["ERROR", "Error", "error", "ErRoR"] {
            assert!(matches!(
                parse_log_level(variant).unwrap(),
                LogLevelOrStdout::LogLevel(log::Level::Error)
            ));
        }
        for variant in ["STDOUT", "Stdout", "stdout"] {
            assert!(matches!(
                parse_log_level(variant).unwrap(),
                LogLevelOrStdout::Stdout
            ));
        }
    }

    #[test]
    fn parse_log_level_rejects_unknown() {
        assert!(parse_log_level("").is_err());
        assert!(parse_log_level("INVALID").is_err());
        assert!(parse_log_level("fatal").is_err());
        assert!(parse_log_level("log").is_err());
    }

    #[test]
    fn parse_log_level_error_message_lists_options() {
        let err = parse_log_level("bogus").unwrap_err().to_string();
        // The error message should name at least the valid options so
        // users can fix typos. Pinning this so the fmt::Display impl
        // used in the bail! macro stays informative.
        for expected in ["error", "warn", "info", "debug", "trace", "stdout"] {
            assert!(
                err.contains(expected),
                "expected '{expected}' to be mentioned in error, got: {err}"
            );
        }
    }

    #[test]
    fn max_rotated_files_accepts_zero_and_still_caps_at_100() {
        let node = |n: u32| -> ResolvedNode {
            let mut custom = custom_node();
            custom.max_rotated_files = Some(n);
            ResolvedNode::new(NodeId::from("n".to_owned()), CoreNodeKind::Custom(custom))
        };

        // 0 is a real configuration: keep the active log only, rotating the
        // previous one away. The documented disk bound
        // `max_log_size * (1 + max_rotated_files)` is one file at 0.
        assert_eq!(node(0).max_rotated_files().unwrap(), Some(0));
        // The upper bound is unchanged.
        assert_eq!(node(100).max_rotated_files().unwrap(), Some(100));
        assert!(node(101).max_rotated_files().is_err());
    }
}

/// Property-based tests for the pure parsing/validation helpers. These
/// complement the Kani proofs in `crate::metadata` — see
/// `docs/formal-verification.md` for how the tiers fit together.
#[cfg(test)]
mod proptest_properties {
    use super::{parse_byte_size, validate_ros2_name};
    use dora_message::id::NodeId;
    use proptest::prelude::*;

    fn node_id() -> NodeId {
        NodeId::from("prop".to_owned())
    }

    /// Strategy for ROS2 names that are valid by construction: optional
    /// leading '/', then '/'-separated tokens matching `[A-Za-z_][A-Za-z0-9_]*`.
    fn valid_ros2_name() -> impl Strategy<Value = String> {
        (
            any::<bool>(),
            prop::collection::vec("[A-Za-z_][A-Za-z0-9_]{0,8}", 1..4),
        )
            .prop_map(|(absolute, tokens)| {
                let joined = tokens.join("/");
                if absolute {
                    format!("/{joined}")
                } else {
                    joined
                }
            })
    }

    proptest! {
        /// Total: never panics, whatever the input.
        #[test]
        fn ros2_name_validation_never_panics(name in ".{0,32}") {
            let _ = validate_ros2_name(&node_id(), "topic", &name);
        }

        /// Completeness: structurally valid names are always accepted.
        #[test]
        fn ros2_name_accepts_valid_names(name in valid_ros2_name()) {
            prop_assert!(validate_ros2_name(&node_id(), "topic", &name).is_ok());
        }

        /// Soundness: every accepted name satisfies all documented rules.
        ///
        /// The strategy samples from the valid alphabet plus a few
        /// near-miss characters ('.', '-', ' ') so that *accepted* names
        /// are common — a fully random unicode strategy would virtually
        /// never enter the accepted branch, making the property vacuous.
        #[test]
        fn ros2_name_accepted_implies_well_formed(name in "[a-zA-Z0-9_/. -]{0,16}") {
            if validate_ros2_name(&node_id(), "topic", &name).is_ok() {
                prop_assert!(!name.is_empty());
                prop_assert!(name.chars().all(|c| c.is_ascii_alphanumeric()
                    || c == '_'
                    || c == '/'));
                prop_assert!(!name.contains("//"));
                prop_assert!(!name.ends_with('/'));
                for (i, token) in name.split('/').enumerate() {
                    if i == 0 && token.is_empty() {
                        continue;
                    }
                    prop_assert!(!token.starts_with(|c: char| c.is_ascii_digit()));
                }
            }
        }

        /// Total: never panics, whatever the input.
        #[test]
        fn byte_size_parsing_never_panics(s in ".{0,24}") {
            let _ = parse_byte_size(&s);
        }

        /// Integer inputs with units multiply exactly (no float rounding)
        /// and overflow is reported as an error, never wrapped.
        #[test]
        fn byte_size_integer_units_are_exact(
            num in any::<u64>(),
            unit_idx in 0usize..7,
            lowercase in any::<bool>(),
        ) {
            // All unit spellings accepted by `parse_byte_size` (which also
            // uppercases, so lowercase forms must behave identically).
            let (unit, multiplier) =
                [("B", 1u64), ("KB", 1 << 10), ("K", 1 << 10), ("MB", 1 << 20),
                 ("M", 1 << 20), ("GB", 1 << 30), ("G", 1 << 30)][unit_idx];
            let unit = if lowercase { unit.to_lowercase() } else { unit.to_string() };
            match parse_byte_size(&format!("{num}{unit}")) {
                Ok(bytes) => prop_assert_eq!(Some(bytes), num.checked_mul(multiplier)),
                Err(_) => prop_assert!(num.checked_mul(multiplier).is_none()),
            }
        }

        /// Bare integers parse to themselves.
        #[test]
        fn byte_size_bare_integer_is_identity(num in any::<u64>()) {
            prop_assert_eq!(parse_byte_size(&num.to_string()).unwrap(), num);
        }

        /// The float fallback path is total for ordinary decimals: any
        /// simple non-negative decimal with a valid unit parses Ok.
        #[test]
        fn byte_size_fractional_inputs_parse(
            int_part in 0u32..1_000_000,
            frac in 0u32..100,
            unit_idx in 0usize..7,
        ) {
            let unit = ["B", "KB", "K", "MB", "M", "GB", "G"][unit_idx];
            let input = format!("{int_part}.{frac:02}{unit}");
            prop_assert!(parse_byte_size(&input).is_ok(), "failed to parse '{input}'");
        }
    }
}