orchestrate: re-probe once before failing 'not enough free nodes'

Transient probe misses (ssh blip, node still warming up after stop or
power-on) surfaced as 'cluster X has only 1 nodes' twice today and were
misdiagnosed as stop-wait failures. The first select_cluster/error path
now sleeps retry_grace_secs (BootTiming, default 5 s), re-probes once,
and re-selects; if it still fails, the error names each node's state
(free/occupied/unreachable) instead of a bare count.
Tests: flaky_probe_recovers_on_retry, insufficient_nodes_error_names_the_reason.

recipes/mimo26-flash-mopd-2x/recipe.toml

@@ -51,17 +51,18 @@ args = [
     "--enable-multimodal",
     "--moe-runner-backend", "marlin",
     "--moe-a2a-backend", "none",
-    # Memory / scheduling (128 GB unified per node). Config B (2026-09-27):
-    # 0.85 → 0.88 — MOPD's +6 GB weights/rank pushed the pool under
-    # context-length (306,752 < 327,680: single 307K+ requests got refused
-    # at admission). The +4 GB should bring the pool back above the ceiling;
-    # measured on RL sibling at 0.85: avail ~16 GB/node after static alloc.
-    "--mem-fraction-static", "0.88",
+    # Memory / scheduling (128 GB unified per node). Config C (2026-09-28):
+    # 0.88 → 0.90 for the 512K target — pool at 0.88 was 481,600, short of
+    # 524,288 by ~43K tokens. Measured: pool grows ~50K tokens per 1 GB of
+    # static budget (~1.9 GB per 0.01 GMU), so 0.90 should land ~530-600K.
+    # Activation headroom after static at 0.90: ~11 GB/node — same regime as
+    # the 0.85 boots where long prefills never OOM'd.
+    "--mem-fraction-static", "0.90",
     # Model is natively 1M; the measured pool at GMU 0.85 is 340,736 tokens
     # (2x TP2: 83 GB weights/rank leaves ~25 GB/rank; 39/48 SWA layers keep it
     # large in tokens). 320K fits a single long request with headroom; a
     # 524K request is pool-infeasible on 2x (the 8x Pro lane runs 512K).
-    "--context-length", "327680",
+    "--context-length", "524288",
     "--max-running-requests", "8",
     "--chunked-prefill-size", "8192",
     "--page-size", "64",

src/orchestrate/mod.rs

@@ -30,6 +30,9 @@ pub struct BootTiming {
     pub boot_wait_secs: u64,
     pub poll_interval_secs: u64,
     pub max_poll_secs: u64,
+    /// Grace before the single re-probe when the first pass saw too few
+    /// free nodes (transient probe misses). Tests set this to 0.
+    pub retry_grace_secs: u64,
 }
 
 impl Default for BootTiming {
@@ -38,6 +41,7 @@ impl Default for BootTiming {
             boot_wait_secs: 60,
             poll_interval_secs: 10,
             max_poll_secs: 120,
+            retry_grace_secs: 5,
         }
     }
 }
@@ -69,13 +73,13 @@ where
     C: PowerController,
     N: NodePinger,
 {
-    let state = req.prober.probe(config).await;
+    let mut last_state = req.prober.probe(config).await;
 
-    let resolved = if let Some(nn) = node_name {
+    let mut resolved = if let Some(nn) = node_name {
         resolve_solo_node(config, nn)?
     } else {
         // Auto-power-on: if any nodes are unreachable, try powering on.
-        let state = if req.auto_power_on && state_has_unreachable(&state, config) {
+        let state = if req.auto_power_on && state_has_unreachable(&last_state, config) {
             let (c, _nccl) = resolve_cluster_simple(config, recipe, cluster_name)?;
             if let Some(pwr) = req.power {
                 power_on_if_off(&c, pwr).await?;
@@ -85,7 +89,7 @@ where
             // Re-probe after power-on.
             req.prober.probe(config).await
         } else {
-            state
+            last_state.clone()
         };
 
         // `--force`: clear occupied flags so select_cluster counts reachable
@@ -98,8 +102,38 @@ where
                 }
             }
         }
+        last_state = state.clone();
 
-        let (c, n) = select_cluster(config, cluster_name, recipe.required_nodes, &state)?;
+        let (c, n) = match select_cluster(config, cluster_name, recipe.required_nodes, &state) {
+            Ok(v) => v,
+            Err(e) => {
+                // A transient probe miss (ssh blip, node warming up) can
+                // make auto-selection see too few free nodes — re-probe
+                // once after the retry grace before giving up.
+                if req.force_occupied {
+                    return Err(e);
+                }
+                tokio::time::sleep(Duration::from_secs(req.boot_timing.retry_grace_secs)).await;
+                let state2 = req.prober.probe(config).await;
+                match select_cluster(config, cluster_name, recipe.required_nodes, &state2) {
+                    Ok(v) => {
+                        state = state2;
+                        last_state = state.clone();
+                        v
+                    }
+                    Err(e2) => {
+                        return Err(anyhow::anyhow!(
+                            "Recipe '{}' requires {} nodes, but cluster '{}' had too few free nodes after re-probe ({}; {})",
+                            recipe.name,
+                            recipe.required_nodes.unwrap_or(0),
+                            cluster_name,
+                            node_details(config, &state2, cluster_name),
+                            e2
+                        ));
+                    }
+                }
+            }
+        };
         let avail: Vec<Node> = c
             .nodes
             .into_iter()
@@ -114,16 +148,76 @@ where
         (c, n.cloned())
     };
 
-    // Validate node count.
+    // Validate node count. A transient probe miss (ssh blip, node still
+    // warming up after stop/power-on) must not kill the serve — re-probe
+    // once after a short grace window before giving up, and on failure
+    // report WHY each node was not counted (unreachable vs occupied).
     if let Some(required) = recipe.required_nodes {
+        let actual = resolved.0.nodes.len() as u32;
+        if actual < required && !req.force_occupied {
+            tokio::time::sleep(Duration::from_secs(req.boot_timing.retry_grace_secs)).await;
+            let state2 = req.prober.probe(config).await;
+            if let Ok((c2, n2)) =
+                select_cluster(config, cluster_name, recipe.required_nodes, &state2)
+            {
+                let avail2: Vec<Node> = c2
+                    .nodes
+                    .iter()
+                    .filter(|n| {
+                        state2
+                            .nodes
+                            .get(&n.name)
+                            .is_some_and(|ns| ns.reachable && !ns.occupied)
+                    })
+                    .cloned()
+                    .collect();
+                if avail2.len() as u32 >= required {
+                    info!(
+                        "cluster re-probe recovered '{}': {} free node(s) (first pass saw {})",
+                        c2.name,
+                        avail2.len(),
+                        actual
+                    );
+                    resolved = (
+                        Cluster {
+                            nodes: avail2,
+                            ..c2
+                        },
+                        n2.cloned(),
+                    );
+                } else {
+                    last_state = state2;
+                }
+            }
+        }
         let actual = resolved.0.nodes.len() as u32;
         if actual < required {
+            let detail = config
+                .clusters
+                .iter()
+                .find(|c| c.name == resolved.0.name)
+                .map(|c| {
+                    c.nodes
+                        .iter()
+                        .map(|n| match last_state.nodes.get(&n.name) {
+                            Some(ns) if !ns.reachable => {
+                                format!("{} unreachable (ssh probe)", n.name)
+                            }
+                            Some(ns) if ns.occupied => format!("{} occupied", n.name),
+                            Some(_) => format!("{} free", n.name),
+                            None => format!("{} unknown", n.name),
+                        })
+                        .collect::<Vec<_>>()
+                        .join(", ")
+                })
+                .unwrap_or_default();
             anyhow::bail!(
-                "Recipe '{}' requires {} nodes, but cluster '{}' has only {} nodes",
+                "Recipe '{}' requires {} nodes, but cluster '{}' has only {} nodes ({})",
                 recipe.name,
                 required,
                 resolved.0.name,
-                actual
+                actual,
+                detail
             );
         }
     }
@@ -156,6 +250,30 @@ fn resolve_solo_node(config: &Config, node_name: &str) -> Result<(Cluster, Optio
     Ok((solo_c, solo_n))
 }
 
+/// One-line per-node status list for the given cluster ("auto" → all
+/// clusters, deduped by node name) — diagnostic for selection failures.
+fn node_details(config: &Config, state: &ClusterState, cluster_name: &str) -> String {
+    let mut seen = std::collections::HashSet::new();
+    let mut parts = Vec::new();
+    for c in &config.clusters {
+        if cluster_name != "auto" && c.name != cluster_name {
+            continue;
+        }
+        for n in &c.nodes {
+            if !seen.insert(n.name.clone()) {
+                continue;
+            }
+            parts.push(match state.nodes.get(&n.name) {
+                Some(ns) if !ns.reachable => format!("{} unreachable", n.name),
+                Some(ns) if ns.occupied => format!("{} occupied", n.name),
+                Some(_) => format!("{} free", n.name),
+                None => format!("{} unknown", n.name),
+            });
+        }
+    }
+    parts.join(", ")
+}
+
 /// Resolve cluster by name or auto treating every config node as free
 /// (reachable, not occupied). Only used for the power-on pre-check
 /// before any probe has been made, so being optimistic is safe:

src/orchestrate/retry_tests.rs

@@ -0,0 +1,158 @@
+//! Regression tests for the transient-probe retry in `prepare_cluster`
+//! (kept out of `tests.rs` for the 500-line cap — `tests.rs` wires it in
+//! as a submodule so the shared helpers stay visible).
+
+use super::{MockPinger, MockPower, MockProber, test_cluster, test_config, test_recipe};
+use crate::cluster_state::{ClusterState, NodeState};
+use crate::config::Config;
+use crate::ends::NodeProber;
+use crate::orchestrate::{BootTiming, PrepareRequest, prepare_cluster};
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+#[tokio::test]
+async fn flaky_probe_recovers_on_retry() {
+    // First probe sees n1 unreachable (ssh blip / node still warming up);
+    // the single re-probe after the retry grace must recover the full
+    // node set instead of failing the serve with "only 1 nodes".
+    let c = test_cluster("c", &["n1", "n2"]);
+    let config = test_config(vec![c]);
+    let recipe = test_recipe(Some(2));
+
+    let mut first = ClusterState::default();
+    first.nodes.insert(
+        "n1".into(),
+        NodeState {
+            reachable: false,
+            occupied: false,
+        },
+    );
+    first.nodes.insert(
+        "n2".into(),
+        NodeState {
+            reachable: true,
+            occupied: false,
+        },
+    );
+    let mut second = ClusterState::default();
+    second.nodes.insert(
+        "n1".into(),
+        NodeState {
+            reachable: true,
+            occupied: false,
+        },
+    );
+    second.nodes.insert(
+        "n2".into(),
+        NodeState {
+            reachable: true,
+            occupied: false,
+        },
+    );
+
+    struct FlakyProber {
+        first: ClusterState,
+        second: ClusterState,
+        count: AtomicUsize,
+    }
+    impl NodeProber for FlakyProber {
+        async fn probe(&self, _config: &Config) -> ClusterState {
+            let n = self.count.fetch_add(1, Ordering::SeqCst);
+            if n == 0 {
+                self.first.clone()
+            } else {
+                self.second.clone()
+            }
+        }
+    }
+    let prober = FlakyProber {
+        first,
+        second,
+        count: AtomicUsize::new(0),
+    };
+    let pinger = MockPinger { reachable: true };
+    let power = MockPower {
+        results: HashMap::new(),
+        turned_on: std::sync::Mutex::new(Vec::new()),
+    };
+
+    let (cluster, _nccl) = prepare_cluster(
+        &config,
+        &recipe,
+        "auto",
+        None,
+        &PrepareRequest {
+            prober: &prober,
+            power: Some(&power),
+            pinger: &pinger,
+            auto_power_on: false,
+            boot_timing: BootTiming {
+                boot_wait_secs: 0,
+                poll_interval_secs: 0,
+                max_poll_secs: 10,
+                retry_grace_secs: 0,
+            },
+            force_occupied: false,
+        },
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(cluster.nodes.len(), 2);
+}
+
+#[tokio::test]
+async fn insufficient_nodes_error_names_the_reason() {
+    // When the retry still can't find enough free nodes, the error must
+    // say WHY each node was excluded (occupied vs unreachable) — the old
+    // bare count hid transient occupancy behind the "only N nodes" wall.
+    let c = test_cluster("c", &["n1", "n2"]);
+    let config = test_config(vec![c]);
+    let recipe = test_recipe(Some(2));
+
+    let mut state = ClusterState::default();
+    state.nodes.insert(
+        "n1".into(),
+        NodeState {
+            reachable: true,
+            occupied: true,
+        },
+    );
+    state.nodes.insert(
+        "n2".into(),
+        NodeState {
+            reachable: true,
+            occupied: false,
+        },
+    );
+    let prober = MockProber { state };
+    let pinger = MockPinger { reachable: true };
+    let power = MockPower {
+        results: HashMap::new(),
+        turned_on: std::sync::Mutex::new(Vec::new()),
+    };
+
+    let result = prepare_cluster(
+        &config,
+        &recipe,
+        "auto",
+        None,
+        &PrepareRequest {
+            prober: &prober,
+            power: Some(&power),
+            pinger: &pinger,
+            auto_power_on: false,
+            boot_timing: BootTiming {
+                boot_wait_secs: 0,
+                poll_interval_secs: 0,
+                max_poll_secs: 10,
+                retry_grace_secs: 0,
+            },
+            force_occupied: false,
+        },
+    )
+    .await;
+
+    let err = result.unwrap_err().to_string();
+    assert!(err.contains("n1 occupied"), "error was: {err}");
+}

src/orchestrate/tests.rs

@@ -1,4 +1,6 @@
 use super::*;
+#[path = "retry_tests.rs"]
+mod retry_tests;
 use crate::cluster_state::NodeState;
 use crate::config::{NcclConfig, Node};
 use std::collections::HashMap;
@@ -193,6 +195,7 @@ async fn cluster_off_auto_power_on_powers_and_selects() {
                 boot_wait_secs: 0,
                 poll_interval_secs: 0,
                 max_poll_secs: 10,
+                retry_grace_secs: 0,
             },
             force_occupied: false,
         },
@@ -250,6 +253,7 @@ async fn cluster_off_no_auto_power_fails() {
                 boot_wait_secs: 0,
                 poll_interval_secs: 0,
                 max_poll_secs: 10,
+                retry_grace_secs: 0,
             },
             force_occupied: false,
         },
@@ -306,6 +310,7 @@ async fn cluster_already_on_selects_directly() {
                 boot_wait_secs: 0,
                 poll_interval_secs: 0,
                 max_poll_secs: 10,
+                retry_grace_secs: 0,
             },
             force_occupied: false,
         },
@@ -361,6 +366,7 @@ async fn occupied_nodes_excluded() {
                 boot_wait_secs: 0,
                 poll_interval_secs: 0,
                 max_poll_secs: 10,
+                retry_grace_secs: 0,
             },
             force_occupied: false,
         },
@@ -418,6 +424,7 @@ async fn force_occupied_keeps_occupied_nodes() {
                 boot_wait_secs: 0,
                 poll_interval_secs: 0,
                 max_poll_secs: 10,
+                retry_grace_secs: 0,
             },
             force_occupied: true,
         },

src/power/mikro.rs

@@ -63,7 +63,10 @@ pub async fn shutdown_named(config: &Config, names: &[String]) {
             continue;
         };
         if let Err(e) = crate::mikrotik::shutdown_device(dev).await {
-            warn!("{}", fluent_i18n::t!("power-off-mikrotik-failed", { "name" => &dev.name, "error" => e.to_string() }));
+            warn!(
+                "{}",
+                fluent_i18n::t!("power-off-mikrotik-failed", { "name" => &dev.name, "error" => e.to_string() })
+            );
         }
         let Some(relay) = &dev.on_entity else {
             continue; // no relay configured: the API shutdown is all we have
@@ -71,12 +74,24 @@ pub async fn shutdown_named(config: &Config, names: &[String]) {
         match &config.home_assistant {
             Some(ha_cfg) => match crate::home_assistant::Client::new(ha_cfg) {
                 Ok(ha) => match ha.turn_off(relay).await {
-                    Ok(()) => info!("{}", fluent_i18n::t!("power-off-mikrotik-relay", { "name" => &dev.name, "relay" => relay })),
-                    Err(e) => warn!("{}", fluent_i18n::t!("power-off-mikrotik-relay-failed", { "name" => &dev.name, "relay" => relay, "error" => e.to_string() })),
+                    Ok(()) => info!(
+                        "{}",
+                        fluent_i18n::t!("power-off-mikrotik-relay", { "name" => &dev.name, "relay" => relay })
+                    ),
+                    Err(e) => warn!(
+                        "{}",
+                        fluent_i18n::t!("power-off-mikrotik-relay-failed", { "name" => &dev.name, "relay" => relay, "error" => e.to_string() })
+                    ),
                 },
-                Err(e) => warn!("{}", fluent_i18n::t!("power-off-mikrotik-relay-failed", { "name" => &dev.name, "relay" => relay, "error" => e.to_string() })),
+                Err(e) => warn!(
+                    "{}",
+                    fluent_i18n::t!("power-off-mikrotik-relay-failed", { "name" => &dev.name, "relay" => relay, "error" => e.to_string() })
+                ),
             },
-            None => warn!("{}", fluent_i18n::t!("power-off-mikrotik-relay-failed", { "name" => &dev.name, "relay" => relay, "error" => "no HA configured" })),
+            None => warn!(
+                "{}",
+                fluent_i18n::t!("power-off-mikrotik-relay-failed", { "name" => &dev.name, "relay" => relay, "error" => "no HA configured" })
+            ),
         }
     }
 }