mecha10-cli 0.6.3

//! Relay `mecha10-node-runner`'s health-check tracing lines into the dev-mode TUI.
//!
//! `Supervisor::monitor_with_signals_unix`'s periodic health check
//! (`packages/node-runner/src/supervisor.rs`) only ever logs a state transition via
//! `tracing::info!`/`warn!` — e.g. `"Node 'camera' health: Running"` / `"Node 'camera' health:
//! Stopped"` — plus restart/backoff lines like `"Restart attempt 1 for 'camera'"` and `"Node
//! 'camera' will restart in 2s"`. Because the CLI spawns node-runner with its stdout/stderr
//! redirected to `logs/<node>.log` (see `services/process/spawn.rs::spawn_with_env`, the same
//! file `lifecycle.rs::tail_node_log` already does a one-shot read of after a fresh spawn), those
//! lines already land on disk — they just had no live reader before this module.
//!
//! [`HealthTailer`] is a minimal live reader: it remembers a byte offset per node's log file,
//! and on each [`HealthTailer::poll`] reads only the bytes appended since the last poll,
//! pattern-matches complete new lines for the events above, and relays a formatted line into the
//! dev-mode TUI's [`crate::ui::LogBuffer`] — reusing the CLI's existing per-node log file (see
//! module docs above) rather than adding a second polling loop or a new IPC channel between
//! node-runner and the CLI.

use crate::paths;
use crate::ui::LogBuffer;
use mecha10_cli_core::utils::strip_node_scope;
use std::collections::HashMap;
use std::fs::File;
use std::io::{Read, Seek, SeekFrom};
use std::path::Path;
use std::time::{Duration, Instant};

/// One node-runner health/restart event, parsed from a single tracing log line.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum HealthEvent {
    /// `"Node '{node}' health: Running"` / `"...: Stopped"` — a health-check transition.
    Transition { node: String, running: bool },
    /// `"Restart attempt {attempt} for '{node}'"` — a restart is starting.
    RestartAttempt { node: String, attempt: u32 },
    /// `"Node '{node}' will restart in {detail}"` — backoff before the next restart attempt.
    WillRestart { node: String, detail: String },
}

impl HealthEvent {
    /// The node name this event is about, exactly as it appeared in the log line (node-runner
    /// always logs its own short, scope-stripped name — see `spawn_node_runner`'s `--project`
    /// doc comment on why the CLI passes it that way).
    fn node(&self) -> &str {
        match self {
            HealthEvent::Transition { node, .. }
            | HealthEvent::RestartAttempt { node, .. }
            | HealthEvent::WillRestart { node, .. } => node,
        }
    }

    /// Render this event as a [`LogBuffer`] line. Tagged `HEALTH:` so `DevModeTui`'s log sidebar
    /// (`ui/dev_tui.rs::draw_logs_sidebar`) can style it distinctly from lifecycle `MODE:`/spawn
    /// `STARTED`/`STOPPED` lines already pushed there.
    fn format(&self) -> String {
        match self {
            HealthEvent::Transition { node, running: true } => format!("🩺 HEALTH: {} is Running", node),
            HealthEvent::Transition { node, running: false } => format!("🩺 HEALTH: {} is Stopped", node),
            HealthEvent::RestartAttempt { node, attempt } => {
                format!("🔄 HEALTH: {} restart attempt {}", node, attempt)
            }
            HealthEvent::WillRestart { node, detail } => format!("🔄 HEALTH: {} will restart in {}", node, detail),
        }
    }
}

/// Find `marker` in `line`, then split the text after it into (name-up-to-next-quote, rest).
/// Shared by the `"Node '<name>' ..."` and `"... for '<name>'"` shapes below.
fn extract_quoted_after<'a>(line: &'a str, marker: &str) -> Option<(&'a str, &'a str)> {
    let start = line.find(marker)? + marker.len();
    let rest = &line[start..];
    let end = rest.find('\'')?;
    Some((&rest[..end], &rest[end + 1..]))
}

/// Parse a `"Node '{node}' health: Running"` / `"...: Stopped"` line — see
/// `supervisor.rs::monitor_with_signals_unix`'s health-check branch.
fn parse_transition(line: &str) -> Option<HealthEvent> {
    let (node, after) = extract_quoted_after(line, "Node '")?;
    if after.starts_with(" health: Running") {
        Some(HealthEvent::Transition {
            node: node.to_string(),
            running: true,
        })
    } else if after.starts_with(" health: Stopped") {
        Some(HealthEvent::Transition {
            node: node.to_string(),
            running: false,
        })
    } else {
        None
    }
}

/// Parse a `"Restart attempt {attempt} for '{node}'"` line — see `supervisor.rs::supervise`'s
/// restart-attempt log.
fn parse_restart_attempt(line: &str) -> Option<HealthEvent> {
    let marker = "Restart attempt ";
    let start = line.find(marker)? + marker.len();
    let rest = &line[start..];
    let sep = rest.find(" for '")?;
    let attempt: u32 = rest[..sep].trim().parse().ok()?;
    let (node, _) = extract_quoted_after(&rest[sep..], "for '")?;
    Some(HealthEvent::RestartAttempt {
        node: node.to_string(),
        attempt,
    })
}

/// Parse a `"Node '{node}' will restart in {detail}"` line — see `supervisor.rs::supervise`'s
/// backoff log, logged just before `sleep(backoff_duration)`.
fn parse_will_restart(line: &str) -> Option<HealthEvent> {
    let (node, after) = extract_quoted_after(line, "Node '")?;
    let detail = after.strip_prefix(" will restart in ")?;
    Some(HealthEvent::WillRestart {
        node: node.to_string(),
        detail: detail.trim().to_string(),
    })
}

/// Parse a single tracing log line for any of the node-runner health/restart events this module
/// relays. Pure and I/O-free so it's unit-testable against raw strings, independent of
/// `tracing_subscriber`'s actual output format (timestamp/level/target prefix) — matching is by
/// substring search, not a fixed prefix, so it doesn't care what comes before the message.
pub(crate) fn parse_health_line(line: &str) -> Option<HealthEvent> {
    parse_transition(line)
        .or_else(|| parse_restart_attempt(line))
        .or_else(|| parse_will_restart(line))
}

/// Tails each running node's `logs/<node>.log` file for new node-runner health/restart lines and
/// relays them into a [`LogBuffer`]. Owns one read offset per node so repeated polls only look at
/// bytes appended since the last poll, and only acts on complete (newline-terminated) lines —
/// leaving a partial trailing line for the next poll rather than mis-splitting it.
pub struct HealthTailer {
    /// Node identifier (as passed to `poll`, e.g. `@mecha10/camera`) -> byte offset already
    /// consumed in its log file.
    offsets: HashMap<String, u64>,
    /// Throttles actual file I/O — `poll` is cheap to call every TUI redraw tick (~100ms, see
    /// `ops/interactive.rs`'s event loop), but re-opening every running node's log file that
    /// often is wasteful, so real work only happens once per `min_interval`.
    min_interval: Duration,
    last_poll: Option<Instant>,
}

impl Default for HealthTailer {
    fn default() -> Self {
        Self::new(Duration::from_millis(500))
    }
}

impl HealthTailer {
    /// Create a tailer that does real file I/O at most once per `min_interval`.
    pub fn new(min_interval: Duration) -> Self {
        Self {
            offsets: HashMap::new(),
            min_interval,
            last_poll: None,
        }
    }

    /// Check every node in `node_identifiers` for newly appended health/restart lines and relay
    /// them into `log_buffer`. No-ops (without touching the filesystem) if called again before
    /// `min_interval` has elapsed since the last real poll — cheap enough to call unconditionally
    /// from the TUI's redraw loop.
    pub fn poll(&mut self, node_identifiers: &[String], log_buffer: &LogBuffer) {
        self.poll_in_dir(node_identifiers, Path::new(paths::project::LOGS_DIR), log_buffer);
    }

    /// Throttle-checked poll against an explicit `logs_dir` - split out from `poll` so tests can
    /// exercise the throttling behavior against a temp directory instead of the process CWD's
    /// real `logs/` folder.
    fn poll_in_dir(&mut self, node_identifiers: &[String], logs_dir: &Path, log_buffer: &LogBuffer) {
        if let Some(last) = self.last_poll {
            if last.elapsed() < self.min_interval {
                return;
            }
        }
        self.last_poll = Some(Instant::now());

        for node in node_identifiers {
            self.poll_node(node, logs_dir, log_buffer);
        }
    }

    /// Read and process whatever's new in one node's log file. Split out from `poll` with an
    /// explicit `logs_dir` so it's unit-testable against a temp directory instead of the process
    /// CWD's real `logs/` folder.
    fn poll_node(&mut self, node_identifier: &str, logs_dir: &Path, log_buffer: &LogBuffer) {
        let short_name = strip_node_scope(node_identifier);
        let log_path = logs_dir.join(format!("{}.log", short_name));

        let Ok(mut file) = File::open(&log_path) else {
            // Log file doesn't exist yet (node just started, or disk logging disabled) - nothing
            // to tail yet, try again next poll.
            return;
        };
        let Ok(metadata) = file.metadata() else { return };
        let len = metadata.len();

        let offset = self.offsets.entry(node_identifier.to_string()).or_insert(0);
        // File shrank (rotated/truncated since we last read it) - restart from the top rather
        // than seeking past the end.
        if len < *offset {
            *offset = 0;
        }
        if len == *offset {
            return;
        }

        if file.seek(SeekFrom::Start(*offset)).is_err() {
            return;
        }
        let mut buf = Vec::new();
        if file.read_to_end(&mut buf).is_err() {
            return;
        }
        let text = String::from_utf8_lossy(&buf);

        // Only consume up through the last newline - an in-progress, not-yet-flushed final line
        // is left for the next poll instead of being parsed (and skipped) half-written.
        let Some(last_newline) = text.rfind('\n') else {
            return;
        };
        *offset += (last_newline + 1) as u64;

        for line in text[..last_newline].lines() {
            if let Some(event) = parse_health_line(line) {
                // Defense in depth: only relay events whose parsed node name matches the log
                // file we read them from (node-runner always logs its own short name, so this
                // should always hold - see `HealthEvent::node`'s docs - but a stray matching
                // substring in unrelated output shouldn't get attributed to the wrong node).
                if event.node() == short_name {
                    log_buffer.push(event.format());
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;

    // ---- parse_health_line: transitions -------------------------------------------------

    #[test]
    fn parses_running_transition() {
        let line = "2024-01-01T00:00:00Z  INFO mecha10_node_runner::supervisor: Node 'camera' health: Running";
        assert_eq!(
            parse_health_line(line),
            Some(HealthEvent::Transition {
                node: "camera".to_string(),
                running: true
            })
        );
    }

    #[test]
    fn parses_stopped_transition() {
        let line = "2024-01-01T00:00:00Z  WARN mecha10_node_runner::supervisor: Node 'lidar' health: Stopped";
        assert_eq!(
            parse_health_line(line),
            Some(HealthEvent::Transition {
                node: "lidar".to_string(),
                running: false
            })
        );
    }

    #[test]
    fn parses_bare_message_without_tracing_prefix() {
        // parse_health_line matches by substring search, not a fixed line prefix, so it's
        // independent of whatever timestamp/level/target format tracing_subscriber emits.
        assert_eq!(
            parse_health_line("Node 'camera' health: Running"),
            Some(HealthEvent::Transition {
                node: "camera".to_string(),
                running: true
            })
        );
    }

    // ---- parse_health_line: restart/backoff ----------------------------------------------

    #[test]
    fn parses_restart_attempt() {
        let line = "INFO mecha10_node_runner::supervisor: Restart attempt 2 for 'camera'";
        assert_eq!(
            parse_health_line(line),
            Some(HealthEvent::RestartAttempt {
                node: "camera".to_string(),
                attempt: 2
            })
        );
    }

    #[test]
    fn parses_will_restart_with_debug_formatted_duration() {
        let line = "WARN mecha10_node_runner::supervisor: Node 'camera' will restart in 2s";
        assert_eq!(
            parse_health_line(line),
            Some(HealthEvent::WillRestart {
                node: "camera".to_string(),
                detail: "2s".to_string()
            })
        );
    }

    // ---- parse_health_line: no false positives --------------------------------------------

    #[test]
    fn ignores_unrelated_log_lines() {
        assert_eq!(parse_health_line("Node 'camera' started (PID: Some(1234))"), None);
        assert_eq!(parse_health_line("Starting supervision for 'camera'"), None);
        assert_eq!(parse_health_line(""), None);
    }

    #[test]
    fn ignores_health_line_with_unknown_status_word() {
        assert_eq!(parse_health_line("Node 'camera' health: Unknown"), None);
    }

    // ---- HealthEvent::format ---------------------------------------------------------------

    #[test]
    fn formats_running_transition_with_health_tag() {
        let event = HealthEvent::Transition {
            node: "camera".to_string(),
            running: true,
        };
        assert_eq!(event.format(), "🩺 HEALTH: camera is Running");
    }

    #[test]
    fn formats_stopped_transition_with_health_tag() {
        let event = HealthEvent::Transition {
            node: "camera".to_string(),
            running: false,
        };
        assert_eq!(event.format(), "🩺 HEALTH: camera is Stopped");
    }

    // ---- HealthTailer::poll_node (real file I/O against a temp dir) ------------------------

    fn write_log(dir: &Path, node: &str, content: &str) {
        let path = dir.join(format!("{}.log", node));
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(path)
            .unwrap();
        file.write_all(content.as_bytes()).unwrap();
    }

    #[test]
    fn poll_node_relays_new_health_transition_into_log_buffer() {
        let dir = tempfile::tempdir().unwrap();
        write_log(dir.path(), "camera", "Node 'camera' health: Running\n");

        let mut tailer = HealthTailer::new(Duration::ZERO);
        let log_buffer = LogBuffer::new(100);

        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);

        assert_eq!(log_buffer.get_logs(), vec!["🩺 HEALTH: camera is Running".to_string()]);
    }

    #[test]
    fn poll_node_only_relays_lines_appended_since_last_poll() {
        let dir = tempfile::tempdir().unwrap();
        write_log(dir.path(), "camera", "Node 'camera' health: Running\n");

        let mut tailer = HealthTailer::new(Duration::ZERO);
        let log_buffer = LogBuffer::new(100);
        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);
        assert_eq!(log_buffer.get_logs().len(), 1);

        // No new content - second poll should relay nothing new.
        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);
        assert_eq!(log_buffer.get_logs().len(), 1);

        // Append a transition - only the new line should be relayed.
        write_log(dir.path(), "camera", "Node 'camera' health: Stopped\n");
        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);
        assert_eq!(
            log_buffer.get_logs(),
            vec![
                "🩺 HEALTH: camera is Running".to_string(),
                "🩺 HEALTH: camera is Stopped".to_string(),
            ]
        );
    }

    #[test]
    fn poll_node_leaves_incomplete_trailing_line_for_next_poll() {
        let dir = tempfile::tempdir().unwrap();
        // No trailing newline yet - simulates reading mid-write.
        write_log(dir.path(), "camera", "Node 'camera' health: Run");

        let mut tailer = HealthTailer::new(Duration::ZERO);
        let log_buffer = LogBuffer::new(100);
        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);
        assert!(
            log_buffer.get_logs().is_empty(),
            "partial line should not be parsed yet"
        );

        // The write completes with a newline on the next poll.
        write_log(dir.path(), "camera", "ning\n");
        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);
        assert_eq!(log_buffer.get_logs(), vec!["🩺 HEALTH: camera is Running".to_string()]);
    }

    #[test]
    fn poll_node_is_a_noop_when_log_file_does_not_exist() {
        let dir = tempfile::tempdir().unwrap();
        let mut tailer = HealthTailer::new(Duration::ZERO);
        let log_buffer = LogBuffer::new(100);

        tailer.poll_node("@mecha10/nonexistent", dir.path(), &log_buffer);

        assert!(log_buffer.get_logs().is_empty());
    }

    #[test]
    fn poll_node_ignores_events_for_a_different_node_in_the_same_file() {
        // Defense-in-depth check: a stray line naming a different node inside this node's own
        // log file (e.g. a copy/pasted message) shouldn't be relayed as this node's event.
        let dir = tempfile::tempdir().unwrap();
        write_log(dir.path(), "camera", "Node 'lidar' health: Running\n");

        let mut tailer = HealthTailer::new(Duration::ZERO);
        let log_buffer = LogBuffer::new(100);
        tailer.poll_node("@mecha10/camera", dir.path(), &log_buffer);

        assert!(log_buffer.get_logs().is_empty());
    }

    #[test]
    fn poll_throttles_real_polls_to_min_interval() {
        let dir = tempfile::tempdir().unwrap();
        write_log(dir.path(), "camera", "Node 'camera' health: Running\n");

        // A large min_interval means the second `poll` call (immediately after the first)
        // should not re-read the file even though new content was appended in between.
        let mut tailer = HealthTailer::new(Duration::from_secs(3600));
        let log_buffer = LogBuffer::new(100);
        let nodes = vec!["@mecha10/camera".to_string()];

        tailer.poll_in_dir(&nodes, dir.path(), &log_buffer);
        assert_eq!(log_buffer.get_logs().len(), 1);

        write_log(dir.path(), "camera", "Node 'camera' health: Stopped\n");
        tailer.poll_in_dir(&nodes, dir.path(), &log_buffer);
        assert_eq!(
            log_buffer.get_logs().len(),
            1,
            "throttled poll should not have read the new line"
        );
    }
}