mecha10_core/

node.rs

use crate::{Context, HealthStatus, NodeState, Result};
use async_trait::async_trait;
use schemars::JsonSchema;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use tracing::{debug, error, info, warn};

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum HealthPriority {
    Critical,

    #[default]
    Normal,

    Background,
}

impl HealthPriority {
    /// Per-variant health-monitoring profile: (heartbeat interval, Redis TTL in seconds,
    /// unresponsive timeout). Previously `interval`/`ttl_seconds`/`timeout` each had their own
    /// 3-arm match over `Self`, which cargo-dupes flagged as duplicate match shapes; the
    /// per-variant values are only defined once here, and the public getters below are plain
    /// field projections.
    const fn profile(&self) -> (Duration, u64, Duration) {
        match self {
            Self::Critical => (Duration::from_secs(1), 5, Duration::from_secs(3)),
            Self::Normal => (Duration::from_secs(5), 60, Duration::from_secs(15)),
            Self::Background => (Duration::from_secs(30), 120, Duration::from_secs(90)),
        }
    }

    /// Get the heartbeat interval for this priority level
    pub const fn interval(&self) -> Duration {
        self.profile().0
    }

    /// Get the Redis TTL (time-to-live) in seconds for this priority level
    ///
    /// The TTL should be longer than the interval to allow for some jitter,
    /// but short enough that stale data expires quickly.
    pub const fn ttl_seconds(&self) -> u64 {
        self.profile().1
    }

    /// Get the timeout threshold for monitoring
    ///
    /// A node is considered unresponsive if no health report is received
    /// within this duration.
    pub const fn timeout(&self) -> Duration {
        self.profile().2
    }

    /// Get the timeout in microseconds (for comparison with timestamps)
    pub const fn timeout_micros(&self) -> u64 {
        self.timeout().as_micros() as u64
    }
}

impl std::fmt::Display for HealthPriority {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Critical => write!(f, "critical"),
            Self::Normal => write!(f, "normal"),
            Self::Background => write!(f, "background"),
        }
    }
}

#[derive(Debug, Clone)]
pub struct HealthReportingConfig {
    pub enabled: bool,

    pub priority: HealthPriority,

    pub store_history: bool,
}

impl Default for HealthReportingConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            priority: HealthPriority::Normal,
            store_history: true,
        }
    }
}

impl HealthReportingConfig {
    /// Shared constructor behind the `critical`/`background` presets, which are always
    /// enabled and only differ in priority and whether history is stored.
    fn with_priority(priority: HealthPriority, store_history: bool) -> Self {
        Self {
            enabled: true,
            priority,
            store_history,
        }
    }

    /// Create config for critical priority nodes (1s interval)
    pub fn critical() -> Self {
        Self::with_priority(HealthPriority::Critical, false)
    }

    /// Create config for normal priority nodes (5s interval)
    pub fn normal() -> Self {
        Self::default()
    }

    /// Create config for background priority nodes (30s interval)
    pub fn background() -> Self {
        Self::with_priority(HealthPriority::Background, true)
    }

    /// Get the effective interval based on priority
    pub fn interval(&self) -> Duration {
        self.priority.interval()
    }

    /// Get the effective TTL based on priority
    pub fn ttl_seconds(&self) -> u64 {
        self.priority.ttl_seconds()
    }
}

pub trait RateConfig {
    /// Get the update rate in Hertz (cycles per second)
    fn rate_hz(&self) -> f32;

    /// Create a tokio interval from the configured rate
    ///
    /// This is a convenience method that converts the rate to a Duration
    /// and creates a tokio::time::Interval.
    fn create_interval(&self) -> tokio::time::Interval {
        let duration = Duration::from_secs_f32(1.0 / self.rate_hz());
        tokio::time::interval(duration)
    }

    /// Get the period (time between updates) as a Duration
    fn period(&self) -> Duration {
        Duration::from_secs_f32(1.0 / self.rate_hz())
    }
}

pub trait HealthStatusExt {
    /// Create a healthy status
    fn healthy(message: impl Into<String>) -> Self;
    /// Create an unhealthy status
    fn unhealthy(message: impl Into<String>) -> Self;
}

impl HealthStatus {
    /// Shared constructor backing both [`HealthStatusExt::healthy`] and
    /// [`HealthStatusExt::unhealthy`] — the two only ever differed in the `healthy` flag and
    /// the default `failure_count` that goes with it.
    fn with_check(healthy: bool, failure_count: u32, message: impl Into<String>) -> Self {
        Self {
            healthy,
            last_check: crate::prelude::now_micros(),
            failure_count,
            message: Some(message.into()),
        }
    }
}

impl HealthStatusExt for HealthStatus {
    fn healthy(message: impl Into<String>) -> Self {
        Self::with_check(true, 0, message)
    }

    fn unhealthy(message: impl Into<String>) -> Self {
        Self::with_check(false, 1, message)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "kebab-case")]
pub enum NodeType {
    Imu,
    Camera,
    ImageClassifier,
    ObjectDetector,
    BehaviorExecutor,
    LlmCommand,
    Teleop,
    Motor,
    WebsocketBridge,
    VideoStreamer,
    Simulator,
    Speaker,
    Listener,
    Rl,
    InstanceTrack,
    Diagnostics,
    Servo,
    PanTilt,
}

impl NodeType {
    pub const ALL: &'static [NodeType] = &[
        NodeType::Imu,
        NodeType::Camera,
        NodeType::ImageClassifier,
        NodeType::ObjectDetector,
        NodeType::BehaviorExecutor,
        NodeType::LlmCommand,
        NodeType::Teleop,
        NodeType::Motor,
        NodeType::WebsocketBridge,
        NodeType::VideoStreamer,
        NodeType::Simulator,
        NodeType::Speaker,
        NodeType::Listener,
        NodeType::Rl,
        NodeType::InstanceTrack,
        NodeType::Diagnostics,
        NodeType::Servo,
        NodeType::PanTilt,
    ];

    /// The kebab-case wire identifier for this type, e.g. `"image-classifier"`.
    ///
    /// This is exactly the string `serde` would produce via
    /// `#[serde(rename_all = "kebab-case")]`, exposed as a plain `&str` for
    /// callers (like `mecha10-schema-registry`) that work with node-type
    /// identifiers as strings rather than JSON values.
    pub fn as_str(&self) -> &'static str {
        match self {
            NodeType::Imu => "imu",
            NodeType::Camera => "camera",
            NodeType::ImageClassifier => "image-classifier",
            NodeType::ObjectDetector => "object-detector",
            NodeType::BehaviorExecutor => "behavior-executor",
            NodeType::LlmCommand => "llm-command",
            NodeType::Teleop => "teleop",
            NodeType::Motor => "motor",
            NodeType::WebsocketBridge => "websocket-bridge",
            NodeType::VideoStreamer => "video-streamer",
            NodeType::Simulator => "simulator",
            NodeType::Speaker => "speaker",
            NodeType::Listener => "listener",
            NodeType::Rl => "rl",
            NodeType::InstanceTrack => "instance-track",
            NodeType::Diagnostics => "diagnostics",
            NodeType::Servo => "servo",
            NodeType::PanTilt => "pan-tilt",
        }
    }
}

impl std::fmt::Display for NodeType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.as_str())
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "kebab-case")]
pub enum NodeTag {
    Sensor,
    Inertial,
    Capture,
    Ai,
    Classifier,
    Detector,
    Vision,
    Planner,
    Control,
    Nlp,
    Input,
    Actuator,
    Motion,
    Relay,
    Websocket,
    Streamer,
    Video,
    Simulator,
    Logger,
}

#[async_trait]
pub trait NodeImpl: Send + Sync + Debug + 'static {
    type Config: DeserializeOwned + Send + Sync + Debug;

    const NODE_TYPE: Option<NodeType> = None;

    const NODE_TAGS: &'static [NodeTag] = &[];

    /// Crate name as it appears in Cargo.toml (e.g. `"mecha10-nodes-camera"`).
    ///
    /// `None` (the default) means this node isn't meant to be scaffolded via `mecha10 add` --
    /// e.g. test/example nodes, or any node that hasn't opted into registration yet. This is
    /// also the gate [`crate::register_node!`] checks: a node whose `CRATE_NAME` is `None`
    /// registers no [`crate::node_registry::NodeDescriptor`] at all, regardless of what its
    /// other registration-related consts are set to.
    const CRATE_NAME: Option<&'static str> = None;

    /// Human-readable one-line description of what this node does, surfaced by a catalog
    /// consumer (e.g. `mecha10 add <node>`'s listing). Empty by default.
    const DESCRIPTION: &'static str = "";

    /// Default `mecha10 add`/run target for this node (e.g. `"robot"` vs. a simulator-only
    /// target). Defaults to `"robot"`.
    const DEFAULT_RUN_TARGET: &'static str = "robot";

    /// Free-form catalog search tags (e.g. `&["camera", "video", "streaming"]`), distinct from
    /// `NODE_TAGS` (a `NodeTag` enum used for health-reporting capability tags). Surfaced by a
    /// catalog consumer (e.g. `mecha10 add <node>`'s search/filtering). Empty by default.
    const CATALOG_TAGS: &'static [&'static str] = &[];

    /// Initialize the node with the given configuration.
    ///
    /// This is called once when the node starts. Use this to:
    /// - Load configuration
    /// - Initialize resources (models, connections, etc.)
    /// - Validate prerequisites
    ///
    /// # Errors
    ///
    /// Return an error if initialization fails. The node will not start.
    async fn init(config: Self::Config) -> Result<Self>
    where
        Self: Sized;

    /// Main execution loop for the node.
    ///
    /// This is where your node's business logic lives. Typical patterns:
    /// - Subscribe to topics and process messages
    /// - Run periodic tasks
    /// - Publish results
    ///
    /// This method should run indefinitely until:
    /// - The context is cancelled (shutdown requested)
    /// - An unrecoverable error occurs
    ///
    /// # Errors
    ///
    /// Return an error if the node encounters an unrecoverable error.
    /// The node will transition to Error state and shutdown.
    async fn run(&mut self, ctx: &Context) -> Result<()>;

    /// Graceful shutdown hook.
    ///
    /// Called when the node is shutting down. Use this to:
    /// - Close connections
    /// - Flush buffers
    /// - Save state
    /// - Release resources
    ///
    /// The default implementation does nothing.
    ///
    /// # Errors
    ///
    /// Errors during shutdown are logged but don't prevent shutdown.
    async fn shutdown(&mut self) -> Result<()> {
        Ok(())
    }

    /// Health check for monitoring.
    ///
    /// Called periodically to check node health. Use this to:
    /// - Report current state
    /// - Check resource usage
    /// - Validate connections
    ///
    /// The default implementation returns healthy status.
    async fn health_check(&self) -> HealthStatus {
        HealthStatus {
            healthy: true,
            last_check: crate::prelude::now_micros(),
            failure_count: 0,
            message: Some("Node is running".to_string()),
        }
    }

    /// Extra diagnostic key/value pairs to attach to this node's health report (surfaced on
    /// the wire as [`crate::messages::system::HealthStatus::diagnostics`]).
    ///
    /// Separate from `health_check()`'s `HealthStatus` (which has a fixed shape shared by every
    /// node) so a node can report open-ended, node-specific data -- e.g. `video-streamer`
    /// reports which `camera_id`(s) it's actually streaming, since that's a per-camera identity
    /// distinct from this node's own `node_id`/`instance_id` and the dashboard needs it to pair
    /// WebRTC signaling correctly (see `packages/nodes/video-streamer/src/lib.rs`'s impl).
    ///
    /// The default implementation reports nothing.
    fn health_diagnostics(&self) -> Option<std::collections::HashMap<String, String>> {
        None
    }

    /// Node-specific pre-run content check.
    ///
    /// This is the "prerun" layer that catches a bad node config *before* the
    /// node's main loop starts, rather than crashing deep inside `run()` -- e.g.
    /// a camera device path that doesn't exist, a model file that was never
    /// downloaded, or two GPIO pins wired to the same physical pin.
    ///
    /// Called once, after `init()` has already produced a fully-typed (and, per
    /// [`crate::config::Validate`], schema/shape-validated) `Self`, and before
    /// `run()` starts. Unlike `Validate::validate()` -- which only sees the raw
    /// deserialized config and runs during config loading -- `preflight()` runs
    /// on the initialized node, so it can also check state derived during
    /// `init()` (e.g. a resolved file path) rather than just raw config fields.
    ///
    /// The default implementation does nothing, so opting in is non-breaking for
    /// every existing `NodeImpl`.
    ///
    /// # Errors
    ///
    /// Return an error with an actionable message (see `node-runner`'s
    /// `preflight.rs` for the established convention: a `❌` header naming what
    /// failed, followed by numbered remediation steps) if the node should not
    /// start. The node will not run.
    fn preflight(&self) -> Result<()> {
        Ok(())
    }
}

/// Formats a `Debug` impl for a node whose controller field is a trait
/// object that can't derive `Debug` (e.g. `Box<dyn SomeController>`).
///
/// Shared by node types that expose `config`, a controller identified by
/// `controller_name` (typically `self.controller.name()`), and an
/// `update_counter` field.
pub fn fmt_node_with_controller(
    f: &mut std::fmt::Formatter<'_>,
    struct_name: &str,
    config: &impl Debug,
    controller_name: &str,
    update_counter: u64,
) -> std::fmt::Result {
    f.debug_struct(struct_name)
        .field("config", config)
        .field("controller", &controller_name)
        .field("update_counter", &update_counter)
        .finish()
}

#[derive(Clone)]
pub struct Node<T: NodeImpl> {
    node: Arc<Mutex<T>>,
    state: Arc<Mutex<NodeState>>,
    // Snapshot of `T::health_diagnostics()` taken once right after `init()`, before `run()`
    // ever starts holding `node`'s lock for the rest of the process's life (see `start()` -
    // it locks `node` and never releases it until shutdown, so the health-reporting task's
    // `try_lock()` in `run_node` below always loses that race in practice and would otherwise
    // report `diagnostics: None` forever). Every current `health_diagnostics()` override
    // (currently just `VideoStreamerNode`'s, reporting its fixed `camera_ids`) only reflects
    // state that's already final by the time `init()` returns - not live run-loop state - so a
    // one-time post-init snapshot is exactly as accurate as a live read would be, without
    // needing to contend with `run()`'s lock at all.
    static_diagnostics: Arc<Option<std::collections::HashMap<String, String>>>,
}

impl<T: NodeImpl> Node<T> {
    /// Create a new node from a configuration.
    ///
    /// This calls `NodeImpl::init()` and wraps the result in a Node.
    ///
    /// # Errors
    ///
    /// Returns an error if initialization fails.
    pub async fn from_config(config: T::Config) -> Result<Self> {
        info!("Initializing node");

        let node = T::init(config).await?;
        let static_diagnostics = node.health_diagnostics();

        Ok(Self {
            node: Arc::new(Mutex::new(node)),
            state: Arc::new(Mutex::new(NodeState::Starting)),
            static_diagnostics: Arc::new(static_diagnostics),
        })
    }

    /// The post-init snapshot of `health_diagnostics()` - see the field doc comment on
    /// `static_diagnostics` for why this is safe to read without locking `node`.
    pub fn static_diagnostics(&self) -> Option<std::collections::HashMap<String, String>> {
        (*self.static_diagnostics).clone()
    }

    /// Start the node with the given context.
    ///
    /// This orchestrates the full lifecycle:
    /// 1. Run `preflight()` node-content checks
    /// 2. Transition to Running state
    /// 3. Call `run()`
    /// 4. Call `shutdown()` on completion or error
    /// 5. Transition to final state
    ///
    /// # Errors
    ///
    /// Returns an error if `preflight()` rejects the node's config, or if the
    /// node fails during execution.
    pub async fn start(&self, ctx: Context) -> Result<()> {
        {
            let node = self.node.lock().await;
            node.preflight()?;
        }

        {
            let mut state = self.state.lock().await;
            *state = NodeState::Running;
            info!("Node transitioned to Running state");
        }

        let result = {
            let mut node = self.node.lock().await;
            debug!("Starting node execution");
            node.run(&ctx).await
        };

        match result {
            Ok(()) => {
                info!("Node completed successfully");
                self.shutdown().await?;
            }
            Err(e) => {
                error!("Node failed with error: {}", e);
                {
                    let mut state = self.state.lock().await;
                    *state = NodeState::Crashed;
                }
                self.shutdown().await?;
                return Err(e);
            }
        }

        Ok(())
    }

    /// Gracefully shutdown the node.
    ///
    /// This calls `shutdown()` and transitions to Stopped state.
    pub async fn shutdown(&self) -> Result<()> {
        info!("Shutting down node");

        {
            let mut state = self.state.lock().await;
            *state = NodeState::Stopping;
        }

        let result = {
            let mut node = self.node.lock().await;
            node.shutdown().await
        };

        {
            let mut state = self.state.lock().await;
            *state = NodeState::Stopped;
            info!("Node stopped");
        }

        if let Err(e) = result {
            warn!("Error during shutdown: {}", e);
            return Err(e);
        }

        Ok(())
    }

    /// Get the current state of the node.
    pub async fn state(&self) -> NodeState {
        self.state.lock().await.clone()
    }

    /// Perform a health check on the node.
    pub async fn health_check(&self) -> HealthStatus {
        let node = self.node.lock().await;
        node.health_check().await
    }

    /// Check if the node is healthy.
    pub async fn is_healthy(&self) -> bool {
        matches!(self.state().await, NodeState::Running | NodeState::Starting)
    }
}

/// Run a node with the given configuration and context.
///
/// This is a convenience function that:
/// 1. Initializes the node from config
/// 2. Starts the node with the context
/// 3. Handles errors
///
/// # Arguments
///
/// * `config` - Node-specific configuration
/// * `ctx` - Execution context
/// * `health_config` - Health reporting configuration
///
/// # Example
///
/// ```rust,no_run
/// use mecha10_core::prelude::*;
/// use mecha10_core::context::ContextBuilder;
///
/// # #[derive(Debug)]
/// # struct MyNode;
/// # #[derive(Debug, Deserialize)]
/// # struct MyConfig { rate: f32 }
/// # #[async_trait]
/// # impl NodeImpl for MyNode {
/// #     type Config = MyConfig;
/// #     async fn init(_config: Self::Config) -> Result<Self> { Ok(Self) }
/// #     async fn run(&mut self, _ctx: &Context) -> Result<()> { Ok(()) }
/// # }
/// #[tokio::main]
/// async fn main() -> Result<()> {
///     let config = MyConfig { rate: 10.0 };
///
///     // Use default health reporting (Normal priority, 5s interval)
///     let ctx = ContextBuilder::new("my_node").build().await?;
///     run_node::<MyNode>(config, ctx, HealthReportingConfig::default()).await?;
///
///     // Critical priority for safety-critical nodes (1s interval)
///     let config = MyConfig { rate: 10.0 };
///     let ctx = ContextBuilder::new("my_node").build().await?;
///     run_node::<MyNode>(config, ctx, HealthReportingConfig::critical()).await?;
///
///     // Background priority for auxiliary nodes (30s interval)
///     let config = MyConfig { rate: 10.0 };
///     let ctx = ContextBuilder::new("my_node").build().await?;
///     run_node::<MyNode>(config, ctx, HealthReportingConfig::background()).await?;
///
///     Ok(())
/// }
/// ```
pub async fn run_node<T: NodeImpl>(
    config: T::Config,
    ctx: Context,
    health_config: HealthReportingConfig,
) -> Result<()> {
    crate::prelude::init_logging();

    #[cfg(feature = "messaging")]
    crate::log_publisher::attach(ctx.node_id(), ctx.clone());

    let node = Node::<T>::from_config(config).await?;

    #[cfg(feature = "messaging")]
    {
        if health_config.enabled {
            let node_for_health = node.node.clone();
            let static_diagnostics_for_health = node.static_diagnostics();
            let ctx_for_health = ctx.clone();
            let priority = health_config.priority;
            let report_interval = health_config.interval();
            let ttl_seconds = health_config.ttl_seconds();
            let store_history = health_config.store_history;

            tokio::spawn(async move {
                use crate::health::HealthReporter;

                let reporter = match HealthReporter::with_config_and_type_tags(
                    &ctx_for_health,
                    priority,
                    ttl_seconds,
                    store_history,
                    T::NODE_TYPE,
                    T::NODE_TAGS,
                )
                .await
                {
                    Ok(r) => r,
                    Err(e) => {
                        warn!("Health reporting disabled: {}", e);
                        return;
                    }
                };

                debug!(
                    "Health reporting enabled: priority={}, interval={:?}, ttl={}s, history={}",
                    priority, report_interval, ttl_seconds, store_history
                );

                let mut interval = tokio::time::interval(report_interval);

                loop {
                    interval.tick().await;
                    debug!("Health tick - checking node health");

                    let msg_status = match node_for_health.try_lock() {
                        Ok(node) => {
                            let status = node.health_check().await;
                            let diagnostics = node.health_diagnostics();
                            crate::messages::system::HealthStatus {
                                timestamp: status.last_check,
                                level: if status.healthy {
                                    crate::messages::system::HealthLevel::Ok
                                } else {
                                    crate::messages::system::HealthLevel::Error
                                },
                                message: status.message.unwrap_or_else(|| "No message".to_string()),
                                diagnostics,
                            }
                        }
                        Err(_) => {
                            // `run()` holds `node_for_health`'s lock for the node's entire
                            // life (see `Node::start`), so this branch is in practice taken
                            // on every tick once the node is actually running - this isn't an
                            // occasional contention fallback. Fall back to the post-init
                            // `static_diagnostics` snapshot (see its field doc comment on
                            // `Node<T>`) rather than `None`, so an override like
                            // `VideoStreamerNode::health_diagnostics()` - whose `camera_ids`
                            // never changes after `init()` anyway - still gets reported here
                            // instead of silently disappearing the moment the node starts
                            // running.
                            debug!("Node busy, reporting as healthy");
                            crate::messages::system::HealthStatus {
                                timestamp: crate::prelude::now_micros(),
                                level: crate::messages::system::HealthLevel::Ok,
                                message: "Node running (busy)".to_string(),
                                diagnostics: static_diagnostics_for_health.clone(),
                            }
                        }
                    };

                    if let Err(e) = reporter.report(msg_status).await {
                        warn!("Failed to report health: {}", e);
                    } else {
                        tracing::trace!("Health report sent successfully");
                    }
                }
            });
        } else {
            info!("Health reporting disabled");
        }
    }
    #[cfg(not(feature = "messaging"))]
    let _ = &health_config;

    node.start(ctx).await
}

#[cfg(all(test, not(feature = "messaging")))]
mod tests {
    use super::*;
    use crate::error::Mecha10Error;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc as StdArc;

    #[derive(Debug, Default, serde::Deserialize)]
    struct FakeConfig;

    #[derive(Debug)]
    struct OkNode {
        ran: StdArc<AtomicBool>,
    }

    #[async_trait]
    impl NodeImpl for OkNode {
        type Config = FakeConfig;

        async fn init(_config: Self::Config) -> Result<Self> {
            Ok(Self {
                ran: StdArc::new(AtomicBool::new(false)),
            })
        }

        async fn run(&mut self, _ctx: &Context) -> Result<()> {
            self.ran.store(true, Ordering::SeqCst);
            Ok(())
        }
    }

    #[derive(Debug)]
    struct RejectingPreflightNode {
        ran: StdArc<AtomicBool>,
    }

    #[async_trait]
    impl NodeImpl for RejectingPreflightNode {
        type Config = FakeConfig;

        async fn init(_config: Self::Config) -> Result<Self> {
            Ok(Self {
                ran: StdArc::new(AtomicBool::new(false)),
            })
        }

        async fn run(&mut self, _ctx: &Context) -> Result<()> {
            self.ran.store(true, Ordering::SeqCst);
            Ok(())
        }

        fn preflight(&self) -> Result<()> {
            Err(Mecha10Error::Configuration("boom: bad node content".to_string()))
        }
    }

    #[tokio::test]
    async fn default_preflight_is_a_noop_and_run_still_executes() {
        let node = Node::<OkNode>::from_config(FakeConfig).await.unwrap();
        let ctx = Context::new("test-node-ok").await.unwrap();

        let ran_handle = { node.node.lock().await.ran.clone() };

        node.start(ctx).await.unwrap();

        assert!(ran_handle.load(Ordering::SeqCst), "run() should have executed");
    }

    #[tokio::test]
    async fn start_fails_fast_when_preflight_rejects_config_and_never_calls_run() {
        let node = Node::<RejectingPreflightNode>::from_config(FakeConfig).await.unwrap();
        let ctx = Context::new("test-node-rejecting").await.unwrap();

        let ran_handle = { node.node.lock().await.ran.clone() };

        let result = node.start(ctx).await;

        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("boom: bad node content"));
        assert!(
            !ran_handle.load(Ordering::SeqCst),
            "run() must not execute when preflight() rejects the node"
        );
    }

    #[test]
    fn node_type_kebab_case_matches_existing_builtin_node_types_strings() {
        let expected: &[(NodeType, &str)] = &[
            (NodeType::Imu, "imu"),
            (NodeType::Camera, "camera"),
            (NodeType::ImageClassifier, "image-classifier"),
            (NodeType::ObjectDetector, "object-detector"),
            (NodeType::BehaviorExecutor, "behavior-executor"),
            (NodeType::LlmCommand, "llm-command"),
            (NodeType::Teleop, "teleop"),
            (NodeType::Motor, "motor"),
            (NodeType::WebsocketBridge, "websocket-bridge"),
            (NodeType::VideoStreamer, "video-streamer"),
            (NodeType::Simulator, "simulator"),
            (NodeType::Speaker, "speaker"),
            (NodeType::Listener, "listener"),
            (NodeType::Rl, "rl"),
            (NodeType::InstanceTrack, "instance-track"),
            (NodeType::Diagnostics, "diagnostics"),
        ];

        assert_eq!(expected.len(), NodeType::ALL.len());

        for (variant, kebab) in expected {
            let json = serde_json::to_string(variant).unwrap();
            assert_eq!(json, format!("\"{}\"", kebab));
            assert_eq!(variant.as_str(), *kebab);

            let round_tripped: NodeType = serde_json::from_str(&json).unwrap();
            assert_eq!(round_tripped, *variant);
        }
    }

    #[test]
    fn node_tag_kebab_case_matches_expected_strings() {
        let expected: &[(NodeTag, &str)] = &[
            (NodeTag::Sensor, "sensor"),
            (NodeTag::Inertial, "inertial"),
            (NodeTag::Capture, "capture"),
            (NodeTag::Ai, "ai"),
            (NodeTag::Classifier, "classifier"),
            (NodeTag::Detector, "detector"),
            (NodeTag::Vision, "vision"),
            (NodeTag::Planner, "planner"),
            (NodeTag::Control, "control"),
            (NodeTag::Nlp, "nlp"),
            (NodeTag::Input, "input"),
            (NodeTag::Actuator, "actuator"),
            (NodeTag::Motion, "motion"),
            (NodeTag::Relay, "relay"),
            (NodeTag::Websocket, "websocket"),
            (NodeTag::Streamer, "streamer"),
            (NodeTag::Video, "video"),
            (NodeTag::Simulator, "simulator"),
            (NodeTag::Logger, "logger"),
        ];

        for (variant, kebab) in expected {
            let json = serde_json::to_string(variant).unwrap();
            assert_eq!(json, format!("\"{}\"", kebab));

            let round_tripped: NodeTag = serde_json::from_str(&json).unwrap();
            assert_eq!(round_tripped, *variant);
        }
    }

    #[test]
    fn node_impl_defaults_node_type_to_none_and_tags_to_empty() {
        assert_eq!(OkNode::NODE_TYPE, None);
        assert!(OkNode::NODE_TAGS.is_empty());
    }

    #[test]
    fn node_impl_defaults_registration_consts() {
        assert_eq!(OkNode::CRATE_NAME, None);
        assert_eq!(OkNode::DESCRIPTION, "");
        assert_eq!(OkNode::DEFAULT_RUN_TARGET, "robot");
        assert!(OkNode::CATALOG_TAGS.is_empty());
    }
}