diff --git a/src/function/agents.rs b/src/function/agents.rs index 8807fa8..071bcb2 100644 --- a/src/function/agents.rs +++ b/src/function/agents.rs @@ -565,7 +565,8 @@ impl Drop for CancelOnDrop { /// Runs a child agent to completion. When the child is a graph agent, /// `graph_inputs` is overlaid on its `initial_state` before the graph -/// starts; it is ignored for config-only agents. +/// starts. Passing inputs to a config-only agent is rejected before the +/// child runs, since nothing could consume them. fn run_child_agent_with_graph_inputs( mut child_ctx: RequestContext, initial_input: Input, diff --git a/src/graph/executor.rs b/src/graph/executor.rs index 9174698..08a293a 100644 --- a/src/graph/executor.rs +++ b/src/graph/executor.rs @@ -109,6 +109,13 @@ impl GraphExecutor { let max_iterations = graph.settings.max_loop_iterations; let graph_timeout = graph.settings.timeout.and_then(wall_clock); let max_concurrency = graph.settings.max_concurrency; + if max_concurrency > Semaphore::MAX_PERMITS { + bail!( + "Graph '{}': settings.max_concurrency {max_concurrency} exceeds the runtime limit of {}", + graph.name, + Semaphore::MAX_PERMITS + ); + } let graph = Arc::new(graph); let start = Instant::now(); @@ -1362,6 +1369,30 @@ nodes: assert!(err.contains("sleeper"), "error should name frontier: {err}"); } + #[tokio::test] + async fn settings_max_concurrency_above_semaphore_limit_errors_instead_of_panicking() { + let ws = TestWorkspace::new(); + let yaml = r#" +name: oversized_cap_test +start: done +settings: + validate_before_run: false +nodes: + done: + type: end + output: "done" +"#; + let mut graph: Graph = serde_yaml::from_str(yaml).unwrap(); + graph.settings.max_concurrency = Semaphore::MAX_PERMITS + 1; + let mut ctx = make_ctx(); + let err = GraphExecutor::new(graph, &ws.dir) + .execute(&mut ctx, create_abort_signal()) + .await + .expect_err("an oversized cap is an error, not a panic"); + let msg = format!("{err:#}"); + assert!(msg.contains("exceeds the runtime limit of"), "{msg}"); + } + #[tokio::test] async fn graph_timeout_zero_is_unbounded() { if !cmd_available("bash") { diff --git a/src/graph/map.rs b/src/graph/map.rs index 16651a5..67964fd 100644 --- a/src/graph/map.rs +++ b/src/graph/map.rs @@ -337,7 +337,7 @@ fn resolve_max_concurrency( Some(ConcurrencyCap::Fixed(0)) => { bail!("map node '{node_id}': max_concurrency 0; expected a positive integer") } - Some(ConcurrencyCap::Fixed(n)) => return Ok(*n), + Some(ConcurrencyCap::Fixed(n)) => return bounded_cap(node_id, *n), Some(ConcurrencyCap::Template(t)) => t, }; @@ -359,7 +359,7 @@ fn resolve_max_concurrency( }; match resolved { - Some(n) if n >= 1 => Ok(n), + Some(n) if n >= 1 => bounded_cap(node_id, n), _ => Err(anyhow!( "map node '{node_id}': max_concurrency template \"{template}\" resolved to \ {value} ({}); expected a positive integer", @@ -368,6 +368,18 @@ fn resolve_max_concurrency( } } +/// `Semaphore::new` panics above `MAX_PERMITS`; a cap that large is a +/// configuration error, not a crash. +fn bounded_cap(node_id: &str, n: usize) -> Result { + if n > Semaphore::MAX_PERMITS { + bail!( + "map node '{node_id}': max_concurrency {n} exceeds the runtime limit of {}", + Semaphore::MAX_PERMITS + ); + } + Ok(n) +} + /// Pre-provision one teammate identity per item before any chain starts, so a /// fast sibling can message one still waiting on the semaphore (the message /// queues in the pre-created inbox). The identity lives for the item's whole @@ -686,6 +698,32 @@ mod tests { assert!(msg.contains("map node 'm': max_concurrency \"4\""), "{msg}"); } + #[test] + fn resolve_max_concurrency_fixed_above_semaphore_limit_errors() { + let state = StateManager::new(HashMap::new()); + let cap = Some(ConcurrencyCap::Fixed(Semaphore::MAX_PERMITS + 1)); + let msg = resolve(cap, &state).unwrap_err().to_string(); + assert_eq!( + msg, + format!( + "map node 'm': max_concurrency {} exceeds the runtime limit of {}", + Semaphore::MAX_PERMITS + 1, + Semaphore::MAX_PERMITS + ) + ); + // The limit itself is fine. + let cap = Some(ConcurrencyCap::Fixed(Semaphore::MAX_PERMITS)); + assert_eq!(resolve(cap, &state).unwrap(), Semaphore::MAX_PERMITS); + } + + #[test] + fn resolve_max_concurrency_template_above_semaphore_limit_errors() { + let state = state_with("budget", Value::from((Semaphore::MAX_PERMITS as u64) + 1)); + let cap = Some(ConcurrencyCap::Template("{{budget}}".into())); + let msg = resolve(cap, &state).unwrap_err().to_string(); + assert!(msg.contains("exceeds the runtime limit of"), "{msg}"); + } + #[test] fn resolve_max_concurrency_accepts_padded_template() { let state = state_with("budget", Value::from(3)); diff --git a/src/graph/validator.rs b/src/graph/validator.rs index c19c855..98f18c1 100644 --- a/src/graph/validator.rs +++ b/src/graph/validator.rs @@ -8,6 +8,7 @@ use anyhow::{Result, bail}; use std::collections::{BTreeMap, HashSet, VecDeque}; use std::path::PathBuf; use std::sync::Arc; +use tokio::sync::Semaphore; #[derive(Debug, Clone)] pub struct ValidationError { @@ -537,6 +538,12 @@ impl GraphValidator { deadlock the executor", )); } + if graph.settings.max_concurrency > Semaphore::MAX_PERMITS { + result.error(ValidationError::new(format!( + "settings.max_concurrency must be <= {} (got {}); the runtime cannot allocate more permits", + Semaphore::MAX_PERMITS, graph.settings.max_concurrency + ))); + } for (node_id, node) in &graph.nodes { if let NodeType::Map(m) = &node.node_type @@ -548,6 +555,18 @@ impl GraphValidator { would deadlock the executor", )); } + if let NodeType::Map(m) = &node.node_type + && let Some(ConcurrencyCap::Fixed(n)) = m.max_concurrency + && n > Semaphore::MAX_PERMITS + { + result.error(ValidationError::with_node( + node_id, + format!( + "map node's `max_concurrency` must be <= {} (got {n}); the runtime cannot allocate more permits", + Semaphore::MAX_PERMITS + ), + )); + } } } @@ -3158,6 +3177,24 @@ mod tests { ); } + #[test] + fn settings_max_concurrency_above_semaphore_limit_errors() { + let mut graph = graph_with(vec![("e", end_node("e"))], "e"); + graph.settings.max_concurrency = Semaphore::MAX_PERMITS + 1; + + let result = validator().validate(&graph); + + assert!( + result.errors.iter().any(|e| e.message.contains(&format!( + "settings.max_concurrency must be <= {} (got {})", + Semaphore::MAX_PERMITS, + Semaphore::MAX_PERMITS + 1 + ))), + "expected graph-level max_concurrency upper-bound error: {:?}", + result.errors + ); + } + #[test] fn settings_max_concurrency_default_is_valid() { let graph = graph_with(vec![("e", end_node("e"))], "e"); @@ -3458,6 +3495,30 @@ mod tests { assert_eq!(w[1].node_id.as_deref(), Some("l")); } + #[test] + fn map_max_concurrency_above_semaphore_limit_errors() { + let mut map = map_node_basic("m", "br", Some("end")); + if let NodeType::Map(ref mut mm) = map.node_type { + mm.max_concurrency = Some(ConcurrencyCap::Fixed(Semaphore::MAX_PERMITS + 1)); + } + let branch = llm_with_state_updates("br", &[("output", "{{output}}")], None); + let graph = graph_with( + vec![("m", map), ("br", branch), ("end", end_node("end"))], + "m", + ); + + let result = validator().validate(&graph); + + assert!( + result.errors.iter().any(|e| e + .message + .contains(&format!("must be <= {}", Semaphore::MAX_PERMITS)) + && e.node_id.as_deref() == Some("m")), + "expected map max_concurrency upper-bound error: {:?}", + result.errors + ); + } + #[test] fn map_max_concurrency_zero_errors() { let mut map = map_node_basic("m", "br", Some("end"));