feat: deployment discovery + stop ambiguity resolution

- Add docker::deploy module with discover_recipe_deployments()
  that queries all nodes for containers by recipe, groups by cluster
- stop now discovers all running deployments of a recipe
- If exactly 1 deployment → stops it
- If multiple → shows each variant with --node hints, errors out
- --node flag changed to Vec<String> (repeatable) for selecting
  specific nodes in the target deployment
- stop_cluster now accepts nodes: &[String] instead of Option<&str>

src/cli.rs

@@ -57,9 +57,9 @@ pub enum Commands {
         #[arg(long, required = true)]
         recipe: String,
 
-        /// Target a specific node within the cluster (default: all).
+        /// Target specific nodes (repeatable, e.g. --node dgx1 --node dgx2).
         #[arg(long)]
-        node: Option<String>,
+        node: Vec<String>,
 
         #[arg(long, default_value = "auto")]
         cluster: String,

src/docker/deploy.rs

@@ -0,0 +1,74 @@
+//! Deployment discovery — query running containers across all nodes and
+//! group them into deployment variants for disambiguation.
+
+use std::collections::{HashMap, HashSet};
+
+use crate::config::Config;
+use crate::ssh::run_ssh;
+
+/// A running deployment variant — a set of nodes on a cluster running a recipe.
+pub struct DeploymentVariant {
+    pub cluster: String,
+    pub nodes: Vec<String>,
+}
+
+/// Discover running containers for a recipe across all unique nodes.
+/// Groups into deployment variants by (cluster, node-set).
+pub async fn discover_recipe_deployments(
+    config: &Config,
+    recipe_name: &str,
+) -> anyhow::Result<Vec<DeploymentVariant>> {
+    let mut seen = HashSet::new();
+    let mut handles = Vec::new();
+    for cluster in &config.clusters {
+        for node in &cluster.nodes {
+            if seen.contains(&node.ssh_ip) {
+                continue;
+            }
+            seen.insert(node.ssh_ip.clone());
+            let cmd = format!(
+                "docker ps --filter label=sm121.recipe={} \
+                 --format '{{{{.Label \"sm121.cluster\"}}}}|{{{{.Names}}}}' 2>/dev/null",
+                recipe_name
+            );
+            let config = config.clone();
+            let node = node.clone();
+            let name = node.name.clone();
+            handles.push(tokio::spawn(async move {
+                let out = run_ssh(&config, &node, cmd, false)
+                    .await
+                    .unwrap_or_default();
+                (name, out)
+            }));
+        }
+    }
+    let mut by_cluster: HashMap<String, HashSet<String>> = HashMap::new();
+    for h in futures::future::join_all(handles).await {
+        let (node_name, output) = h.unwrap_or_default();
+        for line in output.lines() {
+            let line = line.trim();
+            if line.is_empty() {
+                continue;
+            }
+            let mut parts = line.splitn(2, '|');
+            let cluster = parts.next().unwrap_or("").to_string();
+            if cluster.is_empty() {
+                continue;
+            }
+            by_cluster
+                .entry(cluster)
+                .or_default()
+                .insert(node_name.clone());
+        }
+    }
+    let mut variants: Vec<DeploymentVariant> = by_cluster
+        .into_iter()
+        .map(|(cluster, nodes)| {
+            let mut nodes: Vec<String> = nodes.into_iter().collect();
+            nodes.sort();
+            DeploymentVariant { cluster, nodes }
+        })
+        .collect();
+    variants.sort_by(|a, b| a.cluster.cmp(&b.cluster));
+    Ok(variants)
+}

src/docker/mod.rs

@@ -1,3 +1,4 @@
+pub mod deploy;
 pub mod health;
 pub mod mods;
 pub mod tail;
@@ -176,26 +177,25 @@ pub async fn show_logs(
     }
     Ok(())
 }
-/// Stops and removes containers for a recipe on all nodes.
+/// Stops containers for a recipe on specific nodes of a cluster.
 pub async fn stop_cluster(
     config: &Config,
     cluster: &Cluster,
     recipe_name: &str,
-    node_name: Option<&str>,
+    nodes: &[String],
     dry_run: bool,
 ) -> Result<()> {
     info!(
         "{}",
         fluent_i18n::t!("cluster-stopping", { "name" => &cluster.name })
     );
-    let nodes: Vec<&Node> = if let Some(nn) = node_name {
-        let v: Vec<&Node> = cluster.nodes.iter().filter(|n| n.name == nn).collect();
-        if v.is_empty() {
-            anyhow::bail!("Node '{}' not found in cluster '{}'", nn, cluster.name);
-        }
-        v
-    } else {
+    let nodes: Vec<&Node> = if nodes.is_empty() {
         cluster.nodes.iter().collect()
+    } else {
+        nodes
+            .iter()
+            .filter_map(|nn| cluster.nodes.iter().find(|n| n.name == *nn))
+            .collect()
     };
     for node in &nodes {
         run_ssh(
@@ -278,7 +278,9 @@ pub async fn occupied_nodes(config: &Config) -> Result<HashSet<String>> {
     Ok(occupied)
 }
 
-/// Runtime options for [`start_cluster`].
+/// Find the cluster that has running containers for a recipe by reading
+/// the `sm121.cluster` label from live containers.  Used by stop/logs
+/// when `--cluster auto`.
 struct StartClusterOpts {
     cache_dir: std::path::PathBuf,
     keep: bool,

src/main.rs

@@ -113,8 +113,44 @@ async fn main() -> Result<()> {
             cluster,
             node,
         } => {
-            let cluster = resolve_cluster_for_op(&config, &recipe_name, &cluster).await?;
-            docker::stop_cluster(&config, &cluster, &recipe_name, node.as_deref(), dry_run).await?;
+            if !node.is_empty() {
+                // User specified specific nodes — stop only those.
+                let cluster = resolve_cluster_for_op(&config, &recipe_name, &cluster).await?;
+                docker::stop_cluster(&config, &cluster, &recipe_name, &node, dry_run).await?;
+            } else {
+                // Discover running deployments and resolve ambiguity.
+                let variants =
+                    docker::deploy::discover_recipe_deployments(&config, &recipe_name).await?;
+                if variants.is_empty() {
+                    anyhow::bail!("No running containers found for recipe '{}'", recipe_name);
+                }
+                if variants.len() == 1 {
+                    let v = &variants[0];
+                    let cluster = select_cluster_by_name(&config, &v.cluster)?;
+                    docker::stop_cluster(&config, &cluster, &recipe_name, &v.nodes, dry_run)
+                        .await?;
+                } else {
+                    let mut msg = format!(
+                        "Multiple deployments of '{}' found — use --node to select:\n",
+                        recipe_name
+                    );
+                    for (i, v) in variants.iter().enumerate() {
+                        let nodes = v
+                            .nodes
+                            .iter()
+                            .map(|n| format!("--node {}", n))
+                            .collect::<Vec<_>>()
+                            .join(" ");
+                        msg.push_str(&format!(
+                            "  {}: cluster '{}' with {}\n",
+                            i + 1,
+                            v.cluster,
+                            nodes
+                        ));
+                    }
+                    anyhow::bail!(msg);
+                }
+            }
         }
         Commands::Health {
             cluster: cluster_name,