fix: fail fast when a rule exceeds the --max-memory/--max-threads cap

crates/oxo-flow-cli/src/commands/run.rs

@@ -157,6 +157,32 @@ pub async fn run_command(
         interpreter_map: config.workflow.interpreter_map.clone(),
     };
 
+    // Fail fast if any rule's declared request can never fit an explicit
+    // --max-memory / --max-threads cap. Otherwise the run would execute earlier
+    // rules and only discover the impossible one mid-pipeline.
+    let scheduled: Vec<&oxo_flow_core::rule::Rule> = order
+        .iter()
+        .filter_map(|name| config.get_rule(name))
+        .collect();
+    let breaches = oxo_flow_core::scheduler::check_budget_feasibility(
+        &scheduled,
+        exec_config.max_threads,
+        exec_config.max_memory_mb,
+    );
+    if !breaches.is_empty() {
+        progress.finish_and_clear();
+        let detail = breaches
+            .iter()
+            .map(|b| format!("  - {b}"))
+            .collect::<Vec<_>>()
+            .join("\n");
+        return Err(anyhow::anyhow!(
+            "resource budget too small for {} rule(s); no rules were run:\n{}",
+            breaches.len(),
+            detail
+        ));
+    }
+
     let executor = LocalExecutor::new(exec_config);
     let mut success_count = 0;
     let mut fail_count = 0;

crates/oxo-flow-core/src/scheduler.rs

@@ -342,6 +342,43 @@ pub fn parse_memory_mb(memory: &str) -> Option<u64> {
     Some(mb_int)
 }
 
+/// Pre-flight check: report rules whose declared request can never fit an
+/// *explicitly configured* budget (`--max-threads` / `--max-memory`), so the
+/// run can fail fast instead of dying mid-pipeline on an impossible rule.
+/// A cap of `None` means "auto-detect" and is not treated as a hard limit.
+/// Returns one message per breaching rule (empty if feasible).
+pub fn check_budget_feasibility(
+    rules: &[&Rule],
+    max_threads: Option<u32>,
+    max_memory_mb: Option<u64>,
+) -> Vec<String> {
+    let mut breaches = Vec::new();
+    for rule in rules {
+        if let Some(cap) = max_memory_mb {
+            let required = rule
+                .effective_memory()
+                .and_then(parse_memory_mb)
+                .unwrap_or(0);
+            if required > cap {
+                breaches.push(format!(
+                    "rule '{}' requires {}MB memory but --max-memory caps the run at {}MB",
+                    rule.name, required, cap
+                ));
+            }
+        }
+        if let Some(cap) = max_threads {
+            let required = rule.effective_threads();
+            if required > cap {
+                breaches.push(format!(
+                    "rule '{}' requires {} threads but --max-threads caps the run at {}",
+                    rule.name, required, cap
+                ));
+            }
+        }
+    }
+    breaches
+}
+
 /// Validate that a rule's resource requirements don't exceed system capacity.
 /// Returns a list of warning messages (empty if all requirements are within limits).
 pub fn validate_resources_against_system(
@@ -876,6 +913,57 @@ mod tests {
         assert!(warnings.is_empty());
     }
 
+    #[test]
+    fn budget_breach_on_explicit_memory_cap() {
+        let rule = Rule {
+            name: "hungry".to_string(),
+            memory: Some("8G".to_string()),
+            ..Default::default()
+        };
+        let breaches = check_budget_feasibility(&[&rule], None, Some(100));
+        assert_eq!(breaches.len(), 1);
+        assert!(breaches[0].contains("hungry"));
+        assert!(breaches[0].contains("--max-memory"));
+    }
+
+    #[test]
+    fn budget_breach_on_explicit_threads_cap() {
+        let rule = Rule {
+            name: "wide".to_string(),
+            threads: Some(16),
+            ..Default::default()
+        };
+        let breaches = check_budget_feasibility(&[&rule], Some(4), None);
+        assert_eq!(breaches.len(), 1);
+        assert!(breaches[0].contains("wide"));
+        assert!(breaches[0].contains("--max-threads"));
+    }
+
+    #[test]
+    fn budget_no_breach_within_caps() {
+        let rule = Rule {
+            name: "fits".to_string(),
+            threads: Some(2),
+            memory: Some("2G".to_string()),
+            ..Default::default()
+        };
+        let breaches = check_budget_feasibility(&[&rule], Some(4), Some(4096));
+        assert!(breaches.is_empty());
+    }
+
+    #[test]
+    fn budget_none_cap_is_not_a_hard_limit() {
+        // Auto-detect (None) must not be treated as a breach even for a huge request.
+        let rule = Rule {
+            name: "huge".to_string(),
+            threads: Some(1024),
+            memory: Some("1T".to_string()),
+            ..Default::default()
+        };
+        let breaches = check_budget_feasibility(&[&rule], None, None);
+        assert!(breaches.is_empty());
+    }
+
     #[test]
     fn check_available_disk_mb_returns_value() {
         let dir = tempfile::tempdir().unwrap();

tests/cli_integration.rs

@@ -493,6 +493,56 @@ shell = "cat mid.txt > output.txt"
     assert!(stderr.contains("Debugging 1 rules"));
 }
 
+#[test]
+fn cli_run_fails_fast_when_rule_exceeds_max_memory() {
+    // A rule declaring more memory than the explicit --max-memory cap can never
+    // be scheduled. The run must fail up front with a clear message and must NOT
+    // execute the earlier, feasible rule (no wasted work).
+    let dir = tempfile::tempdir().unwrap();
+    let workflow = dir.path().join("test.oxoflow");
+    fs::write(
+        &workflow,
+        r#"
+[workflow]
+name = "budget-test"
+version = "1.0.0"
+
+[[rules]]
+name = "cheap_first"
+output = ["a.txt"]
+shell = "echo did-real-work > a.txt"
+
+[[rules]]
+name = "hungry_second"
+input = ["a.txt"]
+output = ["b.txt"]
+shell = "echo hello > b.txt"
+memory = "8G"
+"#,
+    )
+    .unwrap();
+
+    let output = oxo_flow_cmd()
+        .current_dir(dir.path())
+        .args(["run", workflow.to_str().unwrap(), "--max-memory", "100"])
+        .output()
+        .expect("failed to run command");
+
+    assert!(
+        !output.status.success(),
+        "run should fail when a rule exceeds the memory budget"
+    );
+    let stderr = String::from_utf8_lossy(&output.stderr);
+    assert!(
+        stderr.contains("hungry_second") && stderr.contains("--max-memory"),
+        "error should name the breaching rule and the cap: {stderr}"
+    );
+    assert!(
+        !dir.path().join("a.txt").exists(),
+        "no rules should have run; cheap_first must not have produced a.txt"
+    );
+}
+
 // ─── Cluster CLI tests ──────────────────────────────────────────────────────
 
 #[test]