mecha10-node-runner 0.8.0

//! Node lifecycle manager - handles pre-flight checks and node execution
//!
//! This binary is responsible for:
//! - Running pre-flight system checks (Redis, disk space, ports, permissions, GPU)
//! - Loading node configuration
//! - Checking and pulling missing AI models
//! - Setting up environment variables
//! - Supervising node execution with restart policies and health monitoring
//! - Executing the actual node binary
//!
//! # Usage
//!
//! ```bash
//! # Development mode
//! mecha10-node-runner --project rover-robot @mecha10/camera
//!
//! # With restart policy
//! mecha10-node-runner --restart-policy on-failure --max-retries 5 @mecha10/camera
//!
//! # With health monitoring
//! mecha10-node-runner --health-check-interval 30 @mecha10/camera
//!
//! # Skip pre-flight checks (for debugging)
//! mecha10-node-runner --skip-preflight @mecha10/camera
//!
//! # Require GPU availability
//! mecha10-node-runner --require-gpu @mecha10/camera
//! ```
//!
//! # Design
//!
//! This binary acts as a wrapper around node binaries, ensuring all prerequisites
//! are met before node execution. It handles:
//! - Pre-flight system checks (Redis, disk, ports, permissions, GPU)
//! - Model downloads (lazy loading from HuggingFace)
//! - Config loading with fallback chains
//! - Binary path resolution (cached standalone binary, local `nodes/<name>` build, or a
//!   `mecha10-api` download - see `resolve_node_execution`)
//! - Environment setup for consistent execution
//! - Restart policies with exponential backoff
//! - Process health monitoring
//!
//! See NODE_LIFECYCLE_ARCHITECTURE.md for complete design rationale.

mod limits;
mod preflight;
mod supervisor;

use anyhow::{Context as _, Result};
use clap::Parser;
use limits::ResourceLimits;
use mecha10_cli::services::ModelService;
use mecha10_core::context::Context;
use mecha10_core::wire_identifier::short_name as node_short_name;
use std::collections::HashMap;
use std::path::PathBuf;
use supervisor::{RestartPolicy, Supervisor};
use tracing::{info, warn};

#[derive(Parser)]
#[command(name = "mecha10-node-runner")]
#[command(about = "Node lifecycle manager - handles pre-flight checks and execution")]
struct Args {
    node_name: String,

    /// Accepted for backward compatibility with existing callers (`main.rs.template`,
    /// `ProcessService::spawn_node_runner`) - no longer consulted. Resolution used to delegate
    /// to a `target/{release,debug}/<project>` monorepo binary's own `node <name>` subcommand,
    /// but that subcommand is itself just a thin wrapper around spawning
    /// `mecha10-node-runner`, so resolving back to it here would recurse forever; see
    /// `resolve_node_execution`.
    #[arg(long)]
    #[allow(dead_code)]
    project: String,

    #[arg(long)]
    instance_id: Option<String>,

    /// `video-streamer`-specific relay/diagnostics identity override (LAB-4747) - independent
    /// of `instance_id`. Sets `MECHA10_CAMERA_ID` on the spawned node's environment.
    #[arg(long)]
    camera_id: Option<String>,

    #[arg(long, default_value = "never")]
    restart_policy: String,

    #[arg(long, default_value = "3")]
    max_retries: u32,

    #[arg(long, default_value = "1")]
    backoff: u64,

    #[arg(long, default_value = "30")]
    health_check_interval: u64,

    #[arg(long, default_value = "false")]
    wait_for_deps: bool,

    #[arg(long)]
    startup_timeout: Option<u64>,

    #[arg(long, default_value = "false")]
    skip_preflight: bool,

    #[arg(long, default_value = "false")]
    require_gpu: bool,

    #[arg(long, default_value = "0")]
    memory_limit: u64,

    #[arg(long, default_value = "0")]
    cpu_limit: f32,
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::from_default_env()
                .add_directive(tracing::Level::INFO.into())
                .add_directive("rtc_dtls=error".parse().unwrap()),
        )
        .init();

    let args = Args::parse();
    info!("Node Runner: {}", args.node_name);

    let short_name = node_short_name(&args.node_name).to_string();

    let ctx = Context::new("node-runner")
        .await
        .context("Failed to initialize context")?;

    let config: NodeConfig = ctx
        .load_node_config_for_instance(&short_name, args.instance_id.as_deref())
        .await
        .context("Failed to load node config")?;

    let redis_url = Context::get_redis_url().context("Failed to get Redis URL")?;
    tracing::debug!("Redis URL: {}", redis_url);

    if !args.skip_preflight {
        let preflight_config = config.preflight.as_ref();
        let preflight_check = preflight::PreflightConfig {
            check_ports: preflight_config.map(|p| p.check_ports.clone()).unwrap_or_default(),
            min_disk_space_gb: preflight_config.map(|p| p.min_disk_space_gb).unwrap_or(1),
            require_gpu: args.require_gpu || preflight_config.map(|p| p.require_gpu).unwrap_or(false),
            redis_url: redis_url.clone(),
        };

        preflight::run_preflight_checks(&preflight_check)
            .await
            .context("Pre-flight checks failed")?;
    } else {
        info!("⚠️  Skipping pre-flight checks (--skip-preflight enabled)");
    }

    if args.wait_for_deps && !config.depends_on.is_empty() {
        let timeout_secs = args.startup_timeout.unwrap_or(config.startup_timeout_secs);
        let timeout = std::time::Duration::from_secs(timeout_secs);

        info!("Waiting for dependencies: {:?}", config.depends_on);
        wait_for_dependencies(&config.depends_on, timeout, &redis_url)
            .await
            .context("Failed to wait for dependencies")?;
        info!("All dependencies ready");
    }

    if let Some(model_config) = config.model.as_ref() {
        if let Some(model_name) = model_config.name.as_ref() {
            if model_config.provider.as_deref() == Some("local") {
                ensure_ollama_model_available(model_name).await;
            } else {
                ensure_model_available(model_name, model_config.source(), model_config.version)
                    .await
                    .with_context(|| format!("Failed to ensure model '{}' is available", model_name))?;
            }
        }
    }

    let (binary_path, node_args) = resolve_node_execution(&args.node_name).await?;

    let mut env = HashMap::new();

    env.insert("NODE_NAME".to_string(), args.node_name.clone());
    if let Some(instance_id) = args.instance_id.as_ref() {
        env.insert("NODE_INSTANCE_ID".to_string(), instance_id.clone());
    }
    if let Some(camera_id) = args.camera_id.as_ref() {
        env.insert("MECHA10_CAMERA_ID".to_string(), camera_id.clone());
    }
    env.insert("REDIS_URL".to_string(), redis_url);
    if let Ok(rust_log) = std::env::var("RUST_LOG") {
        env.insert("RUST_LOG".to_string(), rust_log);
    }

    if let Ok(url) = std::env::var("CONTROL_PLANE_URL") {
        env.insert("CONTROL_PLANE_URL".to_string(), url);
    }
    if let Ok(robot_id) = std::env::var("ROBOT_ID") {
        env.insert("ROBOT_ID".to_string(), robot_id);
    }
    if let Ok(api_key) = std::env::var("ROBOT_API_KEY") {
        env.insert("ROBOT_API_KEY".to_string(), api_key);
    }

    setup_log_file(&short_name)?;

    let restart_policy = parse_restart_policy(&args.restart_policy, args.max_retries, args.backoff, &config)?;

    let health_interval = if args.health_check_interval > 0 {
        Some(std::time::Duration::from_secs(args.health_check_interval))
    } else {
        None
    };

    let resource_limits = parse_resource_limits(&args, &config)?;

    let supervisor = Supervisor::new(restart_policy, health_interval, resource_limits);

    supervisor.supervise(&short_name, &binary_path, &node_args, env).await?;

    Ok(())
}

/// Ensure model is available (download if missing)
async fn ensure_model_available(model_name: &str, source: &str, version: Option<i32>) -> Result<()> {
    use indicatif::ProgressBar;

    let model_service = ModelService::with_models_dir("models".into())?;

    match source {
        "registry" => {
            let version = version.with_context(|| {
                format!(
                    "Model '{}' has source \"registry\" but no version configured - a registry model requires an explicit \
                     version (see `models::deployment::push_model_config`/`mecha10 model set`)",
                    model_name
                )
            })?;

            let model_path = model_service.registry_model_dir(model_name, version).join("model.onnx");
            if model_path.exists() {
                info!("Registry model '{}' v{} found", model_name, version);
                return Ok(());
            }

            let base_url = std::env::var("CONTROL_PLANE_URL").context(
                "CONTROL_PLANE_URL must be set to fetch a registry model (source: \"registry\") - it is not required for catalog models",
            )?;
            let api_key = std::env::var("ROBOT_API_KEY").context(
                "ROBOT_API_KEY must be set to fetch a registry model (source: \"registry\") - it is not required for catalog models",
            )?;

            info!("Registry model '{}' v{} not found, downloading...", model_name, version);
            let pb = ProgressBar::new_spinner();
            pb.set_message(format!("Downloading registry model '{}' v{}...", model_name, version));

            model_service
                .pull_registry_version(&base_url, &api_key, model_name, version, Some(&pb))
                .await
                .with_context(|| format!("Failed to pull registry model '{}' v{}", model_name, version))?;

            pb.finish_and_clear();
            info!("Registry model '{}' v{} ready", model_name, version);
        }
        _ => {
            let model_path = PathBuf::from(format!("models/catalog/{}/model.onnx", model_name));

            if model_path.exists() {
                info!("Model '{}' found", model_name);
                return Ok(());
            }

            info!("Model '{}' not found, downloading...", model_name);

            let pb = ProgressBar::new_spinner();
            pb.set_message(format!("Downloading model '{}'...", model_name));

            model_service
                .pull(model_name, Some(&pb))
                .await
                .with_context(|| format!("Failed to pull model '{}'", model_name))?;

            pb.finish_and_clear();
            info!("Model '{}' ready", model_name);
        }
    }

    Ok(())
}

/// Ensure an Ollama model is pulled before spawning a node configured with `"provider": "local"`.
/// Mirrors [`ensure_model_available`]'s "check first, download if missing" shape, but against
/// Ollama's own model store rather than the ONNX catalog's file layout - that store is
/// bind-mounted at `<project_root>/models/catalog` (the same directory the ONNX catalog above
/// uses) by the `ollama` Compose service, see `mecha10_core::ollama`.
///
/// Warns and returns (never fails the node's startup) if Ollama isn't reachable or rejects the
/// pull - this is a best-effort preflight, not a hard dependency, matching how a missing
/// `OPENAI_API_KEY`/`ANTHROPIC_API_KEY` is handled by the CLI's `run` handler rather than here.
async fn ensure_ollama_model_available(model_name: &str) {
    let base_url = std::env::var("OLLAMA_ENDPOINT")
        .map(|endpoint| endpoint.trim_end_matches("/api/chat").to_string())
        .unwrap_or_else(|_| "http://localhost:11434".to_string());

    let client = reqwest::Client::new();

    let installed = match client.get(format!("{base_url}/api/tags")).send().await {
        Ok(response) => response
            .json::<serde_json::Value>()
            .await
            .ok()
            .map(|body| mecha10_core::ollama::parse_installed_model_names(&body))
            .unwrap_or_default()
            .iter()
            .any(|name| name == model_name || name.starts_with(&format!("{model_name}:"))),
        Err(e) => {
            warn!("Could not reach Ollama at {base_url} to check model '{model_name}': {e}");
            return;
        }
    };

    if installed {
        info!("Ollama model '{}' found", model_name);
        return;
    }

    info!("Ollama model '{}' not found, pulling...", model_name);
    let pull_result = client
        .post(format!("{base_url}/api/pull"))
        .json(&serde_json::json!({ "model": model_name, "stream": false }))
        .send()
        .await;

    match pull_result {
        Ok(response) if response.status().is_success() => info!("Ollama model '{}' ready", model_name),
        Ok(response) => warn!("Ollama rejected pulling model '{}': {}", model_name, response.status()),
        Err(e) => warn!("Failed to pull Ollama model '{}': {}", model_name, e),
    }
}

/// Wait for dependencies to be ready using Redis ping/pong
///
/// For each dependency:
/// 1. Try to ping via Redis (publish to /_internal/health/ping)
/// 2. Wait for response on /_internal/health/{dep}/status
/// 3. If timeout, fail with clear error message
///
/// This is a simple check - just verifies the dependency node is responsive via Redis.
/// Full health system integration is deferred to P2.
async fn wait_for_dependencies(deps: &[String], timeout: std::time::Duration, redis_url: &str) -> Result<()> {
    use redis::AsyncCommands;
    use std::time::Instant;

    let client = redis::Client::open(redis_url).context("Failed to connect to Redis")?;
    let mut conn = client
        .get_multiplexed_async_connection()
        .await
        .context("Failed to get Redis connection")?;

    let start = Instant::now();

    for dep_name in deps {
        info!("Checking dependency: {}", dep_name);

        let dep_timeout = timeout
            .checked_sub(start.elapsed())
            .ok_or_else(|| anyhow::anyhow!("Timeout waiting for dependency: {}", dep_name))?;

        let check_start = Instant::now();
        let check_interval = std::time::Duration::from_millis(500);

        loop {
            let key = format!("/_internal/node/{}/alive", dep_name);
            let exists: bool = conn.exists(&key).await.unwrap_or(false);

            if exists {
                info!("Dependency {} is ready", dep_name);
                break;
            }

            if check_start.elapsed() >= dep_timeout {
                return Err(anyhow::anyhow!(
                    "Timeout waiting for dependency '{}' after {:?}. \
                     Ensure the '{}' node is running and responsive.",
                    dep_name,
                    timeout,
                    dep_name
                ));
            }

            tokio::time::sleep(check_interval).await;
        }
    }

    Ok(())
}

/// Strip an optional `<scope>/` namespace prefix (e.g. `@mecha10/motor` -> `motor`), if any.
/// Resolve node execution details (binary path + args) for whatever node name mecha10.json
/// listed - there is no upfront "is this a framework node" classification, and no delegating
/// to the project's own binary (that binary's `node <name>` subcommand is itself just a thin
/// wrapper that spawns `mecha10-node-runner` - see `run_single_node` in `main.rs.template` -
/// so resolving back to it here would recurse forever). Resolution is simply: use a
/// standalone binary already cached for this node; failing that, build it from a local
/// `nodes/<name>` crate if `mecha10_cli_core::types::is_local_project_node` says this project
/// declares one (the same check `mecha10 dev`'s `ProcessService::spawn_node_direct` uses, so
/// the two can't drift onto different classification rules for the same node); failing that,
/// ask `mecha10-api` for a standalone pre-built binary.
async fn resolve_node_execution(node_name: &str) -> Result<(String, Vec<String>)> {
    let short_name = node_short_name(node_name);

    let local_binary = format!("target/release/{}", short_name);
    if PathBuf::from(&local_binary).exists() {
        info!("Using local binary: {}", local_binary);
        return Ok((local_binary, vec![]));
    }

    let local_debug = format!("target/debug/{}", short_name);
    if PathBuf::from(&local_debug).exists() {
        info!("Using local debug binary: {}", local_debug);
        return Ok((local_debug, vec![]));
    }

    if mecha10_cli_core::types::is_local_project_node(std::path::Path::new("mecha10.json"), short_name) {
        info!("🔨 Building local node: {}", short_name);
        let build_output = std::process::Command::new("cargo")
            .args(["build", "-p", short_name])
            .output()
            .context("Failed to run cargo build")?;

        if !build_output.status.success() {
            let stderr = String::from_utf8_lossy(&build_output.stderr);
            anyhow::bail!("Failed to build node '{}': {}", short_name, stderr);
        }

        info!("✅ Built local node: {}", short_name);

        return if PathBuf::from(&local_debug).exists() {
            Ok((local_debug, vec![]))
        } else {
            anyhow::bail!(
                "Binary not found at '{}' after build. Package name may differ from node name.",
                local_debug
            )
        };
    }

    // Not compiled into or buildable from this project - resolve (downloading if needed) as
    // a standalone binary mecha10-api distributes under this short name.
    let resolver = mecha10_node_resolver::NodeResolver::new().context("Failed to initialize node resolver")?;
    let binary_path = resolver
        .resolve(short_name)
        .await
        .with_context(|| format!("Failed to resolve node '{}'", node_name))?;
    info!("Using resolved node binary: {}", binary_path.display());
    Ok((binary_path.to_string_lossy().to_string(), vec![]))
}

/// Setup log file redirection
///
/// Skipped entirely when the parent CLI process has disabled disk logging
/// (`MECHA10_DISK_LOGGING_ENABLED=false`, set from the project's current
/// environment's `environments.<env>.logging.disk_logging_enabled` config) -
/// no `logs/` directory is created and no file handle is opened, on
/// edge/robot deployments that already ship logs to a remote server.
fn setup_log_file(node_name: &str) -> Result<()> {
    let disk_logging_enabled = std::env::var("MECHA10_DISK_LOGGING_ENABLED")
        .map(|v| v != "false")
        .unwrap_or(true);

    if !disk_logging_enabled {
        info!("Disk logging disabled via project config; skipping log file setup");
        return Ok(());
    }

    let logs_dir = PathBuf::from("logs");
    std::fs::create_dir_all(&logs_dir).with_context(|| format!("Failed to create logs directory at {:?}", logs_dir))?;

    let log_path = logs_dir.join(format!("{}.log", node_name));
    std::env::set_var("MECHA10_LOG_FILE", log_path.to_string_lossy().as_ref());

    info!("Logs: logs/{}.log", node_name);

    Ok(())
}

/// Parse restart policy from CLI args and node config
fn parse_restart_policy(
    policy_str: &str,
    max_retries: u32,
    backoff_secs: u64,
    config: &NodeConfig,
) -> Result<RestartPolicy> {
    use supervisor::PolicyType;

    let policy_type = if let Some(runtime) = &config.runtime {
        if let Some(policy) = &runtime.restart_policy {
            match policy.as_str() {
                "never" => PolicyType::Never,
                "on-failure" => PolicyType::OnFailure,
                "always" => PolicyType::Always,
                _ => anyhow::bail!("Invalid restart_policy in config: {}", policy),
            }
        } else {
            match policy_str {
                "never" => PolicyType::Never,
                "on-failure" => PolicyType::OnFailure,
                "always" => PolicyType::Always,
                _ => anyhow::bail!("Invalid restart_policy: {}", policy_str),
            }
        }
    } else {
        match policy_str {
            "never" => PolicyType::Never,
            "on-failure" => PolicyType::OnFailure,
            "always" => PolicyType::Always,
            _ => anyhow::bail!("Invalid restart_policy: {}", policy_str),
        }
    };

    let max_retries = config
        .runtime
        .as_ref()
        .and_then(|r| r.max_retries)
        .unwrap_or(max_retries);

    let backoff = config
        .runtime
        .as_ref()
        .and_then(|r| r.backoff_secs)
        .map(std::time::Duration::from_secs)
        .unwrap_or_else(|| std::time::Duration::from_secs(backoff_secs));

    Ok(RestartPolicy {
        policy: policy_type,
        max_retries,
        backoff,
    })
}

#[derive(serde::Deserialize, Default)]
struct NodeConfig {
    #[serde(default, deserialize_with = "deserialize_optional_model_config")]
    model: Option<ModelConfig>,
    #[serde(default)]
    runtime: Option<RuntimeConfig>,
    #[serde(default)]
    preflight: Option<PreflightConfigData>,
    #[serde(default)]
    depends_on: Vec<String>,
    #[serde(default = "default_startup_timeout")]
    startup_timeout_secs: u64,
}

#[derive(serde::Deserialize)]
struct ModelConfig {
    name: Option<String>,
    #[serde(default)]
    source: Option<String>,
    #[serde(default)]
    version: Option<i32>,
    #[serde(default)]
    provider: Option<String>,
}

/// Deserializes `NodeConfig.model`, tolerating a raw `"model"` value that isn't a JSON
/// object (e.g. a plain string like the `simulator` node's own `model: "rover"` field).
///
/// Only a JSON object is interpreted as an AI-model-catalog reference and deserialized
/// into `ModelConfig` (surfacing an error if it's malformed); anything else - string,
/// number, bool, array, or `null` - resolves to `None`, since it means "this node has no
/// AI-model metadata under this key" rather than "this AI-model config is broken".
fn deserialize_optional_model_config<'de, D>(deserializer: D) -> Result<Option<ModelConfig>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    let value = <serde_json::Value as serde::Deserialize>::deserialize(deserializer)?;
    match value {
        serde_json::Value::Object(_) => serde_json::from_value(value)
            .map(Some)
            .map_err(serde::de::Error::custom),
        _ => Ok(None),
    }
}

impl ModelConfig {
    /// Normalized source, defaulting an absent/unrecognized value to
    /// `"catalog"` so every config written before this field existed keeps
    /// resolving exactly as it did before (`LAB-1802`).
    fn source(&self) -> &str {
        match self.source.as_deref() {
            Some("registry") => "registry",
            _ => "catalog",
        }
    }
}

#[derive(serde::Deserialize)]
struct RuntimeConfig {
    restart_policy: Option<String>,
    max_retries: Option<u32>,
    backoff_secs: Option<u64>,
    memory_limit_mb: Option<u64>,
    cpu_limit_percent: Option<f32>,
}

#[derive(serde::Deserialize, Default)]
struct PreflightConfigData {
    #[serde(default)]
    check_ports: Vec<u16>,
    #[serde(default = "default_min_disk_space")]
    min_disk_space_gb: u64,
    #[serde(default)]
    require_gpu: bool,
}

fn default_startup_timeout() -> u64 {
    30
}

fn default_min_disk_space() -> u64 {
    1
}

/// Parse resource limits from CLI args and node config
///
/// Priority: CLI args > config file > defaults (no limits)
fn parse_resource_limits(args: &Args, config: &NodeConfig) -> Result<ResourceLimits> {
    let memory_mb = if args.memory_limit > 0 {
        Some(args.memory_limit)
    } else {
        config.runtime.as_ref().and_then(|r| r.memory_limit_mb)
    };

    let cpu_percent = if args.cpu_limit > 0.0 {
        if args.cpu_limit > 100.0 {
            anyhow::bail!("CPU limit must be between 0 and 100, got {}", args.cpu_limit);
        }
        Some(args.cpu_limit)
    } else {
        config.runtime.as_ref().and_then(|r| r.cpu_limit_percent)
    };

    if memory_mb.is_some() || cpu_percent.is_some() {
        info!("Resource limits configured:");
        if let Some(mb) = memory_mb {
            info!("  Memory: {} MB", mb);
        }
        if let Some(percent) = cpu_percent {
            info!("  CPU: {:.1}%", percent);
        }
    }

    Ok(ResourceLimits { memory_mb, cpu_percent })
}

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

    static CWD_ENV_MUTEX: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

    #[tokio::test]
    async fn uses_a_cached_standalone_binary_when_present() {
        let _guard = CWD_ENV_MUTEX.lock().await;

        let original_dir = std::env::current_dir().unwrap();
        let temp_dir = tempfile::tempdir().unwrap();

        std::fs::create_dir_all(temp_dir.path().join("target/release")).unwrap();
        std::fs::write(temp_dir.path().join("target/release/speaker"), b"fake-binary").unwrap();
        std::env::set_current_dir(temp_dir.path()).unwrap();

        let result = resolve_node_execution("@mecha10/speaker").await;

        std::env::set_current_dir(&original_dir).unwrap();

        let (binary_path, args) = result.expect("should resolve to the cached standalone binary");
        assert!(
            binary_path.ends_with("speaker"),
            "expected the cached standalone binary path, got: {}",
            binary_path
        );
        assert_eq!(args, Vec::<String>::new());
    }

    #[test]
    fn a_node_declared_as_project_source_is_recognized_as_local() {
        // `resolve_node_execution`'s local-build branch defers entirely to
        // `mecha10_cli_core::types::is_local_project_node` - the same check `mecha10 dev`'s
        // `ProcessService::spawn_node_direct` uses - rather than a node-runner-specific
        // directory check, so the two can't drift onto different classification rules for
        // the same node. This exercises that wiring directly against a real mecha10.json
        // shape; the function's own classification logic is covered in
        // `mecha10-cli-core`'s tests.
        let temp_dir = tempfile::tempdir().unwrap();
        let config_path = temp_dir.path().join("mecha10.json");
        std::fs::write(
            &config_path,
            r#"{"name":"some-project","version":"0.1.0","nodes":{"my-custom-node":"0.1.0","@mecha10/speaker":"0.1.0"}}"#,
        )
        .unwrap();

        assert!(mecha10_cli_core::types::is_local_project_node(
            &config_path,
            "my-custom-node"
        ));
        assert!(!mecha10_cli_core::types::is_local_project_node(
            &config_path,
            "@mecha10/speaker"
        ));
    }

    #[tokio::test]
    async fn no_local_signal_falls_back_to_the_node_resolver() {
        // No cached standalone binary, no `nodes/<name>` source directory - the only thing
        // left to try is resolving (and downloading if needed) a standalone binary via
        // `mecha10-api`, regardless of what the node's name looks like.
        let _guard = CWD_ENV_MUTEX.lock().await;

        let original_dir = std::env::current_dir().unwrap();
        let original_home = std::env::var("HOME").ok();
        let temp_dir = tempfile::tempdir().unwrap();

        std::env::set_current_dir(temp_dir.path()).unwrap();
        std::env::set_var("HOME", temp_dir.path());
        std::env::set_var("MECHA10_API_URL", "http://127.0.0.1:1");

        let result = resolve_node_execution("diagnostics").await;

        std::env::set_current_dir(&original_dir).unwrap();
        match original_home {
            Some(home) => std::env::set_var("HOME", home),
            None => std::env::remove_var("HOME"),
        }
        std::env::remove_var("MECHA10_API_URL");

        let err = result.expect_err("resolve should fail with no reachable mecha10-api and no cached binary");
        assert!(
            err.to_string().contains("Failed to resolve node 'diagnostics'"),
            "expected diagnostics to be routed through NodeResolver, got: {}",
            err
        );
        assert!(
            !err.to_string().contains("Node binary not found"),
            "diagnostics must not fall through to the 'Node binary not found' local-node path: {}",
            err
        );
    }
}

#[cfg(test)]
mod model_source_tests {
    use super::ModelConfig;

    fn config(source: Option<&str>, version: Option<i32>) -> ModelConfig {
        ModelConfig {
            name: Some("test-model".to_string()),
            source: source.map(str::to_string),
            version,
            provider: None,
        }
    }

    #[test]
    fn defaults_to_catalog_when_source_is_absent() {
        assert_eq!(config(None, None).source(), "catalog");
    }

    #[test]
    fn defaults_to_catalog_for_an_unrecognized_source_value() {
        assert_eq!(config(Some("bogus"), None).source(), "catalog");
    }

    #[test]
    fn resolves_registry_source_explicitly() {
        assert_eq!(config(Some("registry"), Some(3)).source(), "registry");
    }

    #[test]
    fn resolves_catalog_source_explicitly() {
        assert_eq!(config(Some("catalog"), None).source(), "catalog");
    }
}

#[cfg(test)]
mod node_config_model_field_tests {
    use super::NodeConfig;

    /// Regression test for a follow-up to LAB-1888: reverting the `is_local_node` gate made
    /// node-runner supervise every node uniformly, including bundled framework nodes like
    /// `simulator`. `simulator`'s own config has a top-level `model: "rover"` string field
    /// (the physics/robot model name) that collides with `NodeConfig.model`, which expects
    /// an AI-model-catalog object. Deserializing simulator's dev config section through
    /// `NodeConfig` must not fail because of this unrelated field-name collision.
    #[test]
    fn tolerates_simulators_string_model_field() {
        let json = serde_json::json!({
            "model": "rover",
            "runtime": {
                "restart_policy": "on-failure"
            }
        });

        let config: NodeConfig = serde_json::from_value(json).expect("simulator-shaped config should deserialize");
        assert!(config.model.is_none());
    }

    /// Guards against regressing the original AI-model-catalog use case (LAB-1802): a real
    /// AI-model-reference object under `"model"` must still deserialize into `ModelConfig`.
    #[test]
    fn still_parses_a_real_ai_model_config_object() {
        let json = serde_json::json!({
            "model": {
                "name": "yolov8n"
            }
        });

        let config: NodeConfig = serde_json::from_value(json).expect("AI-model config should deserialize");
        let model = config.model.expect("model should be Some for an object value");
        assert_eq!(model.name.as_deref(), Some("yolov8n"));
        assert_eq!(model.source(), "catalog");
    }

    /// A malformed AI-model *object* (wrong field type) should still surface as an error,
    /// since that indicates a genuine problem with an actual AI-model config, not an
    /// unrelated node using the same JSON key for something else.
    #[test]
    fn still_errors_on_a_malformed_ai_model_config_object() {
        let json = serde_json::json!({
            "model": {
                "name": 123
            }
        });

        let result: Result<NodeConfig, _> = serde_json::from_value(json);
        assert!(
            result.is_err(),
            "expected malformed model object to fail deserialization"
        );
    }
}