mecha10-core 0.8.0

//! Generic dashboard command relay
//!
//! Subscribes to [`TOPIC_CLI_COMMAND`] over Redis pub/sub and relays dashboard-originated node
//! lifecycle commands (start/stop/restart) into whatever is actually running the node graph in
//! this process.
//!
//! `mecha10 dev` and a production robot process (`mecha10 run`, or a launcher-managed robot
//! binary) manage nodes very differently: `mecha10 dev` supervises every node as a
//! `ProcessService`-owned subprocess, while a compiled project binary runs custom nodes
//! in-process and framework nodes via a handful of `mecha10-node-runner` subprocesses it
//! manages itself. [`CommandRelay`] is generic over a [`NodeController`] so both call sites
//! (`packages/cli`'s dev session, and the generated project binary's `main.rs.template`) share
//! the exact same Redis subscribe/parse/dispatch loop, and only the "how do I actually
//! start/stop a node" mechanics differ.
//!
//! Before LAB-2075 this listener only ever ran inside `mecha10 dev`'s `DevSession::execute` -
//! any robot started via `mecha10 run` or a launcher-deployed binary had no relay wiring
//! between the dashboard and its running nodes at all.
//!
//! `Command::Trigger` (LAB-2082) is a fourth, distinct kind of dispatch: it doesn't start,
//! stop, or restart anything, it fires a node-specific one-shot action (e.g.
//! image-classifier's single-shot classify). Because `CommandRelay` has no notion of a
//! given node's own control-topic wire shape, it doesn't route `Trigger` through
//! [`NodeController`] like the lifecycle commands - it publishes a node-agnostic
//! [`NodeTriggerEvent`] on `system/node/{target}/trigger`, which any node may additionally
//! subscribe to. Every other `Command` variant this relay has no handling for (e.g.
//! `Pause`/`Resume`) now publishes a status event instead of silently vanishing - see
//! `handle_message`'s fallback arm.

use crate::lifecycle::{get_timestamp_ms, topic_node_trigger, NodeLifecycleStatus, NodeStatusEvent, NodeTriggerEvent};
use crate::messages::system::{Command, SystemCommand};
use anyhow::{Context, Result};
use futures::StreamExt;
use std::sync::Arc;

pub const TOPIC_CLI_COMMAND: &str = "system/cli/command";

/// Human-readable message for a [`Command`] variant `CommandRelay` has no dispatch
/// handling for, published to `system/node/{target}/status` and logged so an
/// unsupported command is never silently swallowed (LAB-2082). A free function so
/// its text is unit-testable without a Redis connection.
fn unsupported_command_message(command: &Command) -> String {
    format!("Unsupported command: {:?}", command)
}

type SharedRedisConn = Arc<tokio::sync::Mutex<redis::aio::MultiplexedConnection>>;

#[derive(Clone)]
pub struct RedisStatusPublisher {
    conn: SharedRedisConn,
}

impl RedisStatusPublisher {
    /// Open a dedicated Redis connection for publishing status updates.
    pub async fn connect(redis_url: &str) -> Result<Self> {
        let client = redis::Client::open(redis_url).context("Failed to create Redis client for status publisher")?;
        let conn = client
            .get_multiplexed_async_connection()
            .await
            .context("Failed to get Redis connection for publishing status")?;
        Ok(Self {
            conn: Arc::new(tokio::sync::Mutex::new(conn)),
        })
    }

    /// Publish a status update for `node_name`.
    pub async fn publish_status(
        &self,
        node_name: &str,
        status: NodeLifecycleStatus,
        pid: Option<u32>,
        error: Option<String>,
    ) -> Result<()> {
        let event = NodeStatusEvent {
            node: node_name.to_string(),
            status,
            pid,
            error,
            timestamp: get_timestamp_ms(),
        };

        let topic = format!("system/node/{}/status", node_name);
        let payload = serde_json::to_string(&event)?;

        let mut conn = self.conn.lock().await;
        redis::cmd("PUBLISH")
            .arg(&topic)
            .arg(&payload)
            .query_async::<i64>(&mut *conn)
            .await
            .with_context(|| format!("Failed to publish status for node '{}'", node_name))?;

        tracing::debug!("Published status for '{}': {:?}", node_name, event.status);

        Ok(())
    }
}

#[derive(Debug, Clone)]
pub struct StatusUpdate {
    pub node: String,
    pub status: NodeLifecycleStatus,
    pub pid: Option<u32>,
    pub error: Option<String>,
}

pub type StatusUpdateSender = tokio::sync::mpsc::UnboundedSender<StatusUpdate>;

/// Drain [`StatusUpdate`]s off `updates` and publish each via `publisher`, until `shutdown`
/// fires or the sending half is dropped. A publish failure is logged and does not stop the loop
/// (mirrors `CommandRelay::handle_stop`'s "don't fail the caller on a publish error" behavior).
pub async fn run_status_publisher(
    mut updates: tokio::sync::mpsc::UnboundedReceiver<StatusUpdate>,
    publisher: RedisStatusPublisher,
    mut shutdown: tokio::sync::broadcast::Receiver<()>,
) {
    loop {
        tokio::select! {
            _ = shutdown.recv() => break,
            update = updates.recv() => match update {
                Some(update) => {
                    if let Err(e) = publisher
                        .publish_status(&update.node, update.status.clone(), update.pid, update.error.clone())
                        .await
                    {
                        tracing::warn!("Failed to publish status for '{}': {}", update.node, e);
                    }
                }
                None => break,
            },
        }
    }
}

pub trait NodeController: Send + Sync {
    /// Whether `target` is currently running.
    fn is_running(&self, target: &str) -> impl std::future::Future<Output = bool> + Send;

    /// Start `target`. Returns an identifier (a real PID, or a synthetic value for in-process
    /// nodes) suitable for status reporting.
    fn start(&self, target: &str) -> impl std::future::Future<Output = Result<u32>> + Send;

    /// Stop `target`.
    fn stop(&self, target: &str) -> impl std::future::Future<Output = Result<()>> + Send;
}

pub type CommandRelayLogger = Arc<dyn Fn(String) + Send + Sync>;

pub struct CommandRelay<C: NodeController> {
    redis_url: String,
    controller: C,
    logger: Option<CommandRelayLogger>,
}

impl<C: NodeController> CommandRelay<C> {
    /// Create a new command relay for the given controller.
    pub fn new(redis_url: impl Into<String>, controller: C) -> Self {
        Self {
            redis_url: redis_url.into(),
            controller,
            logger: None,
        }
    }

    crate::builder_setter!(
        /// Attach an additional sink for human-readable activity messages (e.g. a TUI log buffer).
        into_some, with_logger, logger, CommandRelayLogger
    );

    fn log(&self, msg: impl Into<String>) {
        let msg = msg.into();
        if let Some(logger) = &self.logger {
            logger(msg.clone());
        }
        tracing::info!("{msg}");
    }

    /// Start listening for commands.
    ///
    /// Runs until the Redis connection is closed, an error occurs, or `shutdown` fires.
    pub async fn run(self, mut shutdown: tokio::sync::broadcast::Receiver<()>) -> Result<()> {
        let client =
            redis::Client::open(self.redis_url.as_str()).context("Failed to create Redis client for command relay")?;

        let mut pubsub = client
            .get_async_pubsub()
            .await
            .context("Failed to get Redis pub/sub connection")?;

        pubsub
            .subscribe(TOPIC_CLI_COMMAND)
            .await
            .context("Failed to subscribe to command topic")?;

        tracing::info!("Command relay subscribed to {}", TOPIC_CLI_COMMAND);

        let publish_conn = client
            .get_multiplexed_async_connection()
            .await
            .context("Failed to get Redis connection for publishing")?;
        let publisher = RedisStatusPublisher {
            conn: Arc::new(tokio::sync::Mutex::new(publish_conn)),
        };

        let mut msg_stream = pubsub.on_message();

        loop {
            tokio::select! {
                _ = shutdown.recv() => {
                    break;
                }

                msg = msg_stream.next() => {
                    match msg {
                        Some(msg) => {
                            let payload: String = msg.get_payload().unwrap_or_default();
                            if let Err(e) = self.handle_message(&payload, publisher.clone()).await {
                                tracing::error!("Error handling command: {}", e);
                            }
                        }
                        None => {
                            tracing::warn!("Redis pub/sub connection closed");
                            break;
                        }
                    }
                }
            }
        }

        Ok(())
    }

    async fn handle_message(&self, payload: &str, publisher: RedisStatusPublisher) -> Result<()> {
        let cmd: SystemCommand =
            serde_json::from_str(payload).with_context(|| format!("Failed to parse command: {}", payload))?;

        match cmd.command {
            Command::Start => self.handle_start(&cmd.target, publisher).await?,
            Command::Stop => self.handle_stop(&cmd.target, publisher).await?,
            Command::Restart => self.handle_restart(&cmd.target, publisher).await?,
            Command::Trigger => self.handle_trigger(&cmd.target, publisher).await?,
            unsupported => {
                let message = unsupported_command_message(&unsupported);
                self.log(format!("⚠️  [Dashboard] {} (target: '{}')", message, cmd.target));
                publisher
                    .publish_status(&cmd.target, NodeLifecycleStatus::Failed, None, Some(message))
                    .await?;
            }
        }

        Ok(())
    }

    async fn handle_start(&self, target: &str, publisher: RedisStatusPublisher) -> Result<()> {
        if self.controller.is_running(target).await {
            self.log(format!("⚠️  [Dashboard] Node '{}' is already running", target));
            return Ok(());
        }

        self.log(format!("▶️  [Dashboard] Starting node '{}'...", target));
        publisher
            .publish_status(target, NodeLifecycleStatus::Starting, None, None)
            .await?;

        match self.controller.start(target).await {
            Ok(pid) => {
                publisher
                    .publish_status(target, NodeLifecycleStatus::Started, Some(pid), None)
                    .await?;
                self.log(format!("✅ [Dashboard] Started node '{}' (PID: {})", target, pid));
            }
            Err(e) => {
                publisher
                    .publish_status(target, NodeLifecycleStatus::Failed, None, Some(e.to_string()))
                    .await?;
                self.log(format!("❌ [Dashboard] Failed to start node '{}': {}", target, e));
            }
        }

        Ok(())
    }

    async fn handle_stop(&self, target: &str, publisher: RedisStatusPublisher) -> Result<()> {
        if !self.controller.is_running(target).await {
            self.log(format!("⚠️  [Dashboard] Node '{}' is not running", target));
            return Ok(());
        }

        self.log(format!("🛑 [Dashboard] Stopping node '{}'...", target));

        let _ = publisher
            .publish_status(target, NodeLifecycleStatus::Stopping, None, None)
            .await;

        match self.controller.stop(target).await {
            Ok(()) => {
                publisher
                    .publish_status(target, NodeLifecycleStatus::Stopped, None, None)
                    .await?;
                self.log(format!("✅ [Dashboard] Stopped node '{}'", target));
            }
            Err(e) => {
                publisher
                    .publish_status(target, NodeLifecycleStatus::Failed, None, Some(e.to_string()))
                    .await?;
                self.log(format!("❌ [Dashboard] Failed to stop node '{}': {}", target, e));
            }
        }

        Ok(())
    }

    /// Relay a `Command::Trigger` to `target` as a [`NodeTriggerEvent`] on
    /// [`topic_node_trigger`] (LAB-2082).
    ///
    /// Unlike `handle_start`/`handle_stop`/`handle_restart`, this does not go through
    /// [`NodeController`] - `CommandRelay` has no notion of a node's own control-topic
    /// wire shape (e.g. image-classifier's `/vision/classification/control`), and
    /// hard-coding that knowledge here would defeat the point of `NodeController`
    /// being generic across `mecha10 dev` and production binaries alike. Instead this
    /// publishes a node-agnostic trigger event that any node can opt into listening
    /// for on its own `system/node/{name}/trigger` topic, alongside its existing
    /// control topic. Reuses `publisher`'s connection (same module, so its private
    /// `conn` field is accessible here) rather than opening a second one just for this.
    async fn handle_trigger(&self, target: &str, publisher: RedisStatusPublisher) -> Result<()> {
        self.log(format!("📸 [Dashboard] Triggering node '{}'...", target));

        let topic = topic_node_trigger(target);
        let event = NodeTriggerEvent {
            node: target.to_string(),
            timestamp: get_timestamp_ms(),
        };
        let payload = serde_json::to_string(&event)?;

        {
            let mut conn = publisher.conn.lock().await;
            redis::cmd("PUBLISH")
                .arg(&topic)
                .arg(&payload)
                .query_async::<i64>(&mut *conn)
                .await
                .with_context(|| format!("Failed to publish trigger for node '{}'", target))?;
        }

        tracing::debug!("Published trigger for '{}' on {}", target, topic);

        Ok(())
    }

    async fn handle_restart(&self, target: &str, publisher: RedisStatusPublisher) -> Result<()> {
        self.log(format!("🔄 [Dashboard] Restarting node '{}'...", target));

        if self.controller.is_running(target).await {
            self.handle_stop(target, publisher.clone()).await?;
            tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        }

        self.handle_start(target, publisher).await?;

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
    use std::sync::Mutex as StdMutex;

    #[derive(Default)]
    struct FakeController {
        running: AtomicBool,
        start_calls: AtomicU32,
        stop_calls: AtomicU32,
        start_result: StdMutex<Option<Result<u32>>>,
    }

    impl NodeController for FakeController {
        async fn is_running(&self, _target: &str) -> bool {
            self.running.load(Ordering::SeqCst)
        }

        async fn start(&self, _target: &str) -> Result<u32> {
            self.start_calls.fetch_add(1, Ordering::SeqCst);
            let outcome = self.start_result.lock().unwrap().take();
            match outcome {
                Some(Ok(pid)) => {
                    self.running.store(true, Ordering::SeqCst);
                    Ok(pid)
                }
                Some(Err(e)) => Err(e),
                None => {
                    self.running.store(true, Ordering::SeqCst);
                    Ok(42)
                }
            }
        }

        async fn stop(&self, _target: &str) -> Result<()> {
            self.stop_calls.fetch_add(1, Ordering::SeqCst);
            self.running.store(false, Ordering::SeqCst);
            Ok(())
        }
    }

    #[test]
    fn topic_constant_matches_dashboard_contract() {
        assert_eq!(TOPIC_CLI_COMMAND, "system/cli/command");
    }

    #[test]
    fn with_logger_forwards_log_messages_to_the_attached_sink() {
        let captured: Arc<StdMutex<Vec<String>>> = Arc::new(StdMutex::new(Vec::new()));
        let sink = captured.clone();
        let controller = FakeController::default();
        let relay = CommandRelay::new("redis://localhost:6379", controller)
            .with_logger(Arc::new(move |msg| sink.lock().unwrap().push(msg)));

        relay.log("hello");

        assert_eq!(captured.lock().unwrap().as_slice(), ["hello".to_string()]);
    }

    #[tokio::test]
    async fn handle_message_start_calls_controller_start_when_not_running() {
        let controller = FakeController::default();
        let relay = CommandRelay::new("redis://localhost:6379", controller);

        assert!(!relay.controller.is_running("llm-command").await);
        relay.controller.start("llm-command").await.unwrap();
        assert!(relay.controller.is_running("llm-command").await);
        assert_eq!(relay.controller.start_calls.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn handle_message_stop_calls_controller_stop_when_running() {
        let controller = FakeController::default();
        controller.running.store(true, Ordering::SeqCst);
        let relay = CommandRelay::new("redis://localhost:6379", controller);

        relay.controller.stop("llm-command").await.unwrap();
        assert!(!relay.controller.is_running("llm-command").await);
        assert_eq!(relay.controller.stop_calls.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn system_command_start_parses_into_start_variant() {
        let payload = serde_json::to_string(&SystemCommand::start("llm-command")).unwrap();
        let cmd: SystemCommand = serde_json::from_str(&payload).unwrap();
        assert_eq!(cmd.target, "llm-command");
        assert_eq!(cmd.command, Command::Start);
    }

    /// LAB-2082: the dashboard's single-shot "c" trigger publishes `Command::Trigger`
    /// on `TOPIC_CLI_COMMAND` - confirm it round-trips like every other command.
    #[test]
    fn system_command_trigger_parses_into_trigger_variant() {
        let payload = serde_json::to_string(&SystemCommand::trigger("image-classifier")).unwrap();
        let cmd: SystemCommand = serde_json::from_str(&payload).unwrap();
        assert_eq!(cmd.target, "image-classifier");
        assert_eq!(cmd.command, Command::Trigger);
    }

    /// `handle_trigger` publishes on `topic_node_trigger(target)`, not the node's own
    /// control topic - this is the contract image-classifier's `run()` loop relies on
    /// when it subscribes for dashboard-initiated triggers (LAB-2082).
    #[test]
    fn node_trigger_topic_matches_the_convention_nodes_subscribe_to() {
        assert_eq!(
            topic_node_trigger("image-classifier"),
            "system/node/image-classifier/trigger"
        );
    }

    /// LAB-2082: commands `CommandRelay` has no dispatch for (e.g. `Pause`) must
    /// produce a non-empty, descriptive message instead of vanishing at `debug!`.
    #[test]
    fn unsupported_command_message_names_the_offending_command() {
        let message = unsupported_command_message(&Command::Pause);
        assert!(message.contains("Pause"), "message was: {message}");
        assert!(message.contains("Unsupported"), "message was: {message}");
    }

    /// Every non-lifecycle, non-trigger `Command` variant should produce a distinct
    /// message naming that variant, so an operator can tell which unsupported command
    /// they sent from the published status event alone.
    #[test]
    fn unsupported_command_message_is_distinct_per_variant() {
        let resume = unsupported_command_message(&Command::Resume);
        let health_check = unsupported_command_message(&Command::HealthCheck);
        assert_ne!(resume, health_check);
        assert!(resume.contains("Resume"));
        assert!(health_check.contains("HealthCheck"));
    }
}