mecha10-cli 0.6.3

//! Spawning nodes by name for dev mode

use super::models::ensure_node_models_available;
use crate::context::CliContext;
use crate::services::CredentialsService;
use crate::types::{turn_config, ProjectConfig, TurnConfig};
use crate::ui::LogBuffer;
use anyhow::Result;
use serde::Deserialize;
use std::collections::HashMap;

/// Set (to any value) by `DevSession::initialize` when `mecha10 dev --sim-visible` was passed -
/// see [`maybe_inject_sim_visible_env`]. Process-wide rather than threaded as a parameter
/// through every `spawn_nodes_by_name`/`handle_mode_change` call site (initial spawn, every
/// later `t`/`s` interactive mode-change keypress) - those all run within the same `mecha10 dev`
/// process for its whole lifetime, same rationale as `RUST_LOG`/`MECHA10_ENVIRONMENT`.
pub const SIM_VISIBLE_ENV_VAR: &str = "MECHA10_DEV_SIM_VISIBLE";

/// Environment variable the `simulator` node's own `SimulatorNode::init` reads to override
/// `SimulatorConfig.visible = true` - see `packages/nodes/simulator/src/lib.rs`.
const SIMULATOR_VISIBLE_ENV_VAR: &str = "MECHA10_SIMULATOR_VISIBLE";

/// Env vars the simulator subprocess's `serve` command (mecha10-rl, LAB-1976) reads to override
/// `model.json`'s `locomotion.max_linear_speed`/`max_angular_speed` when generating the MJCF
/// actuator `ctrlrange`/`forcerange`. See [`maybe_inject_teleop_locomotion_env`].
const LOCOMOTION_MAX_LINEAR_SPEED_ENV_VAR: &str = "MECHA10_LOCOMOTION_MAX_LINEAR_SPEED";
const LOCOMOTION_MAX_ANGULAR_SPEED_ENV_VAR: &str = "MECHA10_LOCOMOTION_MAX_ANGULAR_SPEED";

/// Env var the simulator subprocess's `serve`/`generate_robot_mjcf` (mecha10-rl) reads to
/// *optionally* override `model.json`'s `locomotion.acceleration`. Unlike
/// [`LOCOMOTION_MAX_LINEAR_SPEED_ENV_VAR`]/[`LOCOMOTION_MAX_ANGULAR_SPEED_ENV_VAR`], this is
/// never mandatory - `model.json` remains a complete, valid source of truth for acceleration on
/// its own, so there is deliberately no `DEFAULT_LOCOMOTION_ACCELERATION` compiled-in constant
/// here to fall back to. See [`inject_teleop_locomotion_env`] for the injection rules.
const LOCOMOTION_ACCELERATION_ENV_VAR: &str = "MECHA10_LOCOMOTION_ACCELERATION";

/// Compiled-in fallback for [`LOCOMOTION_MAX_LINEAR_SPEED_ENV_VAR`]/
/// [`LOCOMOTION_MAX_ANGULAR_SPEED_ENV_VAR`] when a `simulator` node is spawned without a `teleop`
/// node (or teleop's config resolves to its own hardcoded-defaults fallback) - see
/// [`inject_teleop_locomotion_env`]. Matches `model.json`'s former `locomotion.max_linear_speed`/
/// `max_angular_speed` bundled defaults exactly (LAB-1985), so removing those fields from
/// `model.json` (now that `generate_robot_mjcf`, per LAB-1978, requires the env vars or an
/// explicit flag) is not a behavior change for the common bare-simulator case.
const DEFAULT_LOCOMOTION_MAX_LINEAR_SPEED: f32 = 0.5;
const DEFAULT_LOCOMOTION_MAX_ANGULAR_SPEED: f32 = 1.0;

/// Env vars consumed by `TurnServerConfig::from_env` (`packages/video/src/webrtc/connection.rs`)
/// to add a TURN entry to the ICE server list. `TURN_URL_ENV_VAR` doubles as the "operator
/// already configured this manually" signal (LAB-1915): if it's already set in this process's
/// env, both the build-time-baked default and the auto-fetch below are skipped entirely so an
/// explicit export always wins - mirroring how `ProjectEnvironments::relay_url()` treats an
/// explicit `WEBRTC_RELAY_URL` as authoritative over its own derived default. See
/// `maybe_inject_turn_credentials` for the full three-tier resolution order (LAB-1917).
const TURN_URL_ENV_VAR: &str = "MECHA10_WEBRTC_TURN_URL";
const TURN_USERNAME_ENV_VAR: &str = "MECHA10_WEBRTC_TURN_USERNAME";
const TURN_CREDENTIAL_ENV_VAR: &str = "MECHA10_WEBRTC_TURN_CREDENTIAL";

/// Mirrors mecha10-api's `TurnCredentialsResponse` (LAB-1914, `GET
/// /api/webrtc/turn-credentials`) - kept as a local struct rather than an actual dependency on
/// mecha10-api since the CLI only ever deserializes this one response shape.
#[derive(Debug, Deserialize)]
struct TurnCredentialsResponse {
    url: String,
    username: String,
    credential: String,
}

/// Node names that actually consume `WEBRTC_RELAY_URL`/`MECHA10_WEBRTC_TURN_*` - see
/// `packages/video/src/webrtc/connection.rs`. Gates the TURN credentials fetch below so
/// `mecha10 dev` doesn't make a network round-trip to mecha10-api on every invocation regardless
/// of whether any spawned node would ever read the result (LAB-1915) - same rationale as
/// `inject_sim_visible_env`'s `has_simulator` check just above.
fn batch_needs_turn_credentials(node_names: &[String]) -> bool {
    node_names
        .iter()
        .any(|n| matches!(super::strip_node_scope(n), "camera" | "video-streamer"))
}

/// `Ok(Some(_))` on a well-formed 200 response, `Ok(None)` on 404 (TURN not configured
/// server-side - the expected/common case per LAB-1914, not an error), `Err` for anything else
/// (network failure, auth failure, non-2xx/404 status, unexpected body shape). The caller
/// collapses all three "no credentials" outcomes to the same STUN-only fallback; this split just
/// keeps "not configured" distinguishable for logging and tests.
async fn fetch_turn_credentials(control_plane_url: &str, api_key: &str) -> Result<Option<TurnCredentialsResponse>> {
    let url = format!(
        "{}/api/webrtc/turn-credentials",
        control_plane_url.trim_end_matches('/')
    );
    let response = reqwest::Client::new().get(url).bearer_auth(api_key).send().await?;

    if response.status() == reqwest::StatusCode::NOT_FOUND {
        return Ok(None);
    }

    let creds = response.error_for_status()?.json::<TurnCredentialsResponse>().await?;
    Ok(Some(creds))
}

/// Resolve TURN credentials into `project_env`, highest priority first (LAB-1917):
/// 1. An explicit `MECHA10_WEBRTC_TURN_URL`/`_USERNAME`/`_CREDENTIAL` env var already set in
///    this process's env (today's "operator configured this manually" override, unchanged).
/// 2. The build-time-baked `option_env!` default (`mecha10_cli_core::types::turn_config()`,
///    alongside `relay_url()`) - a working default with no network round-trip, for a release
///    build that had the vars exported during `cargo build`/`cargo install` (see
///    `scripts/release/build/build-all-node-binaries.sh` and
///    `.github/workflows/main.yml`'s `build-cli`/`build-nodes` jobs).
/// 3. LAB-1915's best-effort runtime auto-fetch from mecha10-api's
///    `GET /api/webrtc/turn-credentials` (LAB-1914) - last resort, for builds that didn't have
///    the values baked in (e.g. a local dev build).
///
/// Tiers 2 and 3 never block `mecha10 dev` startup: TURN being unconfigured anywhere, mecha10-api
/// being unreachable, auth failing, or the response shape being unexpected all just print a
/// one-line notice and leave `project_env` exactly as it would be today (STUN-only). Tiers 2 and
/// 3 are both skipped entirely - no work at all - when no video/camera-relevant node is part of
/// this spawn batch (see `batch_needs_turn_credentials`), and tier 3's network call is skipped
/// specifically when tier 1 or 2 already resolved something.
async fn maybe_inject_turn_credentials(
    node_names: &[String],
    control_plane_url: &str,
    api_key: Option<&str>,
    project_env: &mut HashMap<String, String>,
) {
    maybe_inject_turn_credentials_with_baked_resolver(node_names, control_plane_url, api_key, project_env, turn_config)
        .await
}

/// [`maybe_inject_turn_credentials`], but with tier 2's baked-default lookup taken as a
/// parameter instead of calling `mecha10_cli_core::types::turn_config()` directly. That real
/// function is wired to `option_env!` constants fixed at *this crate's own* compile time, so
/// there's no way to make it return `Some(_)` from within a `cargo test` run - injecting the
/// resolver lets tests stub "a baked default is present" to actually exercise tier 2 and its
/// priority over tier 3, rather than only ever observing tier 2 as absent.
async fn maybe_inject_turn_credentials_with_baked_resolver(
    node_names: &[String],
    control_plane_url: &str,
    api_key: Option<&str>,
    project_env: &mut HashMap<String, String>,
    resolve_baked: impl Fn() -> Option<TurnConfig>,
) {
    if std::env::var(TURN_URL_ENV_VAR).is_ok() {
        return;
    }
    if !batch_needs_turn_credentials(node_names) {
        return;
    }

    // Tier 2: build-time-baked default.
    if let Some(baked) = resolve_baked() {
        project_env.insert(TURN_URL_ENV_VAR.to_string(), baked.url);
        project_env.insert(TURN_USERNAME_ENV_VAR.to_string(), baked.username);
        project_env.insert(TURN_CREDENTIAL_ENV_VAR.to_string(), baked.credential);
        return;
    }

    let Some(api_key) = api_key else {
        // No credentials to authenticate with (not logged in) - `ROBOT_API_KEY` injection above
        // already warned about this; nothing more to do here.
        return;
    };

    // Tier 3: last-resort runtime auto-fetch (LAB-1915).
    match fetch_turn_credentials(control_plane_url, api_key).await {
        Ok(Some(creds)) => {
            project_env.insert(TURN_URL_ENV_VAR.to_string(), creds.url);
            project_env.insert(TURN_USERNAME_ENV_VAR.to_string(), creds.username);
            project_env.insert(TURN_CREDENTIAL_ENV_VAR.to_string(), creds.credential);
        }
        Ok(None) => {
            println!("\u{2139}\u{fe0f}  TURN server not available, using STUN only");
        }
        Err(e) => {
            tracing::debug!("TURN credentials fetch failed, using STUN only: {}", e);
            println!("\u{2139}\u{fe0f}  TURN server not available, using STUN only");
        }
    }
}

/// When `mecha10 dev --sim-visible` was passed (`SIM_VISIBLE_ENV_VAR` set), inject
/// `MECHA10_SIMULATOR_VISIBLE=1` into `project_env` if a `simulator` node is actually part of
/// this spawn batch, so the subprocess's own env override in `SimulatorNode::init` picks it up.
/// Otherwise (sim-visible requested but no simulator node here - e.g. hardware mode, or a node
/// subset that doesn't include it) prints a clear no-op message instead of silently doing
/// nothing, per LAB-1886's requirement that `--sim-visible` never has unexplained no effect.
fn maybe_inject_sim_visible_env(node_names: &[String], project_env: &mut HashMap<String, String>) {
    let sim_visible_requested = std::env::var(SIM_VISIBLE_ENV_VAR).is_ok();
    inject_sim_visible_env(sim_visible_requested, node_names, project_env);
}

/// Pure decision logic behind [`maybe_inject_sim_visible_env`], split out so it's unit testable
/// without mutating real process environment variables (which would race across parallel test
/// threads).
fn inject_sim_visible_env(
    sim_visible_requested: bool,
    node_names: &[String],
    project_env: &mut HashMap<String, String>,
) {
    if !sim_visible_requested {
        return;
    }

    let has_simulator = node_names.iter().any(|n| super::strip_node_scope(n) == "simulator");
    if has_simulator {
        project_env.insert(SIMULATOR_VISIBLE_ENV_VAR.to_string(), "1".to_string());
    } else if !node_names.is_empty() {
        println!(
            "\u{2139}\u{fe0f}  --sim-visible has no effect: no `simulator` node in this dev session ({}).",
            node_names.join(", ")
        );
    }
}

/// Whenever a `simulator` node is part of this spawn batch, inject
/// `MECHA10_LOCOMOTION_MAX_LINEAR_SPEED`/`MECHA10_LOCOMOTION_MAX_ANGULAR_SPEED` into
/// `project_env` so the simulator subprocess's `serve` command (mecha10-rl, LAB-1976) always has
/// its physics `ctrlrange`/`forcerange` ceiling explicitly set - `serve` now (LAB-1978) fails fast
/// at startup if either var is missing, since `model.json` no longer carries its own
/// `locomotion.max_linear_speed`/`max_angular_speed` fallback (LAB-1985).
///
/// Value priority:
/// 1. If a `teleop` node is *also* present in this batch and
///    `ops::teleop::resolve_teleop_velocity_limits` resolves a *real* project teleop config (not
///    its own hardcoded-defaults fallback), use teleop's configured speed limits as the physics
///    ceiling - unchanged LAB-1977 behavior, so raising teleop's configured limit continues to
///    actually affect the simulator's physics instead of silently disagreeing with a separate
///    `model.json` value.
/// 2. Otherwise (no `teleop` node in this batch, or its config resolution fell back to defaults),
///    use this module's own compiled-in [`DEFAULT_LOCOMOTION_MAX_LINEAR_SPEED`]/
///    [`DEFAULT_LOCOMOTION_MAX_ANGULAR_SPEED`] - matching `model.json`'s former bundled defaults,
///    so a bare `mecha10 dev` session with just the simulator node (no teleop) keeps working
///    exactly as before, just sourced from Rust instead of `model.json`.
///
/// The only case that does *not* inject the speed vars at all is no `simulator` node at all in
/// this batch - mirrors `maybe_inject_sim_visible_env`'s `has_simulator` check; there's no
/// simulator subprocess to pass these vars to otherwise.
///
/// Acceleration (`MECHA10_LOCOMOTION_ACCELERATION`) follows different, looser rules than the two
/// speed vars above: it's *optionally* injected only when a `teleop` node is present in this
/// batch AND its resolved config carries an explicit `acceleration: Some(_)` override. In every
/// other case - no `teleop` node, a `teleop` node whose config resolution fell back to defaults,
/// or a resolved `teleop` config with `acceleration: None` - nothing is inserted, leaving the var
/// absent so the simulator subprocess's `serve`/`generate_robot_mjcf` falls back to `model.json`'s
/// own `locomotion.acceleration`. There is no Rust-side compiled-in fallback for acceleration the
/// way there is for linear/angular speed - `model.json` is a fully valid source of truth for it on
/// its own.
///
/// Callable as `pub(crate)` (rather than only from `spawn_nodes_by_name` in this module) so
/// `handlers::run::run_robot_target` can wire the same injection into `mecha10 run robot`, which
/// execs the built project binary directly instead of going through this module's node-runner /
/// direct-spawn paths.
pub(crate) fn maybe_inject_teleop_locomotion_env(node_names: &[String], project_env: &mut HashMap<String, String>) {
    inject_teleop_locomotion_env(node_names, super::teleop::resolve_teleop_velocity_limits(), project_env)
}

/// Pure decision logic behind [`maybe_inject_teleop_locomotion_env`], split out so it's unit
/// testable without depending on a real `configs/nodes/@mecha10/teleop/config.json` on disk -
/// same rationale as [`inject_sim_visible_env`] taking its own trigger as a parameter instead of
/// reading `std::env::var` directly.
fn inject_teleop_locomotion_env(
    node_names: &[String],
    resolved_teleop_limits: Option<super::teleop::TeleopPhysicsLimits>,
    project_env: &mut HashMap<String, String>,
) {
    let has_simulator = node_names.iter().any(|n| super::strip_node_scope(n) == "simulator");
    if !has_simulator {
        return;
    }

    let has_teleop = node_names.iter().any(|n| super::strip_node_scope(n) == "teleop");

    let (max_linear, max_angular, acceleration) = match (has_teleop, resolved_teleop_limits) {
        (true, Some(limits)) => (limits.max_linear_vel, limits.max_angular_vel, limits.acceleration),
        _ => (
            DEFAULT_LOCOMOTION_MAX_LINEAR_SPEED,
            DEFAULT_LOCOMOTION_MAX_ANGULAR_SPEED,
            None,
        ),
    };

    project_env.insert(LOCOMOTION_MAX_LINEAR_SPEED_ENV_VAR.to_string(), max_linear.to_string());
    project_env.insert(
        LOCOMOTION_MAX_ANGULAR_SPEED_ENV_VAR.to_string(),
        max_angular.to_string(),
    );

    // Optional override - only inserted when a `teleop` node resolved an explicit acceleration
    // value. Absent in every other case so `model.json`'s own value remains the fallback.
    if let Some(acceleration) = acceleration {
        project_env.insert(LOCOMOTION_ACCELERATION_ENV_VAR.to_string(), acceleration.to_string());
    }
}

/// Outcome of attempting to spawn a batch of nodes by name.
///
/// Spawning a node is a best-effort, per-node operation: one node failing to
/// start (e.g. a bad version pin, a download error) should never prevent the
/// other nodes in the batch from starting. Callers need both halves of the
/// result — which nodes actually came up, and which didn't (and why) — so
/// they can decide whether the overall operation (e.g. a lifecycle mode
/// change) should be reported as fully successful.
#[derive(Debug, Default)]
pub struct SpawnOutcome {
    /// Node name -> PID for nodes that started successfully.
    pub pids: HashMap<String, u32>,
    /// (node name, error message) for nodes that failed to spawn.
    pub failures: Vec<(String, String)>,
}

/// Spawn nodes by name
///
/// This is used when lifecycle management is enabled and we have node names
/// instead of NodeToRun objects.
///
/// Uses mecha10-node-runner if available (framework dev mode), otherwise
/// spawns nodes directly via the project binary (standalone mode).
///
/// A per-node spawn failure does not abort the batch — the failing node is
/// simply omitted from `SpawnOutcome::pids` and recorded in
/// `SpawnOutcome::failures` so the caller can decide how to report it. This
/// function itself only returns `Err` for failures that aren't specific to a
/// single node (e.g. failing to build the shared environment).
///
/// # Arguments
///
/// * `ctx` - CLI context
/// * `node_names` - Names of nodes to spawn
/// * `project_name` - Project name (used for direct spawning in standalone mode)
/// * `project_config` - Project configuration (for environment variables)
/// * `log_buffer` - Optional log buffer for logging node start events
pub fn spawn_nodes_by_name(
    ctx: &mut CliContext,
    node_names: &[String],
    project_name: &str,
    project_config: &ProjectConfig,
    log_buffer: Option<&LogBuffer>,
) -> Result<SpawnOutcome> {
    use crate::services::ProcessService;

    println!("Starting nodes...");
    println!();

    // Use shared process service so CommandListener can control these nodes
    let shared_process = ctx.shared_process();
    let mut process_service = shared_process.lock().unwrap();

    // Debug: Log to file to verify we're using shared service
    let _ = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open("/tmp/command-listener.log")
        .and_then(|mut f| {
            use std::io::Write;
            writeln!(
                f,
                "[spawn_nodes_by_name] Using shared ProcessService, ptr={:p}",
                &*process_service
            )
        });

    // Check if node-runner is available (framework dev mode)
    let use_node_runner = ProcessService::is_node_runner_available();

    // Ensure models are downloaded before spawning nodes (both modes)
    // Use block_in_place to allow blocking within the async runtime
    if let Err(e) = tokio::task::block_in_place(|| {
        tokio::runtime::Handle::current().block_on(ensure_node_models_available(node_names))
    }) {
        eprintln!("⚠️  Warning: Failed to check/download models: {}", e);
    }

    if !use_node_runner {
        println!("  (standalone mode - spawning nodes directly)");
        println!();
    }

    // Build project environment variables for node config substitution
    // These are needed by both node-runner and direct spawn for ${VAR} substitution
    let project_env = {
        // API key from global credentials (user must be logged in)
        let credentials_service = CredentialsService::new();
        let api_key = match credentials_service.get_api_key() {
            Ok(Some(api_key)) => {
                tracing::debug!("🔑 Loaded API key from credentials: {}", &api_key[..20]);
                Some(api_key)
            }
            _ => {
                tracing::warn!("⚠️  No API key found in credentials - did you run 'mecha10 auth login'?");
                None
            }
        };

        // Control plane/relay/redis/robot-id/api-key vars - shared with the launcher
        // service's `start_robot_binary` via `mecha10_core::config::robot_process_env`
        // (see its docs for why) so the two callers derive these the same way.
        let mut env: HashMap<String, String> = mecha10_core::config::robot_process_env(
            &project_config.environments.control_plane_url(),
            &project_config.environments.relay_url(),
            &project_config.environments.redis_url(),
            &project_config.robot.id,
            api_key.as_deref(),
        )
        .into_iter()
        .map(|(k, v)| (k.to_string(), v))
        .collect();

        // Thread the resolved CLI log level down to spawned node processes so `mecha10
        // --log-level debug dev` actually changes the runtime logging of the nodes it starts,
        // not just the CLI's own logs (see `packages/runtime/src/logging.rs::init_logging`,
        // which now reads `LOG_LEVEL` first).
        env.insert("LOG_LEVEL".to_string(), ctx.log_level.to_string().to_lowercase());

        // TURN credentials auto-fetched from mecha10-api (LAB-1915) - see
        // `maybe_inject_turn_credentials` docs above for the full skip/fallback rules. Uses
        // `block_in_place` + `Handle::current().block_on` the same way `ensure_node_models_available`
        // does above, to run this async fetch from this otherwise-synchronous function.
        {
            let control_plane_url = project_config.environments.control_plane_url();
            let api_key_for_turn = env.get("ROBOT_API_KEY").cloned();
            tokio::task::block_in_place(|| {
                tokio::runtime::Handle::current().block_on(maybe_inject_turn_credentials(
                    node_names,
                    &control_plane_url,
                    api_key_for_turn.as_deref(),
                    &mut env,
                ))
            });
        }

        // `mecha10 dev --sim-visible` -> `simulator` node's subprocess `--visible` flag (see
        // module docs above and LAB-1886). No-op (with a printed message) when there's no
        // `simulator` node in this batch.
        maybe_inject_sim_visible_env(node_names, &mut env);

        // Locomotion speed limits -> simulator subprocess's ctrlrange/forcerange override
        // (LAB-1977, LAB-1985). Always injected when a `simulator` node is in this batch - using
        // teleop's configured values if a `teleop` node is also present and resolves a real
        // config, otherwise this module's own compiled-in defaults - since `serve` (LAB-1978) now
        // fails fast at startup without these vars set. No-op only when there's no `simulator`
        // node at all - see `maybe_inject_teleop_locomotion_env` docs.
        maybe_inject_teleop_locomotion_env(node_names, &mut env);

        // Debug: Log all environment variables being injected
        tracing::debug!("📋 Environment variables to inject:");
        tracing::debug!(
            "  CONTROL_PLANE_URL: {}",
            env.get("CONTROL_PLANE_URL").unwrap_or(&"<missing>".to_string())
        );
        tracing::debug!(
            "  ROBOT_ID: {}",
            env.get("ROBOT_ID").unwrap_or(&"<missing>".to_string())
        );
        tracing::debug!(
            "  ROBOT_API_KEY: {}",
            if env.contains_key("ROBOT_API_KEY") {
                "<present>"
            } else {
                "<missing>"
            }
        );
        tracing::debug!(
            "  WEBRTC_RELAY_URL: {}",
            env.get("WEBRTC_RELAY_URL").unwrap_or(&"<missing>".to_string())
        );
        tracing::debug!(
            "  REDIS_URL: {}",
            env.get("REDIS_URL").unwrap_or(&"<missing>".to_string())
        );

        Some(env)
    };

    // Spawn each node
    let mut pids = HashMap::new();
    let mut failures = Vec::new();
    for node_name in node_names {
        // mecha10-node-runner supervises every node uniformly - local/custom project nodes
        // (plain, unscoped identifiers) and bundled framework nodes (@mecha10/*) alike - matching
        // `launcher-service::create_node_runner`'s production behavior (see module docs on
        // `ProcessService::spawn_node_runner` for why gating this by node category used to
        // silently break bundled nodes: it never fixed anything, it just hid the real bug).
        let should_use_node_runner = use_node_runner;

        tracing::debug!(
            "🚀 Spawning node: {} (use_node_runner: {})",
            node_name,
            should_use_node_runner
        );
        let result = if should_use_node_runner {
            process_service.spawn_node_runner(node_name, project_name, project_env.clone())
        } else {
            process_service.spawn_node_direct(node_name, project_name, project_env.clone())
        };

        match result {
            Ok(pid) => {
                pids.insert(node_name.clone(), pid);
                // Debug: Log spawned node
                let _ = std::fs::OpenOptions::new()
                    .create(true)
                    .append(true)
                    .open("/tmp/command-listener.log")
                    .and_then(|mut f| {
                        use std::io::Write;
                        writeln!(f, "[spawn_nodes_by_name] Spawned '{}' with PID {}", node_name, pid)
                    });
                if let Some(buffer) = log_buffer {
                    buffer.push(format!("✅ STARTED: {} (PID: {})", node_name, pid));
                }
            }
            Err(e) => {
                eprintln!("Failed to spawn {}: {}", node_name, e);
                if let Some(buffer) = log_buffer {
                    buffer.push(format!("ERROR: Failed to start {}: {}", node_name, e));
                }
                failures.push((node_name.clone(), e.to_string()));
            }
        }
    }

    for (node, pid) in &pids {
        println!("  ✅ {} (PID: {})", node, pid);
    }
    println!();

    Ok(SpawnOutcome { pids, failures })
}

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

    // ---- inject_sim_visible_env ------------------------------------------------------------

    #[test]
    fn injects_visible_env_var_when_sim_visible_and_simulator_present() {
        let mut env = HashMap::new();
        inject_sim_visible_env(true, &["simulator".to_string(), "teleop".to_string()], &mut env);
        assert_eq!(env.get(SIMULATOR_VISIBLE_ENV_VAR), Some(&"1".to_string()));
    }

    #[test]
    fn injects_visible_env_var_for_scoped_simulator_node_name() {
        let mut env = HashMap::new();
        inject_sim_visible_env(true, &["@mecha10/simulator".to_string()], &mut env);
        assert_eq!(env.get(SIMULATOR_VISIBLE_ENV_VAR), Some(&"1".to_string()));
    }

    #[test]
    fn does_not_inject_when_sim_visible_not_requested() {
        let mut env = HashMap::new();
        inject_sim_visible_env(false, &["simulator".to_string()], &mut env);
        assert!(!env.contains_key(SIMULATOR_VISIBLE_ENV_VAR));
    }

    #[test]
    fn does_not_inject_when_no_simulator_node_in_batch() {
        let mut env = HashMap::new();
        inject_sim_visible_env(true, &["teleop".to_string(), "camera".to_string()], &mut env);
        assert!(!env.contains_key(SIMULATOR_VISIBLE_ENV_VAR));
    }

    #[test]
    fn does_not_inject_when_node_list_is_empty() {
        let mut env = HashMap::new();
        inject_sim_visible_env(true, &[], &mut env);
        assert!(!env.contains_key(SIMULATOR_VISIBLE_ENV_VAR));
    }

    // ---- inject_teleop_locomotion_env --------------------------------------------------------

    fn stub_teleop_limits() -> super::super::teleop::TeleopPhysicsLimits {
        super::super::teleop::TeleopPhysicsLimits {
            max_linear_vel: 10.0,
            max_angular_vel: 2.5,
            acceleration: None,
        }
    }

    fn stub_teleop_limits_with_acceleration(acceleration: f32) -> super::super::teleop::TeleopPhysicsLimits {
        super::super::teleop::TeleopPhysicsLimits {
            acceleration: Some(acceleration),
            ..stub_teleop_limits()
        }
    }

    #[test]
    fn injects_both_env_vars_when_teleop_config_resolves_and_both_nodes_present() {
        let mut env = HashMap::new();
        inject_teleop_locomotion_env(
            &["simulator".to_string(), "teleop".to_string()],
            Some(stub_teleop_limits()),
            &mut env,
        );
        assert_eq!(env.get(LOCOMOTION_MAX_LINEAR_SPEED_ENV_VAR), Some(&"10".to_string()));
        assert_eq!(env.get(LOCOMOTION_MAX_ANGULAR_SPEED_ENV_VAR), Some(&"2.5".to_string()));
    }

    #[test]
    fn injects_for_scoped_simulator_and_teleop_node_names() {
        let mut env = HashMap::new();
        inject_teleop_locomotion_env(
            &["@mecha10/simulator".to_string(), "@mecha10/teleop".to_string()],
            Some(stub_teleop_limits()),
            &mut env,
        );
        assert_eq!(env.get(LOCOMOTION_MAX_LINEAR_SPEED_ENV_VAR), Some(&"10".to_string()));
        assert_eq!(env.get(LOCOMOTION_MAX_ANGULAR_SPEED_ENV_VAR), Some(&"2.5".to_string()));
    }

    #[test]
    fn injects_rust_side_defaults_when_teleop_config_resolution_falls_back_to_defaults() {
        let mut env = HashMap::new();
        inject_teleop_locomotion_env(&["simulator".to_string(), "teleop".to_string()], None, &mut env);
        assert_eq!(
            env.get(LOCOMOTION_MAX_LINEAR_SPEED_ENV_VAR),
            Some(&DEFAULT_LOCOMOTION_MAX_LINEAR_SPEED.to_string())
        );
        assert_eq!(
            env.get(LOCOMOTION_MAX_ANGULAR_SPEED_ENV_VAR),
            Some(&DEFAULT_LOCOMOTION_MAX_ANGULAR_SPEED.to_string())
        );
    }

    #[test]
    fn does_not_inject_when_no_simulator_node_present() {
        let mut env = HashMap::new();
        inject_teleop_locomotion_env(&["teleop".to_string()], Some(stub_teleop_limits()), &mut env);
        assert!(!env.contains_key(LOCOMOTION_MAX_LINEAR_SPEED_ENV_VAR));
        assert!(!env.contains_key(LOCOMOTION_MAX_ANGULAR_SPEED_ENV_VAR));
    }

    #[test]
    fn injects_rust_side_defaults_when_no_teleop_node_present() {
        let mut env = HashMap::new();
        inject_teleop_locomotion_env(&["simulator".to_string()], Some(stub_teleop_limits()), &mut env);
        // Teleop's resolved limits are ignored here since no `teleop` node is actually part of
        // this batch (e.g. a stale/irrelevant resolution) - the Rust-side compiled-in defaults
        // are used instead, not teleop's values.
        assert_eq!(
            env.get(LOCOMOTION_MAX_LINEAR_SPEED_ENV_VAR),
            Some(&DEFAULT_LOCOMOTION_MAX_LINEAR_SPEED.to_string())
        );
        assert_eq!(
            env.get(LOCOMOTION_MAX_ANGULAR_SPEED_ENV_VAR),
            Some(&DEFAULT_LOCOMOTION_MAX_ANGULAR_SPEED.to_string())
        );
    }

    #[test]
    fn injects_acceleration_when_teleop_present_with_explicit_override() {
        let mut env = HashMap::new();
        inject_teleop_locomotion_env(
            &["simulator".to_string(), "teleop".to_string()],
            Some(stub_teleop_limits_with_acceleration(3.5)),
            &mut env,
        );
        assert_eq!(env.get(LOCOMOTION_ACCELERATION_ENV_VAR), Some(&"3.5".to_string()));
    }

    #[test]
    fn does_not_inject_acceleration_when_teleop_present_without_explicit_override() {
        let mut env = HashMap::new();
        inject_teleop_locomotion_env(
            &["simulator".to_string(), "teleop".to_string()],
            Some(stub_teleop_limits()),
            &mut env,
        );
        assert!(!env.contains_key(LOCOMOTION_ACCELERATION_ENV_VAR));
    }

    #[test]
    fn does_not_inject_acceleration_when_no_teleop_node_present() {
        let mut env = HashMap::new();
        inject_teleop_locomotion_env(
            &["simulator".to_string()],
            Some(stub_teleop_limits_with_acceleration(3.5)),
            &mut env,
        );
        // Teleop's resolved acceleration is ignored here since no `teleop` node is actually part
        // of this batch, same rationale as the linear/angular speed case above.
        assert!(!env.contains_key(LOCOMOTION_ACCELERATION_ENV_VAR));
    }

    #[test]
    fn does_not_inject_acceleration_when_teleop_config_resolution_falls_back_to_defaults() {
        let mut env = HashMap::new();
        inject_teleop_locomotion_env(&["simulator".to_string(), "teleop".to_string()], None, &mut env);
        assert!(!env.contains_key(LOCOMOTION_ACCELERATION_ENV_VAR));
    }

    #[test]
    fn injects_rust_side_defaults_for_scoped_simulator_node_name_without_teleop() {
        let mut env = HashMap::new();
        inject_teleop_locomotion_env(&["@mecha10/simulator".to_string()], None, &mut env);
        assert_eq!(
            env.get(LOCOMOTION_MAX_LINEAR_SPEED_ENV_VAR),
            Some(&DEFAULT_LOCOMOTION_MAX_LINEAR_SPEED.to_string())
        );
        assert_eq!(
            env.get(LOCOMOTION_MAX_ANGULAR_SPEED_ENV_VAR),
            Some(&DEFAULT_LOCOMOTION_MAX_ANGULAR_SPEED.to_string())
        );
    }

    // `record_spawn_outcome`'s old behavior (merging `SpawnOutcome` into a persistent
    // "last known failure per node" map) now lives on `NodeRegistry` - see
    // `super::registry::tests` for the ported coverage (LAB-1891).

    // ---- batch_needs_turn_credentials ------------------------------------------------------

    #[test]
    fn needs_turn_credentials_when_camera_node_present() {
        assert!(batch_needs_turn_credentials(&[
            "teleop".to_string(),
            "camera".to_string()
        ]));
    }

    #[test]
    fn needs_turn_credentials_for_scoped_video_streamer_node_name() {
        assert!(batch_needs_turn_credentials(&["@mecha10/video-streamer".to_string()]));
    }

    #[test]
    fn does_not_need_turn_credentials_when_no_video_relevant_node() {
        assert!(!batch_needs_turn_credentials(&[
            "teleop".to_string(),
            "motor".to_string()
        ]));
    }

    #[test]
    fn does_not_need_turn_credentials_when_batch_is_empty() {
        assert!(!batch_needs_turn_credentials(&[]));
    }

    // ---- maybe_inject_turn_credentials -----------------------------------------------------
    //
    // Uses a hand-rolled single-request TCP listener rather than pulling in an HTTP mocking
    // crate - this crate has no such dependency yet and the request/response shape here is
    // trivial enough not to justify adding one.

    /// Spawns a background thread that accepts exactly one TCP connection on an OS-assigned
    /// local port, ignores whatever request it receives, and writes back an HTTP response with
    /// the given status line and body (with a correctly computed `Content-Length`). Returns the
    /// `http://127.0.0.1:<port>` base URL to point the fetch at.
    fn spawn_one_shot_http_server(status_line: &'static str, body: &'static str) -> String {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
        let addr = listener.local_addr().expect("local_addr");
        std::thread::spawn(move || {
            use std::io::{Read, Write};
            if let Ok((mut stream, _)) = listener.accept() {
                let mut buf = [0u8; 1024];
                let _ = stream.read(&mut buf);
                let response = format!(
                    "{status_line}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
                    body.len()
                );
                let _ = stream.write_all(response.as_bytes());
                let _ = stream.flush();
            }
        });
        format!("http://{}", addr)
    }

    /// Binds an ephemeral port and immediately drops the listener, so a connection attempt to
    /// the returned base URL reliably fails (connection refused) - used to simulate mecha10-api
    /// being unreachable.
    fn unreachable_base_url() -> String {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
        let addr = listener.local_addr().expect("local_addr");
        drop(listener);
        format!("http://{}", addr)
    }

    #[tokio::test]
    async fn successful_fetch_populates_project_env() {
        let base_url = spawn_one_shot_http_server(
            "HTTP/1.1 200 OK",
            "{\"url\":\"turn:turn.example.com:3478\",\"username\":\"u\",\"credential\":\"c\"}",
        );
        let mut env = HashMap::new();

        maybe_inject_turn_credentials(&["camera".to_string()], &base_url, Some("test-api-key"), &mut env).await;

        assert_eq!(
            env.get(TURN_URL_ENV_VAR),
            Some(&"turn:turn.example.com:3478".to_string())
        );
        assert_eq!(env.get(TURN_USERNAME_ENV_VAR), Some(&"u".to_string()));
        assert_eq!(env.get(TURN_CREDENTIAL_ENV_VAR), Some(&"c".to_string()));
    }

    #[tokio::test]
    async fn not_configured_404_leaves_project_env_unchanged() {
        let base_url = spawn_one_shot_http_server("HTTP/1.1 404 Not Found", "");
        let mut env = HashMap::new();

        maybe_inject_turn_credentials(&["camera".to_string()], &base_url, Some("test-api-key"), &mut env).await;

        assert!(env.is_empty());
    }

    #[tokio::test]
    async fn network_failure_leaves_project_env_unchanged() {
        let base_url = unreachable_base_url();
        let mut env = HashMap::new();

        maybe_inject_turn_credentials(&["camera".to_string()], &base_url, Some("test-api-key"), &mut env).await;

        assert!(env.is_empty());
    }

    #[tokio::test]
    async fn skips_fetch_when_batch_has_no_video_relevant_node() {
        // No server bound at this base URL at all - if the fetch fired anyway (gating broken)
        // this would fail as a connection error rather than silently no-op.
        let base_url = unreachable_base_url();
        let mut env = HashMap::new();

        maybe_inject_turn_credentials(&["teleop".to_string()], &base_url, Some("test-api-key"), &mut env).await;

        assert!(env.is_empty());
    }

    #[tokio::test]
    #[serial_test::serial(turn_url_env_var)]
    async fn explicit_override_skips_fetch_and_project_env_is_preserved_unchanged() {
        std::env::set_var(TURN_URL_ENV_VAR, "turn:operator-set.example.com:3478");

        // No server bound at this base URL at all - if the fetch fired anyway (override not
        // respected) this would fail as a connection error rather than silently no-op.
        let base_url = unreachable_base_url();
        let mut env = HashMap::new();

        maybe_inject_turn_credentials(&["camera".to_string()], &base_url, Some("test-api-key"), &mut env).await;

        std::env::remove_var(TURN_URL_ENV_VAR);

        // project_env itself is untouched - the operator's own shell export already flows
        // through to the spawned subprocess via inherited process env (see
        // `ProcessService::spawn_with_env`), so nothing needs to be copied in here.
        assert!(env.is_empty());
    }

    // ---- three-tier resolution priority (LAB-1917) -----------------------------------------

    fn stub_baked_config() -> TurnConfig {
        TurnConfig {
            url: "turn:baked.example.com:3478".to_string(),
            username: "baked-user".to_string(),
            credential: "baked-secret".to_string(),
        }
    }

    #[tokio::test]
    async fn tier2_baked_default_populates_project_env_when_no_explicit_override() {
        // No server bound at this base URL at all - if tier 3's fetch fired anyway (tier 2 not
        // taking priority) this would fail as a connection error rather than silently no-op.
        let base_url = unreachable_base_url();
        let mut env = HashMap::new();

        maybe_inject_turn_credentials_with_baked_resolver(
            &["camera".to_string()],
            &base_url,
            Some("test-api-key"),
            &mut env,
            || Some(stub_baked_config()),
        )
        .await;

        assert_eq!(
            env.get(TURN_URL_ENV_VAR),
            Some(&"turn:baked.example.com:3478".to_string())
        );
        assert_eq!(env.get(TURN_USERNAME_ENV_VAR), Some(&"baked-user".to_string()));
        assert_eq!(env.get(TURN_CREDENTIAL_ENV_VAR), Some(&"baked-secret".to_string()));
    }

    #[tokio::test]
    #[serial_test::serial(turn_url_env_var)]
    async fn tier1_explicit_env_var_wins_over_tier2_baked_default() {
        std::env::set_var(TURN_URL_ENV_VAR, "turn:operator-set.example.com:3478");

        // No server bound at this base URL at all - if tier 3's fetch fired anyway this would
        // fail as a connection error rather than silently no-op.
        let base_url = unreachable_base_url();
        let mut env = HashMap::new();

        maybe_inject_turn_credentials_with_baked_resolver(
            &["camera".to_string()],
            &base_url,
            Some("test-api-key"),
            &mut env,
            || Some(stub_baked_config()),
        )
        .await;

        std::env::remove_var(TURN_URL_ENV_VAR);

        // project_env itself is untouched: an explicit override already flows through to the
        // spawned subprocess via inherited process env, same as the no-baked-default case above,
        // and the (stubbed) baked default correctly never gets a chance to run.
        assert!(env.is_empty());
    }

    #[tokio::test]
    async fn tier3_fetch_still_runs_when_no_explicit_override_and_no_baked_default() {
        let base_url = spawn_one_shot_http_server(
            "HTTP/1.1 200 OK",
            "{\"url\":\"turn:fetched.example.com:3478\",\"username\":\"fetched-user\",\"credential\":\"fetched-secret\"}",
        );
        let mut env = HashMap::new();

        maybe_inject_turn_credentials_with_baked_resolver(
            &["camera".to_string()],
            &base_url,
            Some("test-api-key"),
            &mut env,
            || None,
        )
        .await;

        assert_eq!(
            env.get(TURN_URL_ENV_VAR),
            Some(&"turn:fetched.example.com:3478".to_string())
        );
        assert_eq!(env.get(TURN_USERNAME_ENV_VAR), Some(&"fetched-user".to_string()));
        assert_eq!(env.get(TURN_CREDENTIAL_ENV_VAR), Some(&"fetched-secret".to_string()));
    }
}