cerulion_cli 0.0.1-alpha

// SPDX-License-Identifier: AGPL-3.0-only
//! Cerulion CLI — thin wrapper around `cerulion_cli_engine`.

mod cli;

use std::path::PathBuf;
use std::process::ExitCode;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

use clap::Parser;

use cerulion_cli_engine::error::CliResult;
use cerulion_cli_engine::workspace::CerulionWorkspace;
use cerulion_cli_engine::{graph_cmd, node_cmd, schema_cmd, topic_cmd, workspace};
use cli::{
    Cli, Commands, GraphAction, NodeAction, SchemaAction, TopicAction, TraceAction, WorkspaceAction,
};

fn main() -> ExitCode {
    // Quiet iceoryx2's own logger by default. It writes a `[W] Notifier { ... }`
    // pretty-print of internal state on every notify path, which dwarfs our
    // tracing output in the demo. Users can still override by exporting
    // `IOX2_LOG_LEVEL` themselves.
    if std::env::var_os("IOX2_LOG_LEVEL").is_none() {
        std::env::set_var("IOX2_LOG_LEVEL", "error");
    }

    let cli = Cli::parse();
    // The TUI takes over the terminal and is fragile to anything writing
    // to stderr/stdout out-of-band — every `tracing::info!("transport
    // manager initialized")` (and friends) shreds the ratatui frame.
    // Skip the tracing setup entirely in TUI mode; users who need debug
    // output can run `RUST_LOG=... cerulion graph run` in a sibling pane.
    if !matches!(cli.command, Commands::Tui) {
        init_logging(cli.verbose);
    }
    // CER-377 #10: bridge iceoryx2's internal logger to `tracing`
    // so iceoryx2's diagnostics (cleanup failures, lock contention,
    // etc.) flow through Cerulion's standard log pipeline AND can
    // be captured per-call by `cerulion clean` for structured
    // failure reporting. Idempotent: returns false if iceoryx2
    // already has a logger; we don't care which install wins
    // because both bridges have the same destination semantics.
    let _ = cerulion_core::iceoryx_logger::install_iceoryx2_tracing_bridge();

    match run(cli) {
        Ok(()) => ExitCode::SUCCESS,
        Err(e) => {
            eprintln!("Error: {}", e);
            ExitCode::FAILURE
        }
    }
}

fn run(cli: Cli) -> CliResult<()> {
    match cli.command {
        Commands::Workspace { action } => match action {
            WorkspaceAction::Create { name } => {
                let cwd = std::env::current_dir()?;
                let ws = workspace::workspace_create(&cwd, &name)?;
                println!("Created workspace at {}", ws.root.display());
                Ok(())
            }
            WorkspaceAction::Init { location } => {
                let path = PathBuf::from(&location)
                    .canonicalize()
                    .unwrap_or_else(|_| PathBuf::from(&location));
                let ws = workspace::workspace_init(&path)?;
                println!("Initialized workspace at {}", ws.root.display());
                Ok(())
            }
        },
        Commands::Node { action } => {
            let ws = discover_workspace()?;
            match action {
                NodeAction::Create {
                    node_type,
                    output,
                    input,
                    trigger_input,
                    policy,
                    raw_ffi,
                } => {
                    // Reject multi-invocation BEFORE any
                    // `parse_port_args` call. With `num_args = 2`
                    // on each flag, a valid single invocation
                    // produces exactly 2 elements; >2 means the
                    // user passed the flag multiple times.
                    // `parse_port_args` has a debug-assert on
                    // `args.len() == 2`, so guarding here lets the
                    // user see a clean validation error instead of
                    // an assertion panic in debug builds.
                    if trigger_input.len() > 2 {
                        return Err(cerulion_cli_engine::error::CliError::Validation(
                            "at most one `-T` per node; for sync over multiple inputs use \
                             `-i` for each plus `--policy sync_window_ms=N`"
                                .to_string(),
                        ));
                    }
                    if input.len() > 2 {
                        return Err(cerulion_cli_engine::error::CliError::Validation(
                            "at most one `-i` per `node create` (use `node modify` to add \
                             more inputs after creation)"
                                .to_string(),
                        ));
                    }
                    if output.len() > 2 {
                        return Err(cerulion_cli_engine::error::CliError::Validation(
                            "at most one `-o` per `node create` (use `node modify` to add \
                             more outputs after creation)"
                                .to_string(),
                        ));
                    }
                    let outputs: Vec<(String, String)> = if output.is_empty() {
                        vec![]
                    } else {
                        vec![parse_port_args(&output)]
                    };
                    let regular_inputs: Vec<(String, String)> = if input.is_empty() {
                        vec![]
                    } else {
                        vec![parse_port_args(&input)]
                    };
                    let trigger_inputs: Vec<(String, String)> = if trigger_input.is_empty() {
                        vec![]
                    } else {
                        vec![parse_port_args(&trigger_input)]
                    };
                    let explicit_policy = match policy.as_deref() {
                        Some(spec) => Some(parse_policy_spec(spec)?),
                        None => None,
                    };
                    let resolved_policy = resolve_create_policy(
                        explicit_policy.as_ref(),
                        trigger_inputs.first(),
                        &regular_inputs,
                    )?;
                    check_dash_t_dash_i_name_collision(&trigger_inputs, &regular_inputs)?;
                    // Combined inputs: trigger inputs first (matches macro
                    // emission order convention).
                    let mut combined_inputs = trigger_inputs.clone();
                    combined_inputs.extend(regular_inputs.iter().cloned());
                    let trigger_name = trigger_inputs
                        .first()
                        .map(|(_, name)| name.clone())
                        .or_else(|| match resolved_policy.as_ref() {
                            Some(cerulion_core::MacroPolicy::DataTrigger { input_name }) => {
                                Some(input_name.clone())
                            }
                            _ => None,
                        });
                    let options = node_cmd::NodeCreateOptions {
                        outputs,
                        inputs: combined_inputs,
                        trigger: trigger_name,
                        raw_ffi,
                    };
                    node_cmd::node_create_with_options(
                        &ws.nodes_dir,
                        &ws.root.join("Cargo.toml"),
                        &node_type,
                        resolved_policy,
                        &options,
                    )?;
                    println!("Created node type '{}'", node_type);
                    Ok(())
                }
                NodeAction::Delete { node_type } => {
                    node_cmd::node_delete(&ws.nodes_dir, &ws.root.join("Cargo.toml"), &node_type)?;
                    println!("Deleted node type '{}'", node_type);
                    Ok(())
                }
                NodeAction::Modify {
                    node_type,
                    input,
                    trigger_input,
                    output,
                    policy,
                } => {
                    // Same raw-vec length check as Create. With
                    // `num_args = 2` on the args, each valid invocation
                    // contributes exactly 2 elements, so any vec longer
                    // than 2 means the user passed the flag multiple
                    // times.
                    if trigger_input.len() > 2 {
                        return Err(cerulion_cli_engine::error::CliError::Validation(
                            "at most one `-T` per call; for sync over multiple inputs use \
                             `-i` for each plus `--policy sync_window_ms=N`"
                                .to_string(),
                        ));
                    }
                    if input.len() > 2 {
                        return Err(cerulion_cli_engine::error::CliError::Validation(
                            "at most one `-i` per `node modify` call".to_string(),
                        ));
                    }
                    if output.len() > 2 {
                        return Err(cerulion_cli_engine::error::CliError::Validation(
                            "at most one `-o` per `node modify` call".to_string(),
                        ));
                    }
                    let regular_inputs: Vec<(String, String)> = if input.is_empty() {
                        vec![]
                    } else {
                        vec![parse_port_args(&input)]
                    };
                    let trigger_inputs: Vec<(String, String)> = if trigger_input.is_empty() {
                        vec![]
                    } else {
                        vec![parse_port_args(&trigger_input)]
                    };
                    let explicit_policy = match policy.as_deref() {
                        Some(spec) => Some(parse_policy_spec(spec)?),
                        None => None,
                    };
                    // `-T` and `--policy data_trigger=NAME` must agree
                    // when both are supplied.
                    let effective_policy: Option<cerulion_core::MacroPolicy> = match (
                        explicit_policy,
                        trigger_inputs.first(),
                    ) {
                        (Some(p), None) => Some(p),
                        (None, Some((_, name))) => Some(cerulion_core::MacroPolicy::DataTrigger {
                            input_name: name.clone(),
                        }),
                        (
                            Some(cerulion_core::MacroPolicy::DataTrigger { input_name }),
                            Some((_, t_name)),
                        ) => {
                            if input_name != *t_name {
                                return Err(cerulion_cli_engine::error::CliError::Validation(
                                    format!(
                                        "`--policy data_trigger={input_name}` and \
                                             `-T <SCHEMA> {t_name}` disagree on the trigger input"
                                    ),
                                ));
                            }
                            Some(cerulion_core::MacroPolicy::DataTrigger {
                                input_name: input_name.clone(),
                            })
                        }
                        (Some(_), Some(_)) => {
                            return Err(cerulion_cli_engine::error::CliError::Validation(
                                    "`-T` is shorthand for `--policy data_trigger=<NAME>`; supply one or the other, not both"
                                        .to_string(),
                                ));
                        }
                        (None, None) => None,
                    };
                    // For `data_trigger=NAME` the named input must be
                    // declared somewhere reachable: this call (`-i` /
                    // `-T`) or a prior modify.
                    if let Some(cerulion_core::MacroPolicy::DataTrigger { input_name }) =
                        effective_policy.as_ref()
                    {
                        let added_in_this_call = trigger_inputs
                            .iter()
                            .chain(regular_inputs.iter())
                            .any(|(_, n)| n == input_name);
                        if !added_in_this_call {
                            let metadata = node_cmd::node_info(&ws.nodes_dir, &node_type)?;
                            let exists = metadata.inputs.iter().any(|p| p.name == *input_name);
                            if !exists {
                                return Err(cerulion_cli_engine::error::CliError::Validation(
                                    format!(
                                        "`--policy data_trigger={input_name}` requires the input \
                                         to exist; add it via `-i SCHEMA {input_name}` (or \
                                         `-T SCHEMA {input_name}`) in the same call or in a \
                                         prior `node modify`"
                                    ),
                                ));
                            }
                        }
                    }
                    if !output.is_empty() {
                        let (schema, name) = parse_port_args(&output);
                        node_cmd::node_modify_add_port(
                            &ws.nodes_dir,
                            &node_type,
                            &name,
                            Some(&schema),
                            true,
                            false,
                        )?;
                        println!("Added output '{}' to '{}'", name, node_type);
                    }
                    let trigger_target_name = match effective_policy.as_ref() {
                        Some(cerulion_core::MacroPolicy::DataTrigger { input_name }) => {
                            Some(input_name.clone())
                        }
                        _ => None,
                    };
                    // Add `-T` first (so its trigger semantics are
                    // applied), then any `-i`.
                    for (schema, name) in trigger_inputs.iter().chain(regular_inputs.iter()) {
                        let mark_as_trigger = trigger_target_name
                            .as_deref()
                            .map(|t| t == name.as_str())
                            .unwrap_or(false);
                        node_cmd::node_modify_add_port(
                            &ws.nodes_dir,
                            &node_type,
                            name,
                            Some(schema),
                            false,
                            mark_as_trigger,
                        )?;
                        if mark_as_trigger {
                            println!(
                                "Added input '{}' to '{}' as the data trigger",
                                name, node_type
                            );
                        } else {
                            println!("Added input '{}' to '{}'", name, node_type);
                        }
                    }
                    // Non-DataTrigger policies require an explicit
                    // mutation to the macro args. DataTrigger is already
                    // applied by the field-level write above when the
                    // trigger input was added; if the trigger pointed at
                    // an existing input, we still need to flip the macro
                    // args (clear conflicting node-level args).
                    if let Some(spec) = effective_policy.as_ref() {
                        match spec {
                            cerulion_core::MacroPolicy::DataTrigger { input_name } => {
                                let added_in_this_call = trigger_inputs
                                    .iter()
                                    .chain(regular_inputs.iter())
                                    .any(|(_, n)| n == input_name);
                                if !added_in_this_call {
                                    // The input pre-existed; the field-
                                    // attr write didn't run, so we need
                                    // to flip it via add_port with
                                    // is_trigger=true on the existing
                                    // field. `node_modify_add_port`
                                    // refuses to add a duplicate, so use
                                    // a dedicated helper.
                                    node_cmd::node_modify_promote_input_to_trigger(
                                        &ws.nodes_dir,
                                        &node_type,
                                        input_name,
                                    )?;
                                    println!(
                                        "Promoted existing input '{}' to data trigger on '{}'",
                                        input_name, node_type
                                    );
                                }
                            }
                            cerulion_core::MacroPolicy::External => {
                                node_cmd::node_modify_ext_trigger(&ws.nodes_dir, &node_type, true)?;
                                println!("Set policy=external on '{}'", node_type);
                            }
                            cerulion_core::MacroPolicy::Period { period_ms } => {
                                node_cmd::node_modify_set_period(
                                    &ws.nodes_dir,
                                    &node_type,
                                    *period_ms,
                                )?;
                                println!("Set policy=period_ms={} on '{}'", period_ms, node_type);
                            }
                            cerulion_core::MacroPolicy::Sync { window_ms } => {
                                node_cmd::node_modify_set_sync(
                                    &ws.nodes_dir,
                                    &node_type,
                                    *window_ms,
                                )?;
                                println!(
                                    "Set policy=sync_window_ms={} on '{}'",
                                    window_ms, node_type
                                );
                            }
                            cerulion_core::MacroPolicy::UnboundedSync => {
                                node_cmd::node_modify_set_policy(
                                    &ws.nodes_dir,
                                    &node_type,
                                    &cerulion_core::MacroPolicy::UnboundedSync,
                                )?;
                                println!("Set policy=unbounded_sync on '{}'", node_type);
                            }
                        }
                    }
                    Ok(())
                }
                NodeAction::Build { node_type, release } => {
                    node_cmd::node_build(&ws.root, &node_type, release)?;
                    println!("Built '{}'", node_type);
                    Ok(())
                }
                NodeAction::Stage {
                    node_type,
                    id,
                    graph,
                    prefix: _,
                    input_binding,
                } => {
                    let graph_name = resolve_graph_name(&ws, graph)?;
                    let metadata = node_cmd::node_info(&ws.nodes_dir, &node_type)?;

                    let outputs: Vec<(String, Option<String>)> = metadata
                        .outputs
                        .iter()
                        .map(|p| (p.name.clone(), p.schema.clone()))
                        .collect();

                    let inputs: Vec<(String, String)> = parse_input_bindings(&input_binding);

                    let node_def =
                        graph_cmd::build_node_def(&node_type, id.as_deref(), &outputs, &inputs);

                    graph_cmd::node_stage(&ws.graphs_dir, &graph_name, node_def)?;
                    println!("Staged '{}' into graph '{}'", node_type, graph_name);
                    Ok(())
                }
                NodeAction::Run {
                    node_type,
                    prefix,
                    id,
                    release,
                    no_cpu_dma_lock,
                    no_monitor_wait,
                } => {
                    let running = setup_ctrlc_handler()?;
                    let prefix = prefix.unwrap_or_else(|| "standalone".to_string());
                    let metadata = node_cmd::node_info(&ws.nodes_dir, &node_type)?;

                    let outputs: Vec<(String, Option<String>)> = metadata
                        .outputs
                        .iter()
                        .map(|p| (p.name.clone(), p.schema.clone()))
                        .collect();

                    let node_def =
                        graph_cmd::build_node_def(&node_type, id.as_deref(), &outputs, &[]);

                    // Create temporary graph
                    let temp_graph = format!("__temp_{}", node_type);
                    graph_cmd::graph_create(&ws.graphs_dir, &temp_graph, Some(&prefix))?;
                    graph_cmd::node_stage(&ws.graphs_dir, &temp_graph, node_def)?;

                    let result = graph_cmd::graph_run(
                        &ws.root,
                        &ws.graphs_dir,
                        &temp_graph,
                        running,
                        graph_cmd::TimeSource::Real, // live RealClock default (matches `graph run`)
                        if no_cpu_dma_lock {
                            graph_cmd::CpuDmaLockMode::Disabled
                        } else {
                            graph_cmd::CpuDmaLockMode::Auto
                        },
                        if no_monitor_wait {
                            graph_cmd::MonitorWaitMode::Disabled
                        } else {
                            graph_cmd::MonitorWaitMode::Auto
                        },
                        true,    // skip validation for temporary single-node graphs
                        release, // honour `node run --release` (CER-193), mirroring `graph run`
                    );

                    // Clean up temporary graph
                    let _ =
                        std::fs::remove_file(ws.graphs_dir.join(format!("{}.yaml", temp_graph)));

                    result
                }
                NodeAction::List => {
                    let nodes = node_cmd::node_list(&ws.nodes_dir)?;
                    if nodes.is_empty() {
                        println!("No nodes found.");
                    } else {
                        println!("{:<20} {:<10} {:<10} POLICY", "TYPE", "INPUTS", "OUTPUTS");
                        for node in &nodes {
                            println!(
                                "{:<20} {:<10} {:<10} {}",
                                node.node_type,
                                node.inputs.len(),
                                node.outputs.len(),
                                format_policy_short(node.policy.as_ref())
                            );
                        }
                    }
                    Ok(())
                }
                NodeAction::Info { node_type } => {
                    let info = node_cmd::node_info(&ws.nodes_dir, &node_type)?;
                    println!("Node type: {}", info.node_type);
                    println!("Policy: {}", format_policy_short(info.policy.as_ref()));
                    if !info.inputs.is_empty() {
                        println!("Inputs:");
                        for port in &info.inputs {
                            println!(
                                "  {} {}",
                                port.name,
                                port.schema.as_deref().unwrap_or("(untyped)")
                            );
                        }
                    }
                    if !info.outputs.is_empty() {
                        println!("Outputs:");
                        for port in &info.outputs {
                            println!(
                                "  {} {}",
                                port.name,
                                port.schema.as_deref().unwrap_or("(untyped)")
                            );
                        }
                    }
                    Ok(())
                }
            }
        }
        Commands::Graph { action } => {
            let ws = discover_workspace()?;
            match action {
                GraphAction::Create { name, prefix } => {
                    graph_cmd::graph_create(&ws.graphs_dir, &name, prefix.as_deref())?;
                    println!("Created graph '{}'", name);
                    Ok(())
                }
                GraphAction::Run {
                    name,
                    time_source,
                    no_validate,
                    release,
                    no_cpu_dma_lock,
                    no_monitor_wait,
                } => {
                    let running = setup_ctrlc_handler()?;
                    graph_cmd::graph_run(
                        &ws.root,
                        &ws.graphs_dir,
                        &name,
                        running,
                        time_source.into(),
                        if no_cpu_dma_lock {
                            graph_cmd::CpuDmaLockMode::Disabled
                        } else {
                            graph_cmd::CpuDmaLockMode::Auto
                        },
                        if no_monitor_wait {
                            graph_cmd::MonitorWaitMode::Disabled
                        } else {
                            graph_cmd::MonitorWaitMode::Auto
                        },
                        no_validate,
                        release,
                    )
                }
                GraphAction::Validate { name, release } => {
                    let report = graph_cmd::graph_validate(&ws.root, &name, release)?;
                    println!("{}", report);
                    if report.all_passed() {
                        Ok(())
                    } else {
                        Err(cerulion_cli_engine::error::CliError::Validation(
                            "graph validation failed".to_string(),
                        ))
                    }
                }
                GraphAction::List => {
                    let graphs = graph_cmd::graph_list(&ws.graphs_dir)?;
                    if graphs.is_empty() {
                        println!("No graphs found.");
                    } else {
                        for name in &graphs {
                            println!("{}", name);
                        }
                    }
                    Ok(())
                }
            }
        }
        Commands::Topic { action } => match action {
            TopicAction::List => {
                let topics = topic_cmd::topic_list()?;
                if topics.is_empty() {
                    println!("No active topics.");
                } else {
                    println!("TOPIC");
                    for topic in &topics {
                        println!("{}", topic.name);
                    }
                }
                Ok(())
            }
            TopicAction::Info { topic } => {
                let info = topic_cmd::topic_info(&topic)?;
                println!("{}", info);
                Ok(())
            }
            TopicAction::Echo { topic } => {
                let running = setup_ctrlc_handler()?;
                topic_cmd::topic_echo(&topic, running, &mut std::io::stdout())
            }
            TopicAction::Hz { topic } => {
                let running = setup_ctrlc_handler()?;
                topic_cmd::topic_hz(&topic, running, &mut std::io::stdout())
            }
        },
        Commands::Tui => cerulion_cli_tui::run().map_err(|e| {
            cerulion_cli_engine::error::CliError::Validation(format!("TUI error: {}", e))
        }),
        Commands::Trace { action } => match action {
            TraceAction::Inspect {
                dir,
                filter,
                limit,
                reverse,
            } => trace_inspect(&dir, filter.as_deref(), limit, reverse),
        },
        Commands::Clean => clean_iceoryx2_state(),
        Commands::Schema { action } => {
            let ws = discover_workspace()?;
            match action {
                SchemaAction::Create { name } => {
                    schema_cmd::schema_create(&ws.schemas_dir, &name)?;
                    println!("Created schema '{}'", name);
                    Ok(())
                }
                SchemaAction::Delete { name } => {
                    schema_cmd::schema_delete(&ws.schemas_dir, &name)?;
                    println!("Deleted schema '{}'", name);
                    Ok(())
                }
                SchemaAction::Info { name } => {
                    // Try ROS2 .msg lookup first (for qualified names like sensor_msgs::Image)
                    let msg_dir = ws.root.join("../native_ros2_messages/msg");
                    if msg_dir.is_dir() {
                        if let Ok(info) = schema_cmd::msg_schema_info(&msg_dir, &name) {
                            print!("{}", info);
                            return Ok(());
                        }
                    }
                    // Fall back to workspace YAML schema
                    let info = schema_cmd::schema_info(&ws.schemas_dir, &name)?;
                    for entry in &info.entries {
                        println!("Schema: {}", entry.name);
                        println!("Description: {}", entry.description);
                        println!("Fields: {}", entry.field_count);
                        for field in &entry.fields {
                            // `type_display` carries a loud "(treated as
                            // nested schema reference)" marker for unknown
                            // type identifiers, so typos can't masquerade
                            // as primitives.
                            println!("  {}: {}", field.name, field.type_display);
                        }
                        println!("Hash: 0x{:016x}", entry.schema_hash);
                    }
                    Ok(())
                }
            }
        }
    }
}

fn discover_workspace() -> CliResult<CerulionWorkspace> {
    let cwd = std::env::current_dir()?;
    CerulionWorkspace::discover(&cwd)
}

/// Remove iceoryx2's on-disk bookkeeping for **dead** nodes only.
///
/// iceoryx2 0.9 exposes `Node::try_cleanup_dead_nodes(config)`
/// (renamed from 0.8's `cleanup_dead_nodes`), which
/// walks the node registry and removes the stale resources of any
/// node whose owning process has died. Live nodes are left alone.
/// A graph that no longer matches a previous run's service shape
/// (e.g. a renamed topic or an output moved to a new node) will
/// see the dead node's services swept away as soon as the dead
/// node is detected, without disturbing any other Cerulion
/// process running concurrently on the same host.
///
/// Both `cerulion clean` and the implicit cleanup at the top of
/// `graph_run` route through
/// `cerulion_cli_engine::ipc_cleanup::cleanup_dead_iceoryx2_nodes_with_diagnostics`.
fn clean_iceoryx2_state() -> CliResult<()> {
    let report = cerulion_cli_engine::ipc_cleanup::cleanup_dead_iceoryx2_nodes_with_diagnostics();
    if report.cleanups == 0 && report.failed_cleanups == 0 {
        println!("No dead iceoryx2 nodes found — nothing to clean.");
        return Ok(());
    }
    println!(
        "Cleaned {} dead iceoryx2 node(s); {} cleanup(s) failed.",
        report.cleanups, report.failed_cleanups
    );
    if report.failed_cleanups == 0 {
        return Ok(());
    }
    // Print per-cause breakdown when we have classified failures.
    if !report.failures_by_cause.is_empty() {
        println!("Failure breakdown:");
        for (cause, count) in &report.failures_by_cause {
            let remediation = match cause.as_str() {
                "permission denied" => {
                    "the dead node's resources belong to a different user; \
                     re-run as that user, or `chmod` the resources under `/tmp/iceoryx2/`"
                }
                "version mismatch" => {
                    "the dead node's on-disk state was produced by a different \
                     iceoryx2 version; live nodes' state is unaffected, but the \
                     dead-node remnants need a manual `rm -rf /tmp/iceoryx2/` to clear"
                }
                "lock contention" => {
                    "another process is concurrently cleaning the same dead node; \
                     usually self-resolves on retry"
                }
                "monitoring resource still in use" => {
                    "iceoryx2's monitoring layer holds a reference; should resolve \
                     after the holding process exits"
                }
                "iceoryx2 internal error" => {
                    "iceoryx2 reported an internal error; \
                     run with `IOX2_LOG_LEVEL=trace cerulion clean` for the underlying message"
                }
                _ => "see iceoryx2 logs for details",
            };
            println!("  {} × {} — {}", count, cause, remediation);
        }
    }
    if !report.unclassified.is_empty() {
        println!(
            "Unclassified failures ({}): run with `RUST_LOG=iceoryx2=trace cerulion clean` for raw causes",
            report.unclassified.len()
        );
    }
    Ok(())
}

/// Implement `cerulion trace inspect <dir>` — read JSON-Lines bag
/// files written by `cerulion_core::trace::BagWriter` and print a
/// human-readable timeline. Files are read in lexicographic order
/// (matches the `trace_XXXX.jsonl` sequential naming).
fn trace_inspect(
    dir: &str,
    filter: Option<&str>,
    limit: Option<usize>,
    reverse: bool,
) -> CliResult<()> {
    let dir_path = std::path::Path::new(dir);
    if !dir_path.is_dir() {
        return Err(cerulion_cli_engine::error::CliError::Validation(format!(
            "trace inspect: `{}` is not a directory",
            dir
        )));
    }
    // Collect bag files in lexicographic order.
    let mut bag_files: Vec<std::path::PathBuf> = std::fs::read_dir(dir_path)
        .map_err(|e| {
            cerulion_cli_engine::error::CliError::Validation(format!(
                "trace inspect: failed to read directory `{}`: {}",
                dir, e
            ))
        })?
        .flatten()
        .map(|e| e.path())
        .filter(|p| {
            p.file_name()
                .and_then(|n| n.to_str())
                .is_some_and(|n| n.starts_with("trace_") && n.ends_with(".jsonl"))
        })
        .collect();
    bag_files.sort();
    if bag_files.is_empty() {
        println!("No trace_*.jsonl files found in `{}`", dir);
        return Ok(());
    }

    // Collect all lines into memory — Regime A entries are tiny
    // (~80 B each) so even an hour of trace at 1000 Hz fits in
    // < 300 MB.
    let mut lines: Vec<String> = Vec::new();
    for file in &bag_files {
        let contents = std::fs::read_to_string(file).map_err(|e| {
            cerulion_cli_engine::error::CliError::Validation(format!(
                "trace inspect: failed to read `{}`: {}",
                file.display(),
                e
            ))
        })?;
        for line in contents.lines() {
            if let Some(topic_filter) = filter {
                // Cheap substring check; the JSON shape always
                // contains `"topic":"<value>"`.
                let needle = format!("\"topic\":\"{}\"", topic_filter);
                if !line.contains(&needle) {
                    continue;
                }
            }
            lines.push(line.to_string());
        }
    }

    if reverse {
        lines.reverse();
    }
    if let Some(n) = limit {
        lines.truncate(n);
    }

    // Format each line as a compact human-readable timeline entry.
    for line in &lines {
        match parse_jsonl_record(line) {
            Some((topic, seq, ts, schema)) => {
                println!("{} seq={} t={}ns schema=0x{:016X}", topic, seq, ts, schema);
            }
            None => {
                // Pass through the raw line — better than dropping
                // unrecognized records silently.
                println!("(malformed) {}", line);
            }
        }
    }

    println!();
    println!(
        "[{} record(s) across {} file(s){}]",
        lines.len(),
        bag_files.len(),
        if let Some(t) = filter {
            format!(" filtered by topic={:?}", t)
        } else {
            String::new()
        }
    );
    Ok(())
}

/// Minimal JSON-Lines parser for the BagWriter format
/// (`{"topic":"X","seq":N,"ts_ns":N,"schema_hash":"0xH"}`). Returns
/// `(topic, sequence, publish_time_ns, schema_hash)` on success.
fn parse_jsonl_record(line: &str) -> Option<(String, u64, u64, u64)> {
    let extract = |key: &str| -> Option<&str> {
        let pat = format!("\"{}\":", key);
        let start = line.find(&pat)? + pat.len();
        let rest = &line[start..];
        let end = if let Some(stripped) = rest.strip_prefix('"') {
            // String value: find the next unescaped quote after position 1.
            1 + stripped.find('"')?
        } else {
            // Number value: read until `,` or `}`.
            rest.find([',', '}'])?
        };
        Some(rest[..=end].trim_matches('"'))
    };
    let topic = extract("topic")?.to_string();
    let seq: u64 = extract("seq")?.parse().ok()?;
    let ts: u64 = extract("ts_ns")?.parse().ok()?;
    let schema_raw = extract("schema_hash")?;
    let schema = u64::from_str_radix(schema_raw.trim_start_matches("0x"), 16).ok()?;
    Some((topic, seq, ts, schema))
}

/// Render a [`cerulion_core::MacroPolicy`] as a short, human-readable label
/// for the `cerulion node list` and `cerulion node info` outputs.
/// `None` falls back to `"-"` so columns stay aligned across rows.
fn format_policy_short(spec: Option<&cerulion_core::MacroPolicy>) -> String {
    use cerulion_core::MacroPolicy;
    match spec {
        Some(MacroPolicy::Period { period_ms }) => format!("period {}ms", period_ms),
        Some(MacroPolicy::Sync { window_ms }) => format!("sync {}ms", window_ms),
        Some(MacroPolicy::UnboundedSync) => "sync ∞".to_string(),
        Some(MacroPolicy::External) => "external".to_string(),
        Some(MacroPolicy::DataTrigger { input_name }) => format!("trigger:{}", input_name),
        None => "-".to_string(),
    }
}

/// Reject duplicate port names across `-T` and `-i` on `node
/// create`. Passing `-T sensor_msgs/Image foo -i sensor_msgs/Imu
/// foo` would otherwise produce a `lib.rs` declaring `foo` twice
/// — invalid Rust caught only at `cargo build` with an opaque
/// "duplicate field" error.
///
/// `node_modify_add_port` already checks for this via
/// `metadata.has_input(port_name)`, and the engine's
/// `node_create_with_options` mirrors the gate too (per the
/// CLI/engine contract alignment rule) — this CLI helper is the
/// user-facing first line: a clear "drop either the `-T` or the
/// `-i`" message before the request ever reaches the engine.
fn check_dash_t_dash_i_name_collision(
    trigger_inputs: &[(String, String)],
    regular_inputs: &[(String, String)],
) -> CliResult<()> {
    use cerulion_cli_engine::error::CliError;
    if let Some((_, t_name)) = trigger_inputs.first() {
        if regular_inputs.iter().any(|(_, n)| n == t_name) {
            return Err(CliError::Validation(format!(
                "port name '{t_name}' appears in both `-T` (trigger) and \
                 `-i` (regular input). Each port name must be unique. \
                 Drop either the `-T` or the `-i` for this name."
            )));
        }
    }
    Ok(())
}

/// Resolve the trigger policy for `cerulion node create` from the
/// `--policy` flag, the `-T` flag, and the regular `-i` inputs.
///
/// Defaulting rules (when `--policy` is absent):
/// - `-T` set → `DataTrigger { input_name: <T's name> }`
///   (an explicit `-T` declares the trigger, so the policy is
///   threaded through as `data_trigger=NAME`).
/// - 0 inputs (no `-i` and no `-T`) → error: a source-only node
///   must declare a non-data policy explicitly
///   (`--policy period_ms=N` or `--policy external`).
/// - 1+ inputs via `-i` only → `None` (the source emits no
///   node-level policy attribute; the runtime fires on any input
///   arrival and emits a warning at graph-build time so the
///   user notices). The warning is intentional: a user might want
///   a different policy (Sync, data_trigger on a
///   specific input), and silence would let unintended firing
///   behavior ship.
///
/// When `--policy` is present:
/// - `--policy data_trigger=NAME` and `-T NAME'` must agree on the
///   trigger input.
/// - `--policy data_trigger=NAME` requires NAME to match exactly
///   one of the `-i` or `-T` inputs.
/// - Non-data policies (Period/Sync/External) ignore the
///   `-T` / `-i` set and use the explicit policy as-is. `-T` is
///   incompatible with non-data policies — the caller errors.
fn resolve_create_policy(
    explicit: Option<&cerulion_core::MacroPolicy>,
    trigger_input: Option<&(String, String)>,
    regular_inputs: &[(String, String)],
) -> CliResult<Option<cerulion_core::MacroPolicy>> {
    use cerulion_cli_engine::error::CliError;
    use cerulion_core::MacroPolicy;
    match (explicit, trigger_input) {
        (Some(MacroPolicy::DataTrigger { input_name }), Some((_, t_name))) => {
            if input_name != t_name {
                return Err(CliError::Validation(format!(
                    "`--policy data_trigger={input_name}` and `-T <SCHEMA> {t_name}` disagree \
                     on the trigger input"
                )));
            }
            Ok(Some(MacroPolicy::DataTrigger {
                input_name: input_name.clone(),
            }))
        }
        (Some(_non_data), Some(_)) => Err(CliError::Validation(
            "`-T` declares a data-trigger input, which conflicts with a non-data `--policy`. \
             Drop `-T` or change the policy."
                .to_string(),
        )),
        (Some(MacroPolicy::DataTrigger { input_name }), None) => {
            let matches_input = regular_inputs.iter().any(|(_, n)| n == input_name);
            if !matches_input {
                return Err(CliError::Validation(format!(
                    "`--policy data_trigger={input_name}` requires `-i SCHEMA {input_name}` \
                     (or `-T SCHEMA {input_name}`) to declare the trigger input"
                )));
            }
            Ok(Some(MacroPolicy::DataTrigger {
                input_name: input_name.clone(),
            }))
        }
        (Some(p), None) => Ok(Some(p.clone())),
        (None, Some((_, name))) => Ok(Some(MacroPolicy::DataTrigger {
            input_name: name.clone(),
        })),
        (None, None) => {
            if regular_inputs.is_empty() {
                Err(CliError::Validation(
                    "source-only nodes (no `-i` or `-T`) must declare a non-data trigger policy. \
                     Pass `--policy period_ms=N` or `--policy external`."
                        .to_string(),
                ))
            } else {
                // 1+ inputs with no explicit `--policy` or `-T`:
                // emit no node-level policy attribute. The runtime
                // fires on any input arrival and emits a warning
                // at graph-build time so the user notices and can
                // pick a more specific policy if desired.
                Ok(None)
            }
        }
    }
}

/// Parse a `--policy <SPEC>` argument into a [`cerulion_core::MacroPolicy`].
///
/// Accepted forms:
/// - `period_ms=N` → `Period { period_ms: N }`
/// - `sync_window_ms=N` → `Sync { window_ms: N }`
/// - `external` → `External`
/// - `data_trigger=NAME` or `trigger=NAME` → `DataTrigger { input_name: NAME }`
///
/// Bare `period_ms` / `sync_window_ms` (no `=N`) are
/// rejected with a "requires a value" error. Zero values for either of
/// the two time-based forms (`period_ms`, `sync_window_ms`) are rejected
/// with a "must be > 0" error since they would spin-loop the scheduler.
/// (Was "three" before CER-398 chunk 14 removed the `deadline_ms` trigger.)
fn parse_policy_spec(spec: &str) -> CliResult<cerulion_core::MacroPolicy> {
    use cerulion_cli_engine::error::CliError;
    use cerulion_core::MacroPolicy;
    let trimmed = spec.trim();
    if trimmed.eq_ignore_ascii_case("external") {
        return Ok(MacroPolicy::External);
    }
    let (key, value) = match trimmed.split_once('=') {
        Some((k, v)) => (k.trim(), Some(v.trim())),
        None => (trimmed, None),
    };
    let parse_u64_positive = |label: &str, v: Option<&str>| -> CliResult<u64> {
        let v = v.ok_or_else(|| {
            CliError::Validation(format!(
                "`--policy {label}` requires a value, e.g. `--policy {label}=100`"
            ))
        })?;
        let parsed = v.parse::<u64>().map_err(|_| {
            CliError::Validation(format!(
                "`--policy {label}={v}` — value must be a positive integer (milliseconds)"
            ))
        })?;
        // Zero-duration trigger policies would spin-loop the
        // scheduler. The macro's `validate_policy_bounds` rejects
        // them at compile time as a defence-in-depth — but the CLI
        // shouldn't synthesize the zero-valued attribute in the
        // first place, so users get an immediate clear error
        // instead of a downstream `cargo build` failure.
        if parsed == 0 {
            return Err(CliError::Validation(format!(
                "`--policy {label}=0` — value must be > 0 (zero-duration {label} would spin-loop the scheduler)"
            )));
        }
        Ok(parsed)
    };
    let parse_name = |label: &str, v: Option<&str>| -> CliResult<String> {
        let v = v.ok_or_else(|| {
            CliError::Validation(format!(
                "`--policy {label}` requires a value, e.g. `--policy {label}=count`"
            ))
        })?;
        if v.is_empty() {
            return Err(CliError::Validation(format!(
                "`--policy {label}=` — value (input name) must not be empty"
            )));
        }
        Ok(v.to_string())
    };
    match key {
        "period_ms" => Ok(MacroPolicy::Period {
            period_ms: parse_u64_positive("period_ms", value)?,
        }),
        "sync_window_ms" => Ok(MacroPolicy::Sync {
            window_ms: parse_u64_positive("sync_window_ms", value)?,
        }),
        "data_trigger" | "trigger" => Ok(MacroPolicy::DataTrigger {
            input_name: parse_name(key, value)?,
        }),
        other => Err(CliError::Validation(format!(
            "unknown policy spec `{other}`. Accepted: \
             `period_ms=N`, `sync_window_ms=N`, `external`, \
             `data_trigger=NAME` (or `trigger=NAME`)"
        ))),
    }
}

fn setup_ctrlc_handler() -> CliResult<Arc<AtomicBool>> {
    let running = Arc::new(AtomicBool::new(true));
    let r = running.clone();
    ctrlc::set_handler(move || {
        r.store(false, Ordering::Relaxed);
    })
    .map_err(|e| {
        cerulion_cli_engine::error::CliError::Validation(format!(
            "failed to set Ctrl+C handler: {}",
            e
        ))
    })?;
    Ok(running)
}

fn init_logging(verbose: bool) {
    cerulion_core::init_logging(verbose);
}

/// Parse port args: \[SCHEMA, NAME\] or \[SCHEMA\] (derive name from schema).
///
/// The CLI accepts both `sensor_msgs/Image` (slash form, the
/// canonical form `USER_API.md` documents) and `sensor_msgs::Image`
/// (Rust-style colon form). We canonicalize to the slash form
/// right here at the input boundary so every downstream consumer
/// (templates codegen, the source code, graph YAML staging) sees
/// one shape.
fn parse_port_args(args: &[String]) -> (String, String) {
    // Clap enforces `num_args = 2` on the flags, so a single
    // invocation gives exactly 2 elements. The caller must check
    // for multi-invocation (4+ elements) BEFORE calling this
    // function. We defensively assert here so a future caller
    // that forgets the guard doesn't silently truncate.
    debug_assert_eq!(
        args.len(),
        2,
        "parse_port_args expects exactly 2 args; caller must reject multi-invocation \
         (`args.len() > 2`) before calling"
    );
    let schema = cerulion_cli_engine::schema_cmd::normalize_schema(&args[0]);
    let name = args.get(1).cloned().unwrap_or_else(|| {
        // Defensive fallback for release builds where the
        // `debug_assert_eq!` is compiled out and clap somehow
        // delivered only 1 arg: derive name from schema.
        schema.rsplit('/').next().unwrap_or(&schema).to_lowercase()
    });
    (schema, name)
}

/// Parse input binding pairs from flat arg list.
fn parse_input_bindings(args: &[String]) -> Vec<(String, String)> {
    args.chunks(2)
        .filter_map(|chunk| {
            if chunk.len() == 2 {
                Some((chunk[0].clone(), chunk[1].clone()))
            } else {
                None
            }
        })
        .collect()
}

/// Resolve graph name: explicit, auto-select if one exists, or error.
fn resolve_graph_name(ws: &CerulionWorkspace, explicit: Option<String>) -> CliResult<String> {
    if let Some(name) = explicit {
        return Ok(name);
    }

    // Auto-select if exactly one graph exists
    let graphs: Vec<_> = std::fs::read_dir(&ws.graphs_dir)?
        .filter_map(|e| e.ok())
        .filter(|e| {
            e.path()
                .extension()
                .map(|ext| ext == "yaml")
                .unwrap_or(false)
        })
        .collect();

    match graphs.len() {
        0 => Err(cerulion_cli_engine::error::CliError::Validation(
            "no graphs found — create one with `cerulion graph create <name>`".to_string(),
        )),
        1 => {
            let name = graphs[0]
                .path()
                .file_stem()
                .unwrap()
                .to_string_lossy()
                .to_string();
            Ok(name)
        }
        _ => Err(cerulion_cli_engine::error::CliError::Validation(
            "multiple graphs found — specify with -g <name>".to_string(),
        )),
    }
}

#[cfg(test)]
mod parse_port_args_tests {
    use super::*;

    #[test]
    fn parse_port_args_canonicalizes_colon_form() {
        // Colon-form schema gets canonicalised to slash form.
        let args = vec!["sensor_msgs::Image".to_string(), "image".to_string()];
        assert_eq!(
            parse_port_args(&args),
            ("sensor_msgs/Image".to_string(), "image".to_string()),
            "colon-form input must be canonicalised to slash form"
        );
    }

    #[test]
    fn parse_port_args_passes_slash_form() {
        let args = vec!["sensor_msgs/Image".to_string(), "image".to_string()];
        assert_eq!(
            parse_port_args(&args),
            ("sensor_msgs/Image".to_string(), "image".to_string()),
            "slash-form input must round-trip unchanged"
        );
    }

    #[test]
    fn parse_port_args_preserves_explicit_name() {
        let args = vec!["sensor_msgs::Image".to_string(), "frame".to_string()];
        assert_eq!(
            parse_port_args(&args),
            ("sensor_msgs/Image".to_string(), "frame".to_string()),
            "explicit NAME arg must be preserved verbatim"
        );
    }

    #[test]
    fn parse_port_args_unqualified_passes_through() {
        // Unqualified schemas (in-workspace types with no package
        // prefix) round-trip unchanged.
        let args = vec!["MyType".to_string(), "mytype".to_string()];
        assert_eq!(
            parse_port_args(&args),
            ("MyType".to_string(), "mytype".to_string())
        );
    }
}

#[cfg(test)]
mod parse_policy_spec_tests {
    use super::*;
    use cerulion_core::MacroPolicy;

    #[test]
    fn period_ms_bare_form_errors() {
        // Bare `--policy period_ms` (no value) is rejected: a
        // value-less time-based policy has no sensible default,
        // and silently picking 100ms would mask user mistakes
        // (e.g. forgetting the `=N` suffix). The user must pass
        // a concrete value, e.g. `--policy period_ms=100`.
        let err = parse_policy_spec("period_ms").unwrap_err();
        assert!(err.to_string().contains("requires a value"), "got: {err}");
    }

    #[test]
    fn period_ms_with_value_parses_normally() {
        let result = parse_policy_spec("period_ms=33").unwrap();
        assert_eq!(result, MacroPolicy::Period { period_ms: 33 });
    }

    #[test]
    fn deadline_ms_policy_removed() {
        // CER-398 chunk 14: the node-level `deadline_ms` trigger was
        // removed (decomposed into a `Data` trigger + the per-input
        // `expect_within_ms` QoS knob). The CLI no longer accepts it as
        // a `--policy` spec — both the `=N` and bare forms fall through
        // to the unknown-policy error.
        for spec in ["deadline_ms=100", "deadline_ms"] {
            let err = parse_policy_spec(spec).unwrap_err();
            assert!(
                err.to_string()
                    .contains("unknown policy spec `deadline_ms`"),
                "got: {err}"
            );
        }
    }

    #[test]
    fn sync_window_ms_bare_form_errors() {
        let err = parse_policy_spec("sync_window_ms").unwrap_err();
        assert!(err.to_string().contains("requires a value"), "got: {err}");
    }

    #[test]
    fn period_ms_zero_errors() {
        // `--policy period_ms=0` would spin-loop the scheduler.
        // Reject at the CLI surface so users get an immediate
        // clear error rather than a downstream `cargo build`
        // failure from `validate_policy_bounds`.
        let err = parse_policy_spec("period_ms=0").unwrap_err();
        assert!(err.to_string().contains("must be > 0"), "got: {err}");
    }

    #[test]
    fn sync_window_ms_zero_errors() {
        let err = parse_policy_spec("sync_window_ms=0").unwrap_err();
        assert!(err.to_string().contains("must be > 0"), "got: {err}");
    }

    #[test]
    fn default_keyword_is_rejected() {
        // `--policy default` used to be an alias for
        // `period_ms=100`, but that was misleading — the real
        // "default" behavior depends on input count (see
        // `resolve_create_policy`). The keyword was removed
        // entirely; users wanting a 100ms period pass
        // `--policy period_ms=100`.
        let err = parse_policy_spec("default").unwrap_err();
        assert!(
            err.to_string().contains("unknown policy spec"),
            "got: {err}"
        );
    }

    #[test]
    fn external_keyword_round_trips() {
        let result = parse_policy_spec("external").unwrap();
        assert_eq!(result, MacroPolicy::External);
    }
}

#[cfg(test)]
mod resolve_create_policy_tests {
    use super::*;
    use cerulion_core::MacroPolicy;

    fn input(name: &str) -> (String, String) {
        ("test_msgs/Foo".to_string(), name.to_string())
    }

    #[test]
    fn zero_inputs_no_policy_errors() {
        // Source-only node with no explicit policy must error —
        // there is no defensible default for a node with no
        // inputs and no time-based trigger.
        let err = resolve_create_policy(None, None, &[]).unwrap_err();
        assert!(err.to_string().contains("source-only nodes"), "got: {err}");
    }

    #[test]
    fn one_input_no_policy_returns_none() {
        // Per the unified defaulting rule: 1+ inputs with no
        // `--policy` returns None (no node-level policy attr is
        // written; the runtime warns + fires on any input).
        // No more silent auto-promotion to `data_trigger=<input>`.
        let inputs = vec![input("data")];
        let result = resolve_create_policy(None, None, &inputs).unwrap();
        assert_eq!(result, None);
    }

    #[test]
    fn two_inputs_no_policy_returns_none() {
        // 2+ inputs without `--policy` already returned None
        // before the unification — the test pins that down.
        let inputs = vec![input("a"), input("b")];
        let result = resolve_create_policy(None, None, &inputs).unwrap();
        assert_eq!(result, None);
    }

    #[test]
    fn explicit_dash_t_still_writes_data_trigger() {
        // An explicit `-T` is an explicit policy declaration, so
        // the policy is threaded through even when `--policy` is
        // omitted. Only the IMPLICIT 1-input auto-promotion went
        // away — `-T` retains its meaning.
        let trigger = input("scan");
        let result = resolve_create_policy(None, Some(&trigger), &[]).unwrap();
        assert_eq!(
            result,
            Some(MacroPolicy::DataTrigger {
                input_name: "scan".to_string(),
            })
        );
    }

    #[test]
    fn explicit_period_ms_unaffected_by_inputs() {
        // A `--policy period_ms=N` survives any number of inputs.
        let explicit = MacroPolicy::Period { period_ms: 50 };
        let inputs = vec![input("a")];
        let result = resolve_create_policy(Some(&explicit), None, &inputs).unwrap();
        assert_eq!(result, Some(MacroPolicy::Period { period_ms: 50 }));
    }
}

#[cfg(test)]
mod dash_t_dash_i_collision_tests {
    use super::*;

    fn port(schema: &str, name: &str) -> (String, String) {
        (schema.to_string(), name.to_string())
    }

    #[test]
    fn no_trigger_no_collision() {
        // No `-T` declared → guard is a no-op regardless of `-i`s.
        let result = check_dash_t_dash_i_name_collision(&[], &[port("sensor_msgs/Image", "image")]);
        assert!(result.is_ok());
    }

    #[test]
    fn trigger_alone_no_collision() {
        // `-T` declared, no `-i` → no possible collision.
        let result = check_dash_t_dash_i_name_collision(&[port("sensor_msgs/Image", "image")], &[]);
        assert!(result.is_ok());
    }

    #[test]
    fn trigger_and_regular_with_different_names_ok() {
        // `-T scan` + `-i image` → different names, accepted.
        let result = check_dash_t_dash_i_name_collision(
            &[port("sensor_msgs/LaserScan", "scan")],
            &[port("sensor_msgs/Image", "image")],
        );
        assert!(result.is_ok());
    }

    #[test]
    fn trigger_and_regular_with_same_name_rejected() {
        // `-T foo` + `-i foo` → name collision, must error.
        let err = check_dash_t_dash_i_name_collision(
            &[port("sensor_msgs/Image", "foo")],
            &[port("sensor_msgs/Imu", "foo")],
        )
        .expect_err("must reject name collision");
        let msg = err.to_string();
        assert!(
            msg.contains("'foo'"),
            "error must name the colliding port; got: {msg}"
        );
        assert!(
            msg.contains("-T") && msg.contains("-i"),
            "error must mention both flags; got: {msg}"
        );
    }

    #[test]
    fn trigger_collides_with_one_of_many_regulars() {
        // `-T foo` + `-i bar -i foo -i baz` (CLI doesn't allow multi
        // `-i` post-CER-377, but the helper is shape-agnostic and
        // should still catch a collision in a multi-element slice).
        let err = check_dash_t_dash_i_name_collision(
            &[port("sensor_msgs/Image", "foo")],
            &[
                port("sensor_msgs/Image", "bar"),
                port("sensor_msgs/Imu", "foo"),
                port("std_msgs/String", "baz"),
            ],
        )
        .expect_err("must reject collision even when not first");
        assert!(err.to_string().contains("'foo'"));
    }
}

#[cfg(test)]
mod clap_parse_tests {
    //! End-to-end clap parsing tests for the post-CER-377 surface.
    //!
    //! These tests exercise the actual `clap::Parser::try_parse_from`
    //! path — proving the `num_args = 2` invariants documented on
    //! `-T`, `-i`, `-o` hold against the binary's argv-style input.
    //!
    //! They do NOT actually create files, so they're fast and don't
    //! touch the filesystem. They only verify clap accepts/rejects
    //! the args and that the parsed shape (Vec lengths) is what the
    //! handler expects.
    use super::cli::{Cli, Commands, GraphAction, NodeAction};
    use clap::Parser;

    fn try_parse(argv: &[&str]) -> Result<Cli, clap::Error> {
        Cli::try_parse_from(argv)
    }

    #[test]
    fn graph_validate_release_flag_parses() {
        let cli = try_parse(&["cerulion", "graph", "validate", "perception", "--release"])
            .expect("`graph validate --release` must parse");
        let Commands::Graph {
            action: GraphAction::Validate { name, release },
        } = cli.command
        else {
            panic!("expected GraphAction::Validate");
        };
        assert_eq!(name, "perception");
        assert!(release);
    }

    #[test]
    fn graph_validate_defaults_to_debug_preference() {
        let cli = try_parse(&["cerulion", "graph", "validate", "perception"])
            .expect("`graph validate` without `--release` must parse");
        let Commands::Graph {
            action: GraphAction::Validate { release, .. },
        } = cli.command
        else {
            panic!("expected GraphAction::Validate");
        };
        assert!(!release, "release preference must default to off");
    }

    #[test]
    fn node_run_release_flag_parses() {
        // `node run --release` must carry the release preference so it
        // can thread through to `graph_run` (CER-193), mirroring
        // `graph run --release`.
        let cli = try_parse(&["cerulion", "node", "run", "camera", "--release"])
            .expect("`node run --release` must parse");
        let Commands::Node {
            action: NodeAction::Run {
                node_type, release, ..
            },
        } = cli.command
        else {
            panic!("expected NodeAction::Run");
        };
        assert_eq!(node_type, "camera");
        assert!(release, "`--release` must set the release preference");
    }

    #[test]
    fn node_run_defaults_to_debug_preference() {
        let cli = try_parse(&["cerulion", "node", "run", "camera"])
            .expect("`node run` without `--release` must parse");
        let Commands::Node {
            action: NodeAction::Run { release, .. },
        } = cli.command
        else {
            panic!("expected NodeAction::Run");
        };
        assert!(!release, "release preference must default to off");
    }

    #[test]
    fn node_run_no_cpu_dma_lock_flag_parses() {
        // CER-440: `node run --no-cpu-dma-lock` must carry the opt-out so
        // single-node runs can disable the default-on CPU DMA / C-state
        // lock, mirroring `graph run --no-cpu-dma-lock`.
        let cli = try_parse(&["cerulion", "node", "run", "camera", "--no-cpu-dma-lock"])
            .expect("`node run --no-cpu-dma-lock` must parse");
        let Commands::Node {
            action:
                NodeAction::Run {
                    node_type,
                    no_cpu_dma_lock,
                    ..
                },
        } = cli.command
        else {
            panic!("expected NodeAction::Run");
        };
        assert_eq!(node_type, "camera");
        assert!(
            no_cpu_dma_lock,
            "`--no-cpu-dma-lock` must set no_cpu_dma_lock = true"
        );
    }

    #[test]
    fn node_run_no_cpu_dma_lock_defaults_false() {
        // CER-440: bare `node run` leaves the lock default-on (flag false).
        let cli = try_parse(&["cerulion", "node", "run", "camera"])
            .expect("`node run` without `--no-cpu-dma-lock` must parse");
        let Commands::Node {
            action: NodeAction::Run {
                no_cpu_dma_lock, ..
            },
        } = cli.command
        else {
            panic!("expected NodeAction::Run");
        };
        assert!(
            !no_cpu_dma_lock,
            "no_cpu_dma_lock must default to off (lock on for the live path)"
        );
    }

    #[test]
    fn node_run_no_monitor_wait_flag_parses() {
        // CER-443: `node run --no-monitor-wait` must carry the opt-out so
        // single-node runs can disable the default-on CPU monitor-wait park,
        // mirroring `graph run --no-monitor-wait`.
        let cli = try_parse(&["cerulion", "node", "run", "camera", "--no-monitor-wait"])
            .expect("`node run --no-monitor-wait` must parse");
        let Commands::Node {
            action:
                NodeAction::Run {
                    node_type,
                    no_monitor_wait,
                    ..
                },
        } = cli.command
        else {
            panic!("expected NodeAction::Run");
        };
        assert_eq!(node_type, "camera");
        assert!(
            no_monitor_wait,
            "`--no-monitor-wait` must set no_monitor_wait = true"
        );
    }

    #[test]
    fn node_run_no_monitor_wait_defaults_false() {
        // CER-443: bare `node run` leaves the park default-on (flag false).
        let cli = try_parse(&["cerulion", "node", "run", "camera"])
            .expect("`node run` without `--no-monitor-wait` must parse");
        let Commands::Node {
            action: NodeAction::Run {
                no_monitor_wait, ..
            },
        } = cli.command
        else {
            panic!("expected NodeAction::Run");
        };
        assert!(
            !no_monitor_wait,
            "no_monitor_wait must default to off (park on for the live path)"
        );
    }

    #[test]
    fn dash_t_with_schema_and_name_parses() {
        let cli = try_parse(&[
            "cerulion",
            "node",
            "create",
            "detector",
            "-T",
            "sensor_msgs/Image",
            "image",
        ])
        .expect("`-T SCHEMA NAME` must parse");
        let Commands::Node { action } = cli.command else {
            panic!("expected Node command");
        };
        let NodeAction::Create { trigger_input, .. } = action else {
            panic!("expected Create action");
        };
        assert_eq!(trigger_input, vec!["sensor_msgs/Image", "image"]);
    }

    #[test]
    fn dash_t_with_only_schema_is_rejected_by_clap() {
        // num_args = 2 means clap rejects single-arg invocations
        // BEFORE the handler runs. The pre-CER-377 `num_args = 1..=2`
        // ambiguity is gone.
        // `Cli` doesn't impl `Debug`, so `.expect_err()` doesn't
        // compile — match on the Result directly.
        let result = try_parse(&[
            "cerulion",
            "node",
            "create",
            "detector",
            "-T",
            "sensor_msgs/Image",
        ]);
        let err = match result {
            Ok(_) => panic!("`-T SCHEMA` alone must be rejected by clap"),
            Err(e) => e,
        };
        // clap's error message for missing arg counts mentions the value name.
        let msg = err.to_string();
        assert!(
            msg.contains("NAME") || msg.contains("required") || msg.contains("argument"),
            "expected clap to flag the missing `NAME` arg; got: {msg}"
        );
    }

    #[test]
    fn dash_i_with_schema_and_name_parses() {
        let cli = try_parse(&[
            "cerulion",
            "node",
            "create",
            "camera",
            "-i",
            "sensor_msgs/LaserScan",
            "scan",
        ])
        .expect("`-i SCHEMA NAME` must parse");
        let Commands::Node {
            action: NodeAction::Create { input, .. },
        } = cli.command
        else {
            panic!("expected NodeAction::Create");
        };
        assert_eq!(input, vec!["sensor_msgs/LaserScan", "scan"]);
    }

    #[test]
    fn multiple_dash_t_flat_vec_caught_by_length_guard() {
        // `-T A nameA -T B nameB` flat-Vecs to len=4. The CLI
        // handler's length guard (`trigger_input.len() > 2`) catches
        // this; here we just verify clap parses it into the flat-Vec
        // shape we rely on.
        let cli = try_parse(&[
            "cerulion",
            "node",
            "create",
            "fusion",
            "-T",
            "sensor_msgs/Image",
            "image",
            "-T",
            "sensor_msgs/Imu",
            "imu",
        ])
        .expect(
            "clap must accept the flat-Vec accumulation; length guard runs later in the handler",
        );
        let Commands::Node {
            action: NodeAction::Create { trigger_input, .. },
        } = cli.command
        else {
            panic!("expected NodeAction::Create");
        };
        assert_eq!(
            trigger_input.len(),
            4,
            "expected 4 args from two `-T` invocations"
        );
        assert_eq!(
            trigger_input,
            vec!["sensor_msgs/Image", "image", "sensor_msgs/Imu", "imu"]
        );
    }

    #[test]
    fn dash_o_with_schema_and_name_parses() {
        let cli = try_parse(&[
            "cerulion",
            "node",
            "create",
            "camera",
            "-o",
            "sensor_msgs/Image",
            "image",
        ])
        .expect("`-o SCHEMA NAME` must parse");
        let Commands::Node {
            action: NodeAction::Create { output, .. },
        } = cli.command
        else {
            panic!("expected NodeAction::Create");
        };
        assert_eq!(output, vec!["sensor_msgs/Image", "image"]);
    }

    #[test]
    fn raw_ffi_flag_parses() {
        let cli = try_parse(&[
            "cerulion",
            "node",
            "create",
            "camera",
            "--raw-ffi",
            "-o",
            "sensor_msgs/Image",
            "image",
        ])
        .expect("`--raw-ffi` must parse alongside `-o`");
        let Commands::Node {
            action: NodeAction::Create { raw_ffi, .. },
        } = cli.command
        else {
            panic!("expected NodeAction::Create");
        };
        assert!(raw_ffi);
    }

    #[test]
    fn policy_period_ms_parses() {
        let cli = try_parse(&[
            "cerulion",
            "node",
            "create",
            "camera",
            "--policy",
            "period_ms=33",
            "-o",
            "sensor_msgs/Image",
            "image",
        ])
        .expect("`--policy period_ms=33` must parse");
        let Commands::Node {
            action: NodeAction::Create { policy, .. },
        } = cli.command
        else {
            panic!("expected NodeAction::Create");
        };
        assert_eq!(policy.as_deref(), Some("period_ms=33"));
    }
}