feat: pre-serve hooks + exec subcommand for cluster mass-execution

Two new features for mass-running commands on a cluster:

1. `before_serve` hooks (config.toml + recipe.toml)

   Configurable commands run on every node BEFORE `serve` / `tmux serve`
   launches containers or sessions.  Cluster-level config:

   ```toml
   [before_serve."4nodes-switch"]
   commands = [
       # default: abort the serve if this fails
       "sudo nvidia-smi -lgc 300,2300",
       # explicit on_failure: abort|warn|continue
       { command = "sudo nvidia-smi -pm 1", on_failure = "warn" },
   ]
   ```

   Recipe-level `before_serve = [...]` appends after the cluster config,
   so cluster setup runs first, then recipe-specific setup.

   Per-command exit-code handling via `on_failure` field:
   - `abort`    (default) — fail-fast, abort the serve
   - `warn`     — log a warning, continue
   - `continue` — stay silent, continue

   Use `--skip-before-serve` (or `cargo run tmux serve ... --skip-before-serve`)
   to bypass for a one-off run.

2. `exec` subcommand

   Mass-execute arbitrary commands.  Use for `sudo shutdown now`,
   `sudo nvidia-smi -lgc ...`, or any other ad-hoc operation:

   ```
   cargo run exec --cluster 4nodes-switch -- "sudo shutdown now"
   cargo run exec --cluster 4nodes-switch --node dgx1 -- "uptime"
   cargo run exec --recipe glm51 -- "cat /tmp/foo"
   ```

   Defaults to running on every node of the cluster; `--node` targets
   a specific node; `--recipe` resolves to the recipe's pinned cluster.

Implementation notes:
- New module `before_serve.rs` orchestrates the hook: combines cluster +
  recipe commands into one ordered list, runs them per-node sequentially
  and across-nodes in parallel.  Failures bubble up as `anyhow::Error`
  with the originating command and node name.
- New module `exec.rs` is a thin wrapper over `run_ssh` that targets a
  cluster or single node and streams output per node.
- `run_ssh` now tracks the remote exit status and returns Err with the
  captured output as context for non-zero codes.  Previously a `false`
  command was indistinguishable from a successful one.
- Tests: 6 new (parsing cluster-level + recipe-level config, ordering,
  default-on-failure); `make check` passes 43/43.

AGENTS.md

@@ -18,11 +18,16 @@ Rust CLI tool for launching vLLM/SGLang containers across a DGX Spark cluster. S
 ```
 cargo run serve --recipe glm51                             # Auto-cluster selection
 cargo run serve --recipe glm51 --cluster 4nodes-switch     # Explicit cluster
+cargo run serve --recipe glm51 --skip-before-serve         # Bypass pre-serve hooks
 cargo run -- -d serve --recipe glm51                       # Dry-run (print only)
 cargo run stop --recipe glm51                              # Stop containers on all nodes
 cargo run health                                           # Health check all clusters
 cargo run health --cluster 2nodes-dac                      # Single cluster
 cargo run health --fingerprint                             # Show host keys for config.toml
+cargo run init-host-keys                                   # Auto-write SSH host_keys
+cargo run exec --cluster 4nodes -- "sudo nvidia-smi -lgc 300,2300"   # Run on all nodes
+cargo run exec --cluster 4nodes --node dgx1 -- "uptime"            # Run on one node
+cargo run exec --recipe glm51 -- "cat /tmp/foo"                     # Use recipe's cluster
 cargo run logs --recipe glm51                              # Last 100 lines per node
 cargo run logs --recipe glm51 --follow                     # Follow head node logs
 cargo run doctor --recipe glm51                            # Check node prerequisites
@@ -80,6 +85,38 @@ A recipe can pin itself to a specific cluster (bypasses `--cluster` / auto-pick)
 cluster = "4nodes-switch"
 ```
 
+### Pre-serve hooks (`[before_serve."<cluster>"]`)
+
+Run shell commands on every node of a cluster **before** `serve` / `tmux serve`
+launches containers or tmux sessions.  Defined in `config.toml`:
+
+```toml
+[before_serve."4nodes-switch"]
+commands = [
+    # Default: abort the serve if this fails
+    "sudo nvidia-smi -lgc 300,2300",
+
+    # Explicit on_failure:
+    #   "abort"    — fail-fast (default)
+    #   "warn"     — log warning, continue
+    #   "continue" — stay silent, continue
+    { command = "sudo nvidia-smi -pm 1", on_failure = "warn" },
+]
+```
+
+Recipes can also declare pre-serve commands that **append** to the cluster
+config (so recipe-level commands run after cluster-level):
+
+```toml
+before_serve = [
+    { command = "echo recipe-level", on_failure = "abort" },
+]
+```
+
+Commands are run sequentially on each node (so order matters) but nodes
+run in parallel.  Use `--skip-before-serve` to bypass for a one-off
+serve.
+
 ## recipes/
 
 TOML format. Directory name is intentionally misspelled. Each recipe lives in its own subdirectory:

src/before_serve.rs

@@ -0,0 +1,286 @@
+use anyhow::Result;
+use tracing::{info, warn};
+
+use crate::config::{Cluster, Config, OnFailure, PreServeCommand, Recipe};
+
+/// Run pre-serve commands on every node of the cluster before launching
+/// containers or tmux sessions.  Combines cluster-level config
+/// (`[before_serve."<cluster>"]`) with recipe-level overrides
+/// (`before_serve = [...]` in recipe.toml).  Cluster-level commands run
+/// first, then recipe-level — each in declared order — so the resulting
+/// state is predictable.
+///
+/// Per-command behaviour on a non-zero exit is controlled by
+/// [`PreServeCommand::on_failure`]:
+/// - `abort`   (default) — fail-fast: abort the whole serve
+/// - `warn`    — log warning, continue with remaining commands
+/// - `continue`— stay silent, continue
+///
+/// If `skip` is true the function returns without running anything.
+pub async fn run(
+    config: &Config,
+    cluster: &Cluster,
+    recipe: &Recipe,
+    skip: bool,
+    dry_run: bool,
+) -> Result<()> {
+    if skip {
+        info!("⏭  Skipping pre-serve hooks (--skip-before-serve)");
+        return Ok(());
+    }
+
+    let cmds = combined_commands(config, cluster, recipe);
+    if cmds.is_empty() {
+        return Ok(());
+    }
+
+    info!(
+        "🔧 Pre-serve hooks: {} command(s) on {} node(s)",
+        cmds.len(),
+        cluster.nodes.len()
+    );
+
+    // Run all commands for one node sequentially (so order matters per
+    // node), and the nodes in parallel.
+    let mut handles = Vec::with_capacity(cluster.nodes.len());
+    for node in &cluster.nodes {
+        let config = config.clone();
+        let node = node.clone();
+        let cmds = cmds.clone();
+        let dry = dry_run;
+        handles.push(tokio::spawn(async move {
+            (
+                node.name.clone(),
+                run_for_node(&config, &node, &cmds, dry).await,
+            )
+        }));
+    }
+
+    let mut any_failed = false;
+    for h in handles {
+        let (name, result) = h.await?;
+        if let Err(e) = result {
+            any_failed = true;
+            warn!("pre-serve: node {} failed: {}", name, e);
+        }
+    }
+
+    if any_failed {
+        anyhow::bail!("Pre-serve hooks failed on at least one node (see warnings above)");
+    }
+    Ok(())
+}
+
+/// Combine cluster-level + recipe-level commands into one ordered list.
+fn combined_commands(
+    config: &Config,
+    cluster: &Cluster,
+    recipe: &Recipe,
+) -> Vec<(&'static str, PreServeCommand)> {
+    let mut out: Vec<(&'static str, PreServeCommand)> = Vec::new();
+    if let Some(table) = config.before_serve.get(&cluster.name) {
+        for c in &table.commands {
+            out.push(("cluster", c.clone()));
+        }
+    }
+    for c in &recipe.before_serve {
+        out.push(("recipe", c.clone()));
+    }
+    out
+}
+
+/// Run a fixed list of commands on a single node, sequentially.
+/// Returns the first command that failed under `on_failure = abort`.
+async fn run_for_node(
+    config: &Config,
+    node: &crate::config::Node,
+    cmds: &[(&'static str, PreServeCommand)],
+    dry_run: bool,
+) -> Result<()> {
+    for (origin, cmd) in cmds {
+        let command = cmd.command();
+        let on_failure = cmd.on_failure();
+        if dry_run {
+            info!("  {} ({}) :: {}", node.name, origin, truncate(command, 120));
+            continue;
+        }
+        let result = crate::ssh::run_ssh(config, node, command.to_string(), dry_run).await;
+        let status = match &result {
+            Ok(_) => "ok",
+            Err(_) => "fail",
+        };
+        let label = format!("[{}|{}|{}]", node.name, origin, status);
+        info!("  {} $ {}", label, command);
+        if let Err(ref e) = result {
+            match on_failure {
+                OnFailure::Abort => {
+                    return Err(anyhow::anyhow!(
+                        "pre-serve command failed on {}: {}",
+                        node.name,
+                        e
+                    ));
+                }
+                OnFailure::Warn => {
+                    warn!("  ⚠  {} ignored: {}", node.name, e);
+                }
+                OnFailure::Continue => {
+                    // silent
+                }
+            }
+        }
+    }
+    Ok(())
+}
+
+fn truncate(s: &str, max: usize) -> String {
+    if s.len() <= max {
+        s.to_string()
+    } else {
+        format!("{}…", &s[..max])
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::{Node, PreServeCommand};
+    use std::collections::HashMap;
+
+    fn test_cluster() -> Cluster {
+        Cluster {
+            name: "c1".to_string(),
+            master_ib_ip: "1.2.3.4".to_string(),
+            nodes: vec![
+                Node {
+                    name: "n0".to_string(),
+                    ssh_ip: "10.0.0.1".to_string(),
+                    ib_ip: "1.2.3.4".to_string(),
+                    rank: 0,
+                    host_key: None,
+                },
+                Node {
+                    name: "n1".to_string(),
+                    ssh_ip: "10.0.0.2".to_string(),
+                    ib_ip: "1.2.3.5".to_string(),
+                    rank: 1,
+                    host_key: None,
+                },
+            ],
+        }
+    }
+
+    fn test_config(before_serve: HashMap<String, crate::config::BeforeServeTable>) -> Config {
+        Config {
+            ssh_user: "u".to_string(),
+            ssh_key_path: "/k".to_string(),
+            clusters: vec![test_cluster()],
+            nccl: HashMap::new(),
+            hf_cache_dir: None,
+            init_port: 25000,
+            before_serve,
+        }
+    }
+
+    #[test]
+    fn combined_orders_cluster_then_recipe() {
+        let mut bs = HashMap::new();
+        bs.insert(
+            "c1".to_string(),
+            crate::config::BeforeServeTable {
+                commands: vec![PreServeCommand::Bare("cluster_cmd_1".into())],
+            },
+        );
+        let cfg = test_config(bs);
+        let recipe = Recipe {
+            before_serve: vec![
+                PreServeCommand::Bare("recipe_cmd_1".into()),
+                PreServeCommand::Bare("recipe_cmd_2".into()),
+            ],
+            ..test_recipe()
+        };
+
+        let cmds = combined_commands(&cfg, &test_cluster(), &recipe);
+        let names: Vec<&str> = cmds.iter().map(|(_, c)| c.command()).collect();
+        let origins: Vec<&str> = cmds.iter().map(|(o, _)| *o).collect();
+        assert_eq!(names, vec!["cluster_cmd_1", "recipe_cmd_1", "recipe_cmd_2"]);
+        assert_eq!(origins, vec!["cluster", "recipe", "recipe"]);
+    }
+
+    #[test]
+    fn combined_empty_when_unconfigured() {
+        let cfg = test_config(HashMap::new());
+        let recipe = Recipe {
+            before_serve: vec![],
+            ..test_recipe()
+        };
+        assert!(combined_commands(&cfg, &test_cluster(), &recipe).is_empty());
+    }
+
+    fn test_recipe() -> Recipe {
+        Recipe {
+            engine: "vllm".to_string(),
+            name: "Test".to_string(),
+            docker_image: String::new(),
+            model: "x".to_string(),
+            port: 0,
+            required_nodes: None,
+            pipeline_parallel_size: None,
+            tensor_parallel_size: 1,
+            args: vec![],
+            env: HashMap::new(),
+            cluster: None,
+            entrypoint: vec![],
+            distributed_backend: "mp".to_string(),
+            before_serve: vec![],
+        }
+    }
+
+    // ── Deserialization tests (the ones that live in this module) ──
+
+    #[test]
+    fn parses_cluster_level_before_serve() {
+        let toml_str = r#"
+ssh_user = "u"
+ssh_key_path = "/k"
+
+[[clusters]]
+name = "c1"
+master_ib_ip = "1.1.1.1"
+nodes = [
+    { name = "n0", ssh_ip = "2.2.2.2", ib_ip = "1.1.1.1", rank = 0 },
+]
+
+[before_serve."c1"]
+commands = [
+    "sudo nvidia-smi -lgc 300,2300",
+    { command = "sudo nvidia-smi -pm 1", on_failure = "warn" },
+]
+"#;
+        let cfg: Config = toml::from_str(toml_str).unwrap();
+        assert!(cfg.before_serve.contains_key("c1"));
+        let list = &cfg.before_serve["c1"].commands;
+        assert_eq!(list.len(), 2);
+        assert_eq!(list[0].command(), "sudo nvidia-smi -lgc 300,2300");
+        assert_eq!(list[0].on_failure(), OnFailure::Abort); // default
+        assert_eq!(list[1].command(), "sudo nvidia-smi -pm 1");
+        assert_eq!(list[1].on_failure(), OnFailure::Warn);
+    }
+
+    #[test]
+    fn parses_recipe_level_before_serve() {
+        let toml_str = r#"
+name = "Demo"
+model = "x"
+before_serve = [
+    "echo a",
+    { command = "echo b", on_failure = "continue" },
+]
+"#;
+        let r: Recipe = toml::from_str(toml_str).unwrap();
+        assert_eq!(r.before_serve.len(), 2);
+        assert_eq!(r.before_serve[0].command(), "echo a");
+        assert_eq!(r.before_serve[0].on_failure(), OnFailure::Abort);
+        assert_eq!(r.before_serve[1].command(), "echo b");
+        assert_eq!(r.before_serve[1].on_failure(), OnFailure::Continue);
+    }
+}

src/cli.rs

@@ -26,6 +26,10 @@ pub enum Commands {
         /// Keep containers after stop (default: auto-remove on stop)
         #[arg(long, default_value = "false")]
         keep: bool,
+
+        /// Skip pre-serve hooks (cluster-level + recipe-level).
+        #[arg(long, default_value = "false")]
+        skip_before_serve: bool,
     },
 
     /// Показать логи контейнеров
@@ -86,6 +90,25 @@ pub enum Commands {
         )]
         cluster: Option<String>,
     },
+
+    /// Run an arbitrary shell command on a cluster or single node
+    Exec {
+        /// Cluster name (mutually exclusive with --recipe).
+        #[arg(long, conflicts_with = "recipe")]
+        cluster: Option<String>,
+
+        /// Recipe name; resolves to the recipe's pinned cluster.
+        #[arg(long, conflicts_with = "cluster")]
+        recipe: Option<String>,
+
+        /// Target a specific node within the cluster (default: all).
+        #[arg(long)]
+        node: Option<String>,
+
+        /// Command to run (everything after `--`).
+        #[arg(trailing_var_arg = true, required = true)]
+        command: Vec<String>,
+    },
 }
 
 #[derive(Subcommand)]
@@ -96,6 +119,9 @@ pub enum TmuxCommands {
         recipe: String,
         #[arg(long, default_value = "auto")]
         cluster: String,
+        /// Skip pre-serve hooks (cluster-level + recipe-level).
+        #[arg(long, default_value = "false")]
+        skip_before_serve: bool,
     },
     /// Остановить tmux-сессии
     Stop {

src/config.rs

@@ -28,6 +28,50 @@ pub struct Cluster {
     pub nodes: Vec<Node>,
 }
 
+/// Behaviour on a non-zero exit code for a [`PreServeCommand`].
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Default, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum OnFailure {
+    /// Abort the serve (default).
+    #[default]
+    Abort,
+    /// Log a warning but continue.
+    Warn,
+    /// Continue silently.
+    Continue,
+}
+
+/// One entry in a `before_serve` block.  Either a bare command string
+/// (defaults to `on_failure = abort`) or a table with explicit `on_failure`.
+#[derive(Clone, Debug, Deserialize)]
+#[serde(untagged)]
+pub enum PreServeCommand {
+    Bare(String),
+    Configured {
+        command: String,
+        #[serde(default)]
+        on_failure: OnFailure,
+    },
+}
+
+impl PreServeCommand {
+    pub fn command(&self) -> &str {
+        match self {
+            PreServeCommand::Bare(c) => c,
+            PreServeCommand::Configured { command, .. } => command,
+        }
+    }
+
+    pub fn on_failure(&self) -> OnFailure {
+        match self {
+            PreServeCommand::Bare(_) => OnFailure::default(),
+            PreServeCommand::Configured { on_failure, .. } => *on_failure,
+        }
+    }
+}
+
+/// Top-level config containing cluster definitions and per-cluster
+/// pre-serve commands.
 #[derive(Clone, Deserialize)]
 pub struct Config {
     pub ssh_user: String,
@@ -41,6 +85,17 @@ pub struct Config {
     /// SGLang `--dist-init-addr` port). Default: 25000.
     #[serde(default = "default_init_port")]
     pub init_port: u16,
+    /// Per-cluster pre-serve commands.  Looked up by cluster name.
+    #[serde(default)]
+    pub before_serve: HashMap<String, BeforeServeTable>,
+}
+
+/// TOML sub-table under `[before_serve."<cluster>"]`.  Currently only
+/// carries `commands`; reserved for future per-cluster tuning.
+#[derive(Clone, Debug, Default, Deserialize)]
+pub struct BeforeServeTable {
+    #[serde(default)]
+    pub commands: Vec<PreServeCommand>,
 }
 
 fn default_init_port() -> u16 {
@@ -92,6 +147,10 @@ pub struct Recipe {
     pub entrypoint: Vec<String>,
     #[serde(default = "default_distributed_backend")]
     pub distributed_backend: String,
+    /// Recipe-level pre-serve commands.  Appended after cluster-level
+    /// commands from `[before_serve."<cluster>"]`.
+    #[serde(default)]
+    pub before_serve: Vec<PreServeCommand>,
 }
 
 fn default_engine() -> String {
@@ -340,6 +399,7 @@ nodes = [
             nccl: HashMap::new(),
             hf_cache_dir: None,
             init_port: 25000,
+            before_serve: HashMap::new(),
         };
         config.nccl.insert("test-cluster".to_string(), test_nccl());
 
@@ -357,6 +417,7 @@ nodes = [
             nccl: HashMap::new(),
             hf_cache_dir: None,
             init_port: 25000,
+            before_serve: HashMap::new(),
         };
         config.nccl.insert("test-cluster".to_string(), test_nccl());
 
@@ -374,6 +435,7 @@ nodes = [
             nccl: HashMap::new(),
             hf_cache_dir: None,
             init_port: 25000,
+            before_serve: HashMap::new(),
         };
         let result = select_cluster(&config, "auto");
         assert!(result.is_err());
@@ -388,6 +450,7 @@ nodes = [
             nccl: HashMap::new(),
             hf_cache_dir: None,
             init_port: 25000,
+            before_serve: HashMap::new(),
         };
         let result = select_cluster(&config, "nonexistent");
         assert!(result.is_err());
@@ -404,6 +467,7 @@ nodes = [
             nccl: nccl_map,
             hf_cache_dir: None,
             init_port: 25000,
+            before_serve: HashMap::new(),
         };
         let cluster = select_cluster_by_name(&config, "test-cluster").unwrap();
         assert_eq!(cluster.name, "test-cluster");
@@ -419,6 +483,7 @@ nodes = [
             nccl: HashMap::new(),
             hf_cache_dir: None,
             init_port: 25000,
+            before_serve: HashMap::new(),
         };
         let result = select_cluster_by_name(&config, "no-such-cluster");
         assert!(result.is_err());

src/docker/mod.rs

@@ -12,6 +12,7 @@ pub async fn serve(
     recipe_name: &str,
     cluster_name: &str,
     keep: bool,
+    skip_before_serve: bool,
     dry_run: bool,
 ) -> Result<()> {
     let recipe = load_recipe(recipe_name).await?;
@@ -34,6 +35,8 @@ pub async fn serve(
         }
     }
 
+    crate::before_serve::run(config, &cluster, &recipe, skip_before_serve, dry_run).await?;
+
     let image = try_repaired_image(config, &cluster, &recipe).await?;
     let mut recipe = recipe;
     if image != recipe.docker_image {

src/docker/tests.rs

@@ -48,6 +48,7 @@ fn trec() -> Recipe {
         cluster: None,
         entrypoint: vec![],
         distributed_backend: "mp".into(),
+        before_serve: vec![],
     }
 }
 
@@ -181,6 +182,7 @@ fn sglang_recipe() -> Recipe {
         cluster: None,
         entrypoint: vec!["sglang".into(), "serve".into()],
         distributed_backend: "mp".into(),
+        before_serve: vec![],
     }
 }
 

src/doctor.rs

@@ -322,6 +322,7 @@ mod tests {
             cluster: None,
             entrypoint: vec![],
             distributed_backend: "mp".to_string(),
+            before_serve: vec![],
         }
     }
 

src/exec.rs

@@ -0,0 +1,105 @@
+use anyhow::{Context, Result};
+use tracing::{info, warn};
+
+use crate::config::{Cluster, Config, Recipe, select_cluster};
+
+/// Resolve the target cluster for `cargo run exec` from `--cluster` or
+/// `--recipe` (exactly one of which must be given).
+pub async fn resolve_target(
+    config: &Config,
+    cluster_name: Option<&str>,
+    recipe_name: Option<&str>,
+) -> Result<(Cluster, Option<Recipe>)> {
+    match (cluster_name, recipe_name) {
+        (Some(_), Some(_)) => {
+            anyhow::bail!("Specify either --cluster or --recipe, not both")
+        }
+        (None, None) => {
+            anyhow::bail!("Must specify either --cluster or --recipe")
+        }
+        (Some(name), None) => {
+            let cluster = crate::config::select_cluster_by_name(config, name)?;
+            Ok((cluster, None))
+        }
+        (None, Some(rname)) => {
+            let recipe = crate::config::load_recipe(rname).await?;
+            let cname = recipe.cluster.as_deref().unwrap_or("auto");
+            let (cluster, _) = select_cluster(config, cname)
+                .with_context(|| format!("resolving cluster for recipe '{}'", rname))?;
+            Ok((cluster, Some(recipe)))
+        }
+    }
+}
+
+/// Run `command` on every node of `cluster` (or a single node if `node`
+/// is given) and stream the output.  Used by the `cargo run exec`
+/// subcommand to support mass-execution flows like `sudo shutdown now`.
+pub async fn run(
+    config: &Config,
+    cluster: &Cluster,
+    node_name: Option<&str>,
+    command: &str,
+    dry_run: bool,
+) -> Result<()> {
+    let targets: Vec<_> = match node_name {
+        Some(name) => cluster
+            .nodes
+            .iter()
+            .filter(|n| n.name == name)
+            .cloned()
+            .collect(),
+        None => cluster.nodes.clone(),
+    };
+
+    if targets.is_empty() {
+        if let Some(n) = node_name {
+            anyhow::bail!("Node '{}' not found in cluster '{}'", n, cluster.name);
+        }
+        anyhow::bail!("Cluster '{}' has no nodes", cluster.name);
+    }
+
+    info!("⚡ exec: {} ({} node(s))", command, targets.len());
+
+    let mut handles = Vec::with_capacity(targets.len());
+    for node in targets {
+        let config = config.clone();
+        let command = command.to_string();
+        let dry = dry_run;
+        handles.push(tokio::spawn(async move {
+            let name = node.name.clone();
+            let result = crate::ssh::run_ssh(&config, &node, command.clone(), dry).await;
+            (name, command, result)
+        }));
+    }
+
+    let mut any_failed = false;
+    for h in handles {
+        let (name, command, result) = h.await?;
+        match result {
+            Ok(output) => {
+                if !output.trim().is_empty() {
+                    println!("\n--- {} ---", name);
+                    print!("{}", output);
+                    if !output.ends_with('\n') {
+                        println!();
+                    }
+                }
+            }
+            Err(e) => {
+                any_failed = true;
+                warn!("✗ {} :: {} :: {}", name, command, e);
+            }
+        }
+    }
+
+    if any_failed {
+        warn!("exec completed with errors (some nodes reported failures)");
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    // Most logic here is thin wrappers over `run_ssh`; integration
+    // tests against a live cluster cover the bulk of the behaviour.
+}

src/main.rs

@@ -1,10 +1,12 @@
 use anyhow::Result;
 use clap::Parser;
 
+mod before_serve;
 mod cli;
 mod config;
 mod docker;
 mod doctor;
+mod exec;
 mod health;
 mod host_keys;
 mod ssh;
@@ -28,8 +30,9 @@ async fn main() -> Result<()> {
             recipe,
             cluster,
             keep,
+            skip_before_serve,
         } => {
-            docker::serve(&config, &recipe, &cluster, keep, dry_run).await?;
+            docker::serve(&config, &recipe, &cluster, keep, skip_before_serve, dry_run).await?;
         }
         Commands::Logs {
             recipe: recipe_name,
@@ -91,6 +94,18 @@ async fn main() -> Result<()> {
         Commands::InitHostKeys { cluster } => {
             host_keys::init_host_keys(&config, cluster.as_deref(), &cli.config).await?;
         }
+        Commands::Exec {
+            cluster,
+            recipe,
+            node,
+            command,
+        } => {
+            let (cluster, _recipe) =
+                exec::resolve_target(&config, cluster.as_deref(), recipe.as_deref()).await?;
+            // Join the trailing args into a single command line.
+            let cmd_str = command.join(" ");
+            exec::run(&config, &cluster, node.as_deref(), &cmd_str, dry_run).await?;
+        }
     }
 
     Ok(())

src/ssh.rs

@@ -38,6 +38,11 @@ pub async fn ssh_session(config: &Config, node: &Node) -> Result<client::Handle<
 }
 
 /// Runs a command on a node via SSH and returns stdout.
+///
+/// The remote command's exit status is tracked: if it is non-zero,
+/// an Err is returned whose `to_string()` includes the captured output
+/// for diagnosis.  Callers that don't care about exit status can ignore
+/// the Result and inspect stdout via the returned `String`.
 pub async fn run_ssh(
     config: &Config,
     node: &Node,
@@ -54,12 +59,30 @@ pub async fn run_ssh(
     channel.exec(true, command.as_bytes()).await?;
 
     let mut output = String::new();
+    let mut exit_status: Option<u32> = None;
     while let Some(msg) = channel.wait().await {
-        if let ChannelMsg::Data { data } = msg {
-            output.push_str(&String::from_utf8_lossy(&data));
+        match msg {
+            ChannelMsg::Data { data } => {
+                output.push_str(&String::from_utf8_lossy(&data));
+            }
+            ChannelMsg::ExitStatus { exit_status: s } => {
+                exit_status = Some(s);
+            }
+            _ => {}
         }
     }
-    Ok(output)
+
+    match exit_status {
+        Some(0) | None => Ok(output),
+        Some(code) => Err(anyhow::anyhow!(
+            "ssh {}@{}: command exited with status {}\n  command: {}\n  output:\n{}",
+            config.ssh_user,
+            node.ssh_ip,
+            code,
+            command,
+            output
+        )),
+    }
 }
 
 /// Runs a command via SSH with exponential backoff retry.

src/tmux/mod.rs

@@ -69,8 +69,12 @@ impl SessionHealth {
 /// 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 } => {
-            serve::tmux_serve(config, &recipe, &cluster, dry_run).await?;
+        TmuxCommands::Serve {
+            recipe,
+            cluster,
+            skip_before_serve,
+        } => {
+            serve::tmux_serve(config, &recipe, &cluster, skip_before_serve, dry_run).await?;
         }
         TmuxCommands::Stop { recipe, cluster } => {
             ops::tmux_stop(config, &recipe, &cluster, dry_run).await?;

src/tmux/serve.rs

@@ -12,6 +12,7 @@ pub async fn tmux_serve(
     config: &Config,
     recipe_name: &str,
     cluster_name: &str,
+    skip_before_serve: bool,
     dry_run: bool,
 ) -> Result<()> {
     let recipe_path = format!("recipes/{}/recipe.toml", recipe_name);
@@ -44,6 +45,8 @@ pub async fn tmux_serve(
         }
     }
 
+    crate::before_serve::run(config, &cluster, &recipe, skip_before_serve, dry_run).await?;
+
     let manager = TmuxManager {
         session_prefix: recipe_name.to_string(),
     };