Dark-Alex-17/coyote
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(TASK-016): engine + test fixes from the Phase-1 re-review

Browse Source 
- 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
This commit is contained in:
Dark-Alex-17 committed 2026-09-14 10:50:33 -06:00
1 parent 18875a4cc9
commit 80ace77e44
9 files changed
+519 -70
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/config/request_context.rs
+24 -6
View File
@@ -8200,6 +8200,20 @@ mod tests {
.collect()
}
fn fallback_capture(node_ids: &[&str]) -> Vec<String> {
node_ids
.iter()
.map(|id| {
format!(
"{id}: llm node declares `fallback` with `output_schema` but no \
`state_updates`; on failure the \"…failed: <chain>\" 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,
src/config/tool_scope.rs
+1 -35
View File
@@ -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<Mutex<Vec<String>>> = OnceLock::new();
fn warn_messages() -> &'static Mutex<Vec<String>> {
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() {
src/graph/agent.rs
+172 -21
View File
@@ -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<AgentExecutionOutcome> {
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: <chain>"` (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: <chain>"` 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<String>,
@@ -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<String> {
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<Output = Result<String>> + 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, &registry, run_attempt).await;
let result = retry_transient(
node_id,
node,
parent_ctx,
&assignment,
&registry,
run_attempt,
)
.await;
if retire_peer_on_return && let (Some(registry), Some((id, _))) = (&registry, &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<Inbox>)>,
@@ -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::<String, _>(
anyhow!("connection error: warn-capture reset")
.context("Agent 'test_agent' failed"),
)
})
}),
)
.await
.expect_err("all attempts fail");
assert_eq!(attempts, 3);
let warns: Vec<String> = 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<String, String> {
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()));
src/graph/executor.rs
+2 -1
View File
@@ -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)?
src/graph/map.rs
+5 -3
View File
@@ -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}");
}
src/graph/mod.rs
+49 -4
View File
@@ -35,13 +35,24 @@ pub(crate) fn wall_clock(secs: u64) -> Option<Duration> {
}
/// 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::<reqwest::Error>()
.is_some_and(|e| e.is_connect() || e.is_timeout())
|| c.is::<tokio::time::error::Elapsed>()
}) {
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)));
}
}
src/graph/validator.rs
+221
View File
@@ -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: <chain>\" 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
);
}
}
src/main.rs
+2
View File
@@ -13,6 +13,8 @@ mod mcp;
mod parsers;
mod sandbox;
mod supervisor;
#[cfg(test)]
mod testing;
mod vault;
#[macro_use]
src/testing.rs
+43
View File
@@ -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<Mutex<Vec<String>>> = 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<Vec<String>> {
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);
});
}