mecha10_core/

node.rs

// Mecha10 Node Trait and Lifecycle Management
//
// This module defines the core Node trait and lifecycle management for Mecha10 nodes.
// Every Mecha10 node implements the NodeImpl trait to define its behavior.

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};

// ============================================================================
// Health Priority
// ============================================================================

/// Node health reporting priority
///
/// Determines heartbeat frequency and timeout thresholds based on
/// node criticality. Safety-critical nodes report more frequently.
///
/// # Priority Levels
///
/// | Priority | Interval | TTL | Timeout | Use Case |
/// |----------|----------|-----|---------|----------|
/// | Critical | 1s | 5s | 3s | motor-controller, estop |
/// | Normal | 5s | 60s | 15s | sensors, AI nodes |
/// | Background | 30s | 120s | 90s | logging, diagnostics |
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// // Create config with critical priority
/// let config = HealthReportingConfig::critical();
///
/// // Or specify priority explicitly
/// let config = HealthReportingConfig {
///     priority: HealthPriority::Critical,
///     ..Default::default()
/// };
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum HealthPriority {
    /// Safety-critical nodes that require immediate failure detection.
    ///
    /// Examples: motor-controller, estop, safety-monitor
    ///
    /// - Interval: 1 second
    /// - TTL: 5 seconds
    /// - Timeout: 3 seconds
    Critical,

    /// Standard operational nodes (default).
    ///
    /// Examples: camera, sensors, simulator
    ///
    /// - Interval: 5 seconds
    /// - TTL: 60 seconds
    /// - Timeout: 15 seconds
    #[default]
    Normal,

    /// Background/auxiliary nodes where delayed detection is acceptable.
    ///
    /// Examples: logging, telemetry, diagnostics
    ///
    /// - Interval: 30 seconds
    /// - TTL: 120 seconds
    /// - Timeout: 90 seconds
    Background,
}

impl HealthPriority {
    /// Get the heartbeat interval for this priority level
    pub const fn interval(&self) -> Duration {
        match self {
            Self::Critical => Duration::from_secs(1),
            Self::Normal => Duration::from_secs(5),
            Self::Background => Duration::from_secs(30),
        }
    }

    /// 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 {
        match self {
            Self::Critical => 5,
            Self::Normal => 60,
            Self::Background => 120,
        }
    }

    /// 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 {
        match self {
            Self::Critical => Duration::from_secs(3),
            Self::Normal => Duration::from_secs(15),
            Self::Background => Duration::from_secs(90),
        }
    }

    /// Get the timeout in microseconds (for comparison with timestamps)
    pub const fn timeout_micros(&self) -> u64 {
        match self {
            Self::Critical => 3_000_000,
            Self::Normal => 15_000_000,
            Self::Background => 90_000_000,
        }
    }
}

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"),
        }
    }
}

// ============================================================================
// Configuration
// ============================================================================

/// Configuration for node health reporting
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// // Default config (Normal priority, 5s interval)
/// let config = HealthReportingConfig::default();
///
/// // Critical priority (1s interval)
/// let config = HealthReportingConfig::critical();
///
/// // Background priority (30s interval)
/// let config = HealthReportingConfig::background();
///
/// // Custom config
/// let config = HealthReportingConfig {
///     enabled: true,
///     priority: HealthPriority::Normal,
///     store_history: false, // Disable history for high-frequency nodes
/// };
/// ```
#[derive(Debug, Clone)]
pub struct HealthReportingConfig {
    /// Whether health reporting is enabled (default: true)
    pub enabled: bool,

    /// Priority level determines interval and timeout (default: Normal)
    pub priority: HealthPriority,

    /// Whether to store health history in Redis (default: true)
    ///
    /// Disable for high-frequency critical nodes to reduce Redis load.
    /// History is stored as a sorted set with 24-hour retention.
    pub store_history: bool,
}

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

impl HealthReportingConfig {
    /// Create config for critical priority nodes (1s interval)
    pub fn critical() -> Self {
        Self {
            enabled: true,
            priority: HealthPriority::Critical,
            store_history: false, // Disable history for critical nodes to reduce load
        }
    }

    /// 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 {
            enabled: true,
            priority: HealthPriority::Background,
            store_history: 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()
    }
}

/// Trait for node configurations that specify an update rate
///
/// This trait provides a standard interface for nodes that publish at a fixed rate.
/// Implement this trait on your node's config struct to enable convenient rate-based
/// interval creation.
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// #[derive(Debug, Clone, Serialize, Deserialize)]
/// pub struct CameraConfig {
///     pub fps: f32,
///     pub width: u32,
///     pub height: u32,
/// }
///
/// impl RateConfig for CameraConfig {
///     fn rate_hz(&self) -> f32 {
///         self.fps
///     }
/// }
///
/// // Now you can use the config to create intervals easily:
/// # async fn example(config: CameraConfig) {
/// let mut interval = config.create_interval();
/// loop {
///     interval.tick().await;
///     // Publish at configured rate
/// }
/// # }
/// ```
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())
    }
}

// ============================================================================
// Health Status Helper
// ============================================================================

/// Helper methods for creating HealthStatus instances
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 HealthStatusExt for HealthStatus {
    fn healthy(message: impl Into<String>) -> Self {
        Self {
            healthy: true,
            last_check: crate::prelude::now_micros(),
            failure_count: 0,
            message: Some(message.into()),
        }
    }

    fn unhealthy(message: impl Into<String>) -> Self {
        Self {
            healthy: false,
            last_check: crate::prelude::now_micros(),
            failure_count: 1,
            message: Some(message.into()),
        }
    }
}

// ============================================================================
// Node Type and Tags
// ============================================================================

/// Built-in node kind.
///
/// This is a deliberately *closed* set -- it drives 1:1 config-schema
/// resolution in `mecha10-schema-registry` (a node has exactly one config
/// struct), so it can't also double as an open-ended "what kind of thing is
/// this" label for custom/project-local nodes. See [`NodeImpl::NODE_TYPE`]
/// and the "Node type and tags" section of the Node Management doc
/// (<https://linear.app/laboratory-one/document/node-management-1b7f4324db40>)
/// for the full design rationale.
#[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,
}

impl NodeType {
    /// Every built-in node type, in declaration order.
    ///
    /// `mecha10-schema-registry`'s `BUILTIN_NODE_TYPES` derives from this
    /// instead of maintaining its own parallel `&[&str]` list.
    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,
    ];

    /// 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",
        }
    }
}

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

/// Capability tag describing something specific a node *does*.
///
/// Unlike [`NodeType`], this is meant to be usable by custom/project-local
/// nodes too -- see [`NodeImpl::NODE_TAGS`]. Each variant names a concrete
/// capability (never a structural/meta label like "publishes messages",
/// which is true of nearly every node and carries no routing information).
#[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,
}

// ============================================================================
// NodeImpl Trait
// ============================================================================

/// Core trait that all Mecha10 nodes must implement.
///
/// This trait defines the lifecycle of a node:
/// 1. `init()` - Initialize the node with configuration
/// 2. `run()` - Main execution loop
/// 3. `shutdown()` - Graceful shutdown
/// 4. `health_check()` - Report health status
///
/// # Example
///
/// ```rust
/// use mecha10_core::prelude::*;
///
/// #[derive(Debug)]
/// struct MyNode {
///     config: MyConfig,
/// }
///
/// #[derive(Debug, Deserialize)]
/// struct MyConfig {
///     rate: f32,
/// }
///
/// #[async_trait]
/// impl NodeImpl for MyNode {
///     type Config = MyConfig;
///
///     async fn init(config: Self::Config) -> Result<Self> {
///         info!("Initializing node with rate: {}", config.rate);
///         Ok(Self { config })
///     }
///
///     async fn run(&mut self, ctx: &Context) -> Result<()> {
///         let mut images = ctx.subscribe(sensor::CAMERA_RGB).await?;
///
///         while let Some(image) = images.recv().await {
///             // Process image
///             info!("Received image: {}x{}", image.width, image.height);
///         }
///
///         Ok(())
///     }
/// }
/// ```
#[async_trait]
pub trait NodeImpl: Send + Sync + Debug + 'static {
    /// Configuration type for this node
    type Config: DeserializeOwned + Send + Sync + Debug;

    /// Built-in node type, if this is one of the framework-provided node
    /// kinds. `None` for custom/project-local nodes -- `NodeType` is
    /// deliberately closed (it drives 1:1 config-schema resolution in
    /// `mecha10-schema-registry`), so a custom node correctly stays `None`
    /// here rather than being forced to misrepresent itself as one of the
    /// built-in variants. This matches
    /// `mecha10-schema-registry::resolve_node_type`'s existing `None` ->
    /// generic-schema-fallback behavior for anything unrecognized.
    const NODE_TYPE: Option<NodeType> = None;

    /// Capability tags. Empty by default. Unlike `NODE_TYPE`,
    /// custom/project-local nodes are encouraged to set this -- tagging e.g.
    /// a custom lidar node `NodeTag::Sensor` lets it be picked up by any
    /// dashboard widget that requires that tag, without needing an entry in
    /// `mecha10-schema-registry` at all.
    const NODE_TAGS: &'static [NodeTag] = &[];

    /// 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()),
        }
    }

    /// 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(())
    }
}

// ============================================================================
// Node Runner
// ============================================================================

/// Wrapper that manages the lifecycle of a NodeImpl.
///
/// This provides:
/// - State management
/// - Lifecycle orchestration
/// - Error handling
/// - Health monitoring
#[derive(Clone)]
pub struct Node<T: NodeImpl> {
    node: Arc<Mutex<T>>,
    state: Arc<Mutex<NodeState>>,
}

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?;

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

    /// 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<()> {
        // Node-specific pre-run content check (LAB-1773) - runs after `init()`
        // (so it can see fully-typed, already-validated config/state) and
        // before `run()`, so a bad config fails fast instead of crashing deep
        // in the node's main loop.
        {
            let node = self.node.lock().await;
            node.preflight()?;
        }

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

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

        // Handle result
        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");

        // Transition to shutting down
        {
            let mut state = self.state.lock().await;
            *state = NodeState::Stopping;
        }

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

        // Transition to stopped
        {
            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)
    }
}

// ============================================================================
// Helper Functions
// ============================================================================

/// 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
/// use mecha10_core::prelude::*;
///
/// #[tokio::main]
/// async fn main() -> Result<()> {
///     let config = MyConfig { rate: 10.0 };
///     let ctx = Context::new("my_node");
///
///     // Use default health reporting (Normal priority, 5s interval)
///     run_node::<MyNode>(config, ctx, HealthReportingConfig::default()).await?;
///
///     // Critical priority for safety-critical nodes (1s interval)
///     run_node::<MyNode>(config, ctx, HealthReportingConfig::critical()).await?;
///
///     // Background priority for auxiliary nodes (30s interval)
///     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<()> {
    // Initialize logging (safe to call multiple times - see
    // `prelude::init_logging`'s doc comment for why the Redis-forwarding
    // layer it attaches has to be feature-gated there, not here).
    crate::prelude::init_logging();

    // Wire the (now-attached, if "messaging" is enabled) Redis log layer up
    // to this node's Context so captured log events actually get published
    // to `/system/log` - without this, node logs only ever reached local
    // stdout/journald, and a connected robot's logs never reached the
    // platform's `/logs/stream` subscribers no matter what the dashboard or
    // aggregator side did. See `crate::log_publisher`.
    #[cfg(feature = "messaging")]
    crate::log_publisher::attach(ctx.node_id(), ctx.clone());

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

    // Start automatic health reporting (requires messaging feature)
    #[cfg(feature = "messaging")]
    {
        if health_config.enabled {
            // Create clones for the health reporting task
            let node_for_health = node.node.clone();
            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;

                // Try to create health reporter with priority config, tagged
                // with this node's built-in type/tags (if any).
                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;
                    }
                };

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

                let mut interval = tokio::time::interval(report_interval);
                info!("Health reporting loop starting");

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

                    // Try to get health status from node without blocking
                    // If the node is busy (lock held by run()), report as healthy but busy
                    let msg_status = match node_for_health.try_lock() {
                        Ok(node) => {
                            let status = node.health_check().await;
                            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: None,
                            }
                        }
                        Err(_) => {
                            // Node is busy (lock held by run()) - report as healthy
                            // This is normal for long-running nodes
                            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: None,
                            }
                        }
                    };

                    // Report to centralized monitoring
                    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
}

// `Context::new()` only avoids a real Redis connection attempt when the
// `messaging` feature is off (see `Context::new`'s `#[cfg(feature =
// "messaging")]` branch) - with it on, constructing a `Context` needs a live
// broker, which isn't available in a unit test. This whole module exercises
// the real `Node::start()` -> `preflight()` -> `run()` wiring end-to-end in
// the (also real, supported) no-messaging configuration; `NodeImpl::preflight()`'s
// default-no-op and per-node override behavior itself is feature-independent
// and covered directly in each node crate's own `config.rs` tests (e.g.
// `mecha10-nodes-camera`, `mecha10-nodes-motor`).
#[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"),
        ];

        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());
    }
}