From 80ace77e4402eabd29cd810d301d2c269896997b Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Sat, 12 Sep 2026 14:42:42 -0600 Subject: [PATCH] feat(TASK-016): engine + test fixes from the Phase-1 re-review - warn-per-retry test via shared test-only WarnCollector (src/testing.rs) - extraction-placement test: extraction failures are never retried - is_transient_error: typed reqwest/Elapsed pre-check + anchored 429 matchers - outcome_from docstring corrected; validator warning for fallback+output_schema without state_updates (llm + agent) - validator error for max_attempts == 0 (llm + agent); runtime backstop kept - nits: generalized branch-escape error text, Continue payload comment, BFS test assignments assert, node_id threaded into AgentNodeExecutor warns - bundled-graph warning baseline records 3 true-positive step-runner nodes (orient, edge_case_sweep, write_handoff) pending state_updates there --- src/config/request_context.rs | 30 ++++- src/config/tool_scope.rs | 36 +----- src/graph/agent.rs | 193 +++++++++++++++++++++++++---- src/graph/executor.rs | 3 +- src/graph/map.rs | 8 +- src/graph/mod.rs | 53 +++++++- src/graph/validator.rs | 221 ++++++++++++++++++++++++++++++++++ src/main.rs | 2 + src/testing.rs | 43 +++++++ 9 files changed, 519 insertions(+), 70 deletions(-) create mode 100644 src/testing.rs diff --git a/src/config/request_context.rs b/src/config/request_context.rs index ef04078..27ce19b 100644 --- a/src/config/request_context.rs +++ b/src/config/request_context.rs @@ -8200,6 +8200,20 @@ mod tests { .collect() } + fn fallback_capture(node_ids: &[&str]) -> Vec { + node_ids + .iter() + .map(|id| { + format!( + "{id}: llm node declares `fallback` with `output_schema` but no \ + `state_updates`; on failure the \"…failed: \" string is \ + only written through `state_updates`, so the fallback node \ + cannot see why it was reached" + ) + }) + .collect() + } + let _guard = TestConfigDirGuard::new(); Agent::install_builtin_agents(false).unwrap(); @@ -8238,7 +8252,9 @@ mod tests { // nodes, plus variable-shadowing on `coder` and `step-runner` (they // declare variables whose names are also `initial_state` keys). A new // rule that fires on a shipped asset (or a new bundled graph) must be - // recorded here deliberately. + // recorded here deliberately. The fallback-capture warnings on + // `step-runner` are true positives, recorded until those nodes gain + // `state_updates` that capture the failure string. let expected_warnings = BTreeMap::from([ ("adversary".to_string(), Vec::new()), ("code-reviewer".to_string(), Vec::new()), @@ -8266,9 +8282,8 @@ mod tests { ("finding-verifier".to_string(), Vec::new()), ("librarian".to_string(), Vec::new()), ("review-gauntlet".to_string(), Vec::new()), - ( - "step-runner".to_string(), - [ + ("step-runner".to_string(), { + let mut lines = [ shadowed(&["plans_dir", "project_dir"]), unreachable(&[ "check_handoff", @@ -8290,9 +8305,12 @@ mod tests { "verify_tests", "write_handoff", ]), + fallback_capture(&["edge_case_sweep", "orient", "write_handoff"]), ] - .concat(), - ), + .concat(); + lines.sort(); + lines + }), ]); assert_eq!( warnings_by_agent, expected_warnings, diff --git a/src/config/tool_scope.rs b/src/config/tool_scope.rs index 4e8be6f..d106ae3 100644 --- a/src/config/tool_scope.rs +++ b/src/config/tool_scope.rs @@ -918,42 +918,8 @@ mod tests { use super::*; use crate::config::mcp_tool_policy::LayerSource; use crate::function::ToolCall; - use log::{Level, LevelFilter, Log, Metadata, Record}; + use crate::testing::{install_warn_collector, warn_messages}; use std::sync::atomic::Ordering; - use std::sync::{Mutex, Once, OnceLock}; - - struct WarnCollector; - - static WARN_MESSAGES: OnceLock>> = OnceLock::new(); - - fn warn_messages() -> &'static Mutex> { - WARN_MESSAGES.get_or_init(Mutex::default) - } - - impl Log for WarnCollector { - fn enabled(&self, metadata: &Metadata) -> bool { - metadata.level() <= Level::Warn - } - - fn log(&self, record: &Record) { - if self.enabled(record.metadata()) { - warn_messages() - .lock() - .unwrap() - .push(record.args().to_string()); - } - } - - fn flush(&self) {} - } - - fn install_warn_collector() { - static INSTALL: Once = Once::new(); - INSTALL.call_once(|| { - log::set_logger(&WarnCollector).expect("no other logger should be installed"); - log::set_max_level(LevelFilter::Warn); - }); - } #[test] fn mcp_runtime_new_is_empty() { diff --git a/src/graph/agent.rs b/src/graph/agent.rs index 974d8b8..751bae7 100644 --- a/src/graph/agent.rs +++ b/src/graph/agent.rs @@ -20,6 +20,8 @@ const DEFAULT_TIMEOUT_SECS: u64 = 300; #[derive(Debug, PartialEq, Eq)] pub(super) enum AgentExecutionOutcome { + /// Carries the raw agent output for tests; the executor's agent arm + /// ignores the payload. Continue(String), FellBack(String), } @@ -28,25 +30,36 @@ pub struct AgentNodeExecutor; impl AgentNodeExecutor { pub(super) async fn execute( + node_id: &str, node: &AgentNode, state_manager: &mut StateManager, parent_ctx: &mut RequestContext, retire_peer_on_return: bool, ) -> Result { - let result = run(node, state_manager, parent_ctx, retire_peer_on_return).await; - outcome_from(node, state_manager, result) + let result = run( + node_id, + node, + state_manager, + parent_ctx, + retire_peer_on_return, + ) + .await; + outcome_from(node_id, node, state_manager, result) } } /// Turns the node run's final result into a routing outcome. With a /// `fallback` declared, any failure — retries exhausted, hard error, or /// extraction failure — is written into state as -/// `"Agent node failed: "` (through the node's own state_updates / -/// output-schema path, so the fallback node can interpolate it) and routes -/// to the fallback. Without one the error propagates unchanged; unlike llm -/// nodes there is no teaching bail, because agent nodes always failed loudly -/// and callers assert on the existing "Agent 'X' failed" chains. +/// `"Agent node failed: "` and routes to the fallback. Only the +/// node's `state_updates` can capture that string (bound as `{{output}}` +/// during interpolation); the output-schema auto-merge never applies to it, +/// because it only merges object values and the failure is a plain string. +/// Without a fallback the error propagates unchanged; unlike llm nodes +/// there is no teaching bail, because agent nodes always failed loudly and +/// callers assert on the existing "Agent 'X' failed" chains. fn outcome_from( + node_id: &str, node: &AgentNode, state_manager: &mut StateManager, result: Result, @@ -55,7 +68,7 @@ fn outcome_from( Ok(raw) => Ok(AgentExecutionOutcome::Continue(raw)), Err(e) => match &node.fallback { Some(fb) => { - warn!("agent node failed, routing to fallback '{fb}': {e:#}"); + warn!("agent node '{node_id}' failed, routing to fallback '{fb}': {e:#}"); state_updates::apply( state_manager, &Value::String(format!("Agent node failed: {e:#}")), @@ -70,6 +83,7 @@ fn outcome_from( } async fn run( + node_id: &str, node: &AgentNode, state_manager: &mut StateManager, parent_ctx: &mut RequestContext, @@ -102,10 +116,37 @@ async fn run( .await }) }); - let raw = run_with_retries(node, parent_ctx, retire_peer_on_return, &mut run_attempt).await?; + attempt_and_extract( + node_id, + node, + state_manager, + parent_ctx, + retire_peer_on_return, + &mut run_attempt, + ) + .await +} + +/// Drives the retry loop, then extraction, then state updates. Extraction +/// stays outside the retry loop: by the time it runs the agent itself +/// succeeded, so an extraction failure is never retried. +async fn attempt_and_extract( + node_id: &str, + node: &AgentNode, + state_manager: &mut StateManager, + parent_ctx: &mut RequestContext, + retire_peer_on_return: bool, + run_attempt: &mut AttemptRunner<'_>, +) -> Result { + let raw = run_with_retries( + node_id, + node, + parent_ctx, + retire_peer_on_return, + run_attempt, + ) + .await?; - // Extraction stays outside the retry loop: by this point the agent - // itself succeeded, so an extraction failure is never retried. let output_value = match &node.output_schema { Some(schema) => structured::extract(&raw, schema, parent_ctx) .await @@ -159,6 +200,7 @@ fn boxed_attempt<'a>(fut: impl Future> + Send + 'a) -> A /// after extraction. Chains re-arm the same identity for later steps and /// leave retirement to the chain runner. async fn run_with_retries( + node_id: &str, node: &AgentNode, parent_ctx: &mut RequestContext, retire_peer_on_return: bool, @@ -167,7 +209,15 @@ async fn run_with_retries( let assignment = parent_ctx.peer_assignment.clone(); let registry = parent_ctx.peer_registry.clone(); - let result = retry_transient(node, parent_ctx, &assignment, ®istry, run_attempt).await; + let result = retry_transient( + node_id, + node, + parent_ctx, + &assignment, + ®istry, + run_attempt, + ) + .await; if retire_peer_on_return && let (Some(registry), Some((id, _))) = (®istry, &assignment) { registry.mark_finished(id); @@ -177,6 +227,7 @@ async fn run_with_retries( } async fn retry_transient( + node_id: &str, node: &AgentNode, parent_ctx: &mut RequestContext, assignment: &Option<(String, Arc)>, @@ -196,7 +247,9 @@ async fn retry_transient( match run_attempt(parent_ctx).await { Ok(out) => return Ok(out), Err(e) if is_transient_error(&e) && attempt < node.max_attempts => { - warn!("agent node attempt {attempt} failed (transient): {e:#}; retrying"); + warn!( + "agent node '{node_id}' attempt {attempt} failed (transient): {e:#}; retrying" + ); last_err = Some(e); } Err(e) => return Err(e), @@ -262,6 +315,7 @@ mod tests { use super::*; use crate::config::{AppState, WorkingMode, default_max_agent_depth}; use crate::supervisor::mailbox::{Inbox, PeerRegistry, graph_agent_id}; + use crate::testing::{install_warn_collector, warn_messages}; use serde_json::json; use std::collections::HashMap; use std::sync::Arc; @@ -305,9 +359,10 @@ mod tests { node.teammates = true; let mut state = manager_with(&[]); - let err = AgentNodeExecutor::execute(&node, &mut state, ctx, retire_peer_on_return) - .await - .expect_err("agent past max depth should fail before running"); + let err = + AgentNodeExecutor::execute("test_node", &node, &mut state, ctx, retire_peer_on_return) + .await + .expect_err("agent past max depth should fail before running"); let chain = format!("{err:#}"); assert!(chain.contains("Agent 'test_agent' failed"), "{chain}"); @@ -349,6 +404,7 @@ mod tests { let mut attempts = 0u32; let out = run_with_retries( + "test_node", &node, &mut ctx, false, @@ -376,6 +432,7 @@ mod tests { let mut attempts = 0u32; let err = run_with_retries( + "test_node", &node, &mut ctx, false, @@ -398,6 +455,7 @@ mod tests { let mut attempts = 0u32; let err = run_with_retries( + "test_node", &node, &mut ctx, false, @@ -424,6 +482,7 @@ mod tests { let mut attempts = 0u32; let out = run_with_retries( + "test_node", &node, &mut ctx, false, @@ -470,6 +529,7 @@ mod tests { let mut observations: Vec<(String, bool, bool)> = Vec::new(); let out = run_with_retries( + "test_node", &node, &mut ctx, true, @@ -515,6 +575,7 @@ mod tests { let mut attempts = 0u32; let err = run_with_retries( + "test_node", &node, &mut ctx, true, @@ -543,6 +604,7 @@ mod tests { let node = retryable_node(1); run_with_retries( + "test_node", &node, &mut ctx, false, @@ -565,6 +627,7 @@ mod tests { let mut attempts = 0u32; let err = run_with_retries( + "test_node", &node, &mut ctx, false, @@ -580,6 +643,92 @@ mod tests { assert!(format!("{err:#}").contains("agent node exhausted retries")); } + #[tokio::test] + async fn retry_transient_warns_once_per_retried_attempt_with_the_context_chain() { + install_warn_collector(); + let mut ctx = plain_ctx(); + let node = retryable_node(3); + let mut attempts = 0u32; + + run_with_retries( + "warn_capture", + &node, + &mut ctx, + false, + &mut attempt_runner(|_ctx| { + attempts += 1; + boxed_attempt(async move { + Err::( + anyhow!("connection error: warn-capture reset") + .context("Agent 'test_agent' failed"), + ) + }) + }), + ) + .await + .expect_err("all attempts fail"); + + assert_eq!(attempts, 3); + let warns: Vec = warn_messages() + .lock() + .unwrap() + .iter() + .filter(|m| m.contains("'warn_capture'")) + .cloned() + .collect(); + assert_eq!( + warns.len(), + 2, + "one warn per retried attempt (the final attempt propagates instead): {warns:?}" + ); + for (i, warn) in warns.iter().enumerate() { + let attempt = i + 1; + assert!( + warn.contains(&format!( + "agent node 'warn_capture' attempt {attempt} failed (transient)" + )), + "{warn}" + ); + assert!( + warn.contains("Agent 'test_agent' failed: connection error: warn-capture reset"), + "warn must carry the full context chain: {warn}" + ); + } + } + + #[tokio::test] + async fn extraction_failure_after_a_successful_run_is_never_retried() { + let mut ctx = plain_ctx(); + let mut node = retryable_node(2); + node.output_schema = Some(json!({"type": "object"})); + let mut state = manager_with(&[]); + let mut attempts = 0u32; + + let err = attempt_and_extract( + "test_node", + &node, + &mut state, + &mut ctx, + false, + &mut attempt_runner(|_ctx| { + attempts += 1; + boxed_attempt(async move { Ok("not json".to_string()) }) + }), + ) + .await + .expect_err("extraction must fail without an extractor model"); + + assert_eq!( + attempts, 1, + "an extraction failure must not re-run the agent" + ); + let chain = format!("{err:#}"); + assert!( + chain.contains("output failed structured-output extraction"), + "{chain}" + ); + } + fn failure_capture_updates() -> HashMap { let mut u = HashMap::new(); u.insert("failure".into(), "{{output}}".into()); @@ -594,7 +743,7 @@ mod tests { node.fallback = Some("recover".into()); let mut state = manager_with(&[]); - let outcome = AgentNodeExecutor::execute(&node, &mut state, &mut ctx, false) + let outcome = AgentNodeExecutor::execute("test_node", &node, &mut state, &mut ctx, false) .await .unwrap(); @@ -620,7 +769,7 @@ mod tests { let node = node_with("hi", None); let mut state = manager_with(&[]); - let outcome = outcome_from(&node, &mut state, Ok("raw".into())).unwrap(); + let outcome = outcome_from("test_node", &node, &mut state, Ok("raw".into())).unwrap(); assert_eq!(outcome, AgentExecutionOutcome::Continue("raw".into())); } @@ -631,7 +780,8 @@ mod tests { node.fallback = Some("recover".into()); let mut state = manager_with(&[]); - let outcome = outcome_from(&node, &mut state, Err(extraction_error())).unwrap(); + let outcome = + outcome_from("test_node", &node, &mut state, Err(extraction_error())).unwrap(); assert_eq!(outcome, AgentExecutionOutcome::FellBack("recover".into())); let failure = state @@ -651,7 +801,7 @@ mod tests { let node = node_with("hi", Some(failure_capture_updates())); let mut state = manager_with(&[]); - let err = outcome_from(&node, &mut state, Err(extraction_error())) + let err = outcome_from("test_node", &node, &mut state, Err(extraction_error())) .expect_err("no fallback must propagate"); assert_eq!( @@ -670,6 +820,7 @@ mod tests { let mut attempts = 0u32; let result = run_with_retries( + "test_node", &node, &mut ctx, false, @@ -682,7 +833,7 @@ mod tests { ) .await; let mut state = manager_with(&[]); - let outcome = outcome_from(&node, &mut state, result).unwrap(); + let outcome = outcome_from("test_node", &node, &mut state, result).unwrap(); assert_eq!(attempts, 2, "fallback must not preempt the retry budget"); assert_eq!(outcome, AgentExecutionOutcome::FellBack("recover".into())); diff --git a/src/graph/executor.rs b/src/graph/executor.rs index 805516d..08a405c 100644 --- a/src/graph/executor.rs +++ b/src/graph/executor.rs @@ -575,7 +575,8 @@ pub(super) async fn step( match &node.node_type { NodeType::Agent(agent_node) => { let outcome = - AgentNodeExecutor::execute(agent_node, state, ctx, !step_ctx.branch_mode).await?; + AgentNodeExecutor::execute(current, agent_node, state, ctx, !step_ctx.branch_mode) + .await?; let targets = match outcome { AgentExecutionOutcome::Continue(_) => { static_next_targets(node, current, "agent", step_ctx.branch_mode)? diff --git a/src/graph/map.rs b/src/graph/map.rs index 56ec9cf..3e35032 100644 --- a/src/graph/map.rs +++ b/src/graph/map.rs @@ -294,8 +294,8 @@ async fn run_chain_step( branch_nodes.sort_unstable(); bail!( "routed to '{target}' which is outside the branch subgraph rooted at \ - '{}' (branch nodes: {}). Script `_next` targets inside a map branch \ - must stay within the branch.", + '{}' (branch nodes: {}). Routing targets inside a map branch — script \ + `_next` or node fallback — must stay within the branch.", chain.entry, branch_nodes.join(", ") ); @@ -1282,7 +1282,8 @@ nodes: ); assert!( chain.contains( - "Script `_next` targets inside a map branch must stay within the branch." + "Routing targets inside a map branch — script `_next` or node fallback — \ + must stay within the branch." ), "{chain}" ); @@ -2507,6 +2508,7 @@ nodes: let (_, assignments) = h.provision("gate", 2); + assert_eq!(assignments.len(), 2); for (id, _) in &assignments { assert!(id.starts_with("graph_agent_rescue-agent_"), "{id}"); } diff --git a/src/graph/mod.rs b/src/graph/mod.rs index 1d7ae9e..d5cc6e8 100644 --- a/src/graph/mod.rs +++ b/src/graph/mod.rs @@ -35,13 +35,24 @@ pub(crate) fn wall_clock(secs: u64) -> Option { } /// Whether an error looks like a transient transport/provider failure worth -/// retrying. Providers and HTTP stacks surface these as rendered text rather -/// than typed errors, so the whole context chain is substring-matched. +/// retrying. Typed transport errors anywhere in the chain (reqwest +/// connect/timeout failures and tokio's `Elapsed`) are recognized first; +/// providers that surface failures only as rendered text fall back to +/// substring matching over the whole context chain. pub(crate) fn is_transient_error(err: &anyhow::Error) -> bool { + if err.chain().any(|c| { + c.downcast_ref::() + .is_some_and(|e| e.is_connect() || e.is_timeout()) + || c.is::() + }) { + return true; + } let s = format!("{err:#}"); s.contains("timed out") || s.contains("rate limit") - || s.contains("429") + || s.contains("HTTP 429") + || s.contains("status 429") + || s.contains("429 Too Many") || s.contains("Connection reset") || s.contains("Connection refused") || s.contains("produced no output") @@ -82,7 +93,9 @@ mod tests { fn is_transient_error_matches_expected_signatures() { assert!(is_transient_error(&anyhow!("request timed out after 30s"))); assert!(is_transient_error(&anyhow!("rate limit reached"))); - assert!(is_transient_error(&anyhow!("429 too many requests"))); + assert!(is_transient_error(&anyhow!("HTTP 429"))); + assert!(is_transient_error(&anyhow!("status 429"))); + assert!(is_transient_error(&anyhow!("429 Too Many Requests"))); assert!(is_transient_error(&anyhow!("Connection reset by peer"))); assert!(is_transient_error(&anyhow!("Connection refused"))); assert!(is_transient_error(&anyhow!("llm produced no output"))); @@ -113,5 +126,37 @@ mod tests { ))); assert!(!is_transient_error(&anyhow!("hit max_iterations"))); assert!(!is_transient_error(&anyhow!("authentication failed"))); + assert!(!is_transient_error(&anyhow!( + "script exited 1: see line 429 in a comment" + ))); + } + + #[tokio::test(start_paused = true)] + async fn is_transient_error_recognizes_typed_elapsed_without_matching_text() { + let elapsed = tokio::time::timeout(Duration::from_secs(1), std::future::pending::<()>()) + .await + .expect_err("pending future must time out"); + let err = anyhow::Error::new(elapsed).context("calling provider"); + + // "deadline has elapsed" contains no recognized substring, so only + // the typed pre-check can classify this as transient. + let rendered = format!("{err:#}"); + assert!(!rendered.contains("timed out"), "{rendered}"); + assert!(is_transient_error(&err)); + } + + #[tokio::test] + async fn is_transient_error_recognizes_typed_reqwest_connect_error() { + // `no_proxy` keeps environments with HTTP(S)_PROXY set from turning + // the refused connection into a proxied response. + let client = reqwest::Client::builder().no_proxy().build().unwrap(); + let err = client + .get("http://127.0.0.1:1") + .send() + .await + .expect_err("closed local port must refuse the connection"); + + assert!(err.is_connect()); + assert!(is_transient_error(&anyhow::Error::new(err))); } } diff --git a/src/graph/validator.rs b/src/graph/validator.rs index 0fb3fde..297c406 100644 --- a/src/graph/validator.rs +++ b/src/graph/validator.rs @@ -176,6 +176,8 @@ impl GraphValidator { self.validate_rag_nodes(graph, &mut result); self.validate_llm_nodes(graph, &mut result); self.validate_llm_skills(graph, &mut result); + self.validate_max_attempts(graph, &mut result); + self.validate_fallback_capture(graph, &mut result); self.validate_max_concurrency(graph, &mut result); self.validate_max_concurrency_template(graph, &mut result); self.validate_orchestration_limits(graph, &mut result); @@ -374,6 +376,54 @@ impl GraphValidator { } } + fn validate_max_attempts(&self, graph: &Graph, result: &mut ValidationResult) { + for (node_id, node) in &graph.nodes { + let kind = match &node.node_type { + NodeType::Llm(l) if l.max_attempts == 0 => "llm", + NodeType::Agent(a) if a.max_attempts == 0 => "agent", + _ => continue, + }; + result.error(ValidationError::with_node( + node_id, + format!( + "{kind} node's `max_attempts` must be >= 1 (got 0); the node \ + would exhaust retries without ever attempting a run" + ), + )); + } + } + + fn validate_fallback_capture(&self, graph: &Graph, result: &mut ValidationResult) { + for (node_id, node) in &graph.nodes { + let kind = match &node.node_type { + NodeType::Llm(l) + if l.fallback.is_some() + && l.output_schema.is_some() + && l.state_updates.is_none() => + { + "llm" + } + NodeType::Agent(a) + if a.fallback.is_some() + && a.output_schema.is_some() + && a.state_updates.is_none() => + { + "agent" + } + _ => continue, + }; + result.warning(ValidationError::with_node( + node_id, + format!( + "{kind} node declares `fallback` with `output_schema` but no \ + `state_updates`; on failure the \"…failed: \" string is \ + only written through `state_updates`, so the fallback node \ + cannot see why it was reached" + ), + )); + } + } + fn validate_node_references(&self, graph: &Graph, result: &mut ValidationResult) { for (node_id, node) in &graph.nodes { for (target, label) in declared_targets(node) { @@ -4418,4 +4468,175 @@ mod tests { assert_eq!(ids(&branch_subgraph(&graph, "a")), vec!["a"]); } + + #[test] + fn llm_max_attempts_zero_is_an_error() { + let mut node = llm_node("l", None, Some("end")); + if let NodeType::Llm(ref mut n) = node.node_type { + n.max_attempts = 0; + } + let graph = graph_with(vec![("l", node), ("end", end_node("end"))], "l"); + + let result = validator().validate(&graph); + + assert!( + result + .errors + .iter() + .any(|e| e.node_id.as_deref() == Some("l") + && e.message.contains("`max_attempts` must be >= 1")), + "expected max_attempts error: {:?}", + result.errors + ); + } + + #[test] + fn agent_max_attempts_zero_is_an_error() { + let mut node = agent_node("a", "helper", Some("end")); + if let NodeType::Agent(ref mut n) = node.node_type { + n.max_attempts = 0; + } + let graph = graph_with(vec![("a", node), ("end", end_node("end"))], "a"); + + let result = validator().validate(&graph); + + assert!( + result + .errors + .iter() + .any(|e| e.node_id.as_deref() == Some("a") + && e.message.contains("`max_attempts` must be >= 1")), + "expected max_attempts error: {:?}", + result.errors + ); + } + + #[test] + fn max_attempts_of_one_is_not_flagged() { + let graph = graph_with( + vec![ + ("l", llm_node("l", None, Some("end"))), + ("end", end_node("end")), + ], + "l", + ); + + let result = validator().validate(&graph); + + assert!( + !result + .errors + .iter() + .any(|e| e.message.contains("`max_attempts`")), + "unexpected max_attempts error: {:?}", + result.errors + ); + } + + #[test] + fn llm_fallback_with_output_schema_and_no_state_updates_warns() { + let mut node = llm_with_output_schema("l", &["summary"], Some("end")); + if let NodeType::Llm(ref mut n) = node.node_type { + n.fallback = Some("fb".into()); + } + let graph = graph_with( + vec![ + ("l", node), + ("fb", end_node("fb")), + ("end", end_node("end")), + ], + "l", + ); + + let result = validator().validate(&graph); + + assert!( + result + .warnings + .iter() + .any(|e| e.node_id.as_deref() == Some("l") + && e.message + .contains("declares `fallback` with `output_schema`")), + "expected fallback capture warning: {:?}", + result.warnings + ); + } + + #[test] + fn agent_fallback_with_output_schema_and_no_state_updates_warns() { + let mut node = agent_node("a", "helper", Some("end")); + if let NodeType::Agent(ref mut n) = node.node_type { + n.fallback = Some("fb".into()); + n.output_schema = Some(serde_json::json!({ "type": "object" })); + } + let graph = graph_with( + vec![ + ("a", node), + ("fb", end_node("fb")), + ("end", end_node("end")), + ], + "a", + ); + + let result = validator().validate(&graph); + + assert!( + result + .warnings + .iter() + .any(|e| e.node_id.as_deref() == Some("a") + && e.message + .contains("declares `fallback` with `output_schema`")), + "expected fallback capture warning: {:?}", + result.warnings + ); + } + + #[test] + fn fallback_with_output_schema_and_state_updates_is_not_flagged() { + let mut node = llm_with_output_schema("l", &["summary"], Some("end")); + if let NodeType::Llm(ref mut n) = node.node_type { + n.fallback = Some("fb".into()); + n.state_updates = Some(HashMap::from([("summary".into(), "{{output}}".into())])); + } + let graph = graph_with( + vec![ + ("l", node), + ("fb", end_node("fb")), + ("end", end_node("end")), + ], + "l", + ); + + let result = validator().validate(&graph); + + assert!( + !result.warnings.iter().any(|e| e + .message + .contains("declares `fallback` with `output_schema`")), + "unexpected fallback capture warning: {:?}", + result.warnings + ); + } + + #[test] + fn output_schema_without_fallback_is_not_flagged() { + let graph = graph_with( + vec![ + ("l", llm_with_output_schema("l", &["summary"], Some("end"))), + ("end", end_node("end")), + ], + "l", + ); + + let result = validator().validate(&graph); + + assert!( + !result.warnings.iter().any(|e| e + .message + .contains("declares `fallback` with `output_schema`")), + "unexpected fallback capture warning: {:?}", + result.warnings + ); + } } diff --git a/src/main.rs b/src/main.rs index 091a822..31f81d8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,6 +13,8 @@ mod mcp; mod parsers; mod sandbox; mod supervisor; +#[cfg(test)] +mod testing; mod vault; #[macro_use] diff --git a/src/testing.rs b/src/testing.rs new file mode 100644 index 0000000..6ff02ea --- /dev/null +++ b/src/testing.rs @@ -0,0 +1,43 @@ +//! Test-only helpers shared across modules' unit tests. + +use log::{Level, LevelFilter, Log, Metadata, Record}; +use std::sync::{Mutex, Once, OnceLock}; + +struct WarnCollector; + +static WARN_MESSAGES: OnceLock>> = OnceLock::new(); + +/// Every warn-level message captured since the collector was installed. The +/// buffer is process-global and shared by all tests, so assert by filtering +/// for a marker unique to the test rather than on the whole buffer. +pub(crate) fn warn_messages() -> &'static Mutex> { + WARN_MESSAGES.get_or_init(Mutex::default) +} + +impl Log for WarnCollector { + fn enabled(&self, metadata: &Metadata) -> bool { + metadata.level() <= Level::Warn + } + + fn log(&self, record: &Record) { + if self.enabled(record.metadata()) { + warn_messages() + .lock() + .unwrap() + .push(record.args().to_string()); + } + } + + fn flush(&self) {} +} + +/// Installs the process-wide warn collector. `log::set_logger` accepts one +/// logger per process, so every test that captures warns must install through +/// this shared entry point. +pub(crate) fn install_warn_collector() { + static INSTALL: Once = Once::new(); + INSTALL.call_once(|| { + log::set_logger(&WarnCollector).expect("no other logger should be installed"); + log::set_max_level(LevelFilter::Warn); + }); +}