From 1c68331738978544ddb11d6bbc2155450155b016 Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Sat, 12 Sep 2026 11:44:38 -0600 Subject: [PATCH 01/45] feat(TASK-008): add shared transient matcher and retry support to agent nodes Hoist the llm node's transient-error matcher to graph::is_transient_error, extend it with broken-pipe / connection-error / error-sending-request signatures, and give AgentNode an opt-in max_attempts retry loop with llm-parity semantics: contexts applied before the transient check, fresh wall-clock budget per attempt, peer identity (assignment + registry) re-armed before each retry, retirement exactly once after the final attempt, and structured extraction outside the loop. --- Cargo.toml | 1 + src/graph/agent.rs | 399 ++++++++++++++++++++++++++++++++++++++--- src/graph/executor.rs | 1 + src/graph/llm.rs | 34 +--- src/graph/map.rs | 1 + src/graph/mod.rs | 54 ++++++ src/graph/types.rs | 42 ++++- src/graph/validator.rs | 1 + 8 files changed, 474 insertions(+), 59 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index f0e05bf..c5368c7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -146,6 +146,7 @@ ctor = "1.0.13" pretty_assertions = "1.4.0" rmcp = { version = "3.1.2", features = ["server"] } serial_test = "3" +tokio = { version = "1.34.0", features = ["test-util"] } [[bin]] name = "coyote" diff --git a/src/graph/agent.rs b/src/graph/agent.rs index 03b9830..85a6a0b 100644 --- a/src/graph/agent.rs +++ b/src/graph/agent.rs @@ -1,3 +1,4 @@ +use super::is_transient_error; use super::state::StateManager; use super::state_updates; use super::structured; @@ -5,10 +6,14 @@ use super::types::AgentNode; use super::wall_clock; use crate::config::RequestContext; use crate::function::agents::run_agent_for_graph; -use anyhow::{Context, Result}; -use log::debug; +use crate::supervisor::mailbox::{Inbox, PeerRegistry}; +use anyhow::{Context, Error, Result, anyhow}; +use log::{debug, warn}; use serde_json::Value; use std::collections::HashMap; +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; use tokio::time::timeout; const DEFAULT_TIMEOUT_SECS: u64 = 300; @@ -34,31 +39,26 @@ impl AgentNodeExecutor { } let secs = node.timeout.unwrap_or(DEFAULT_TIMEOUT_SECS); - let bound = wall_clock(secs); + let agent_name = node.agent.clone(); - // run_agent_for_graph takes the identity off the ctx; keep a handle so - // a frontier peer is retired the moment the agent stops, not after - // extraction. Chains re-arm the same identity for later steps and - // leave retirement to the chain runner. - let peer = parent_ctx.peer_registry.clone().zip( - parent_ctx - .peer_assignment - .as_ref() - .map(|(id, _)| id.clone()), - ); - - let fut = run_agent_for_graph(parent_ctx, &node.agent, &prompt, graph_inputs); - let raw_result = match bound { - Some(d) => timeout(d, fut).await, - None => Ok(fut.await), - }; - if retire_peer_on_return && let Some((registry, id)) = &peer { - registry.mark_finished(id); - } - let raw = raw_result - .with_context(|| format!("Agent '{}' timed out after {}s", node.agent, secs))? - .with_context(|| format!("Agent '{}' failed", node.agent))?; + let mut run_attempt = attempt_runner(move |ctx| { + let agent_name = agent_name.clone(); + let prompt = prompt.clone(); + let graph_inputs = graph_inputs.clone(); + boxed_attempt(async move { + bounded_attempt( + &agent_name, + secs, + run_agent_for_graph(ctx, &agent_name, &prompt, graph_inputs), + ) + .await + }) + }); + let raw = + run_with_retries(node, parent_ctx, retire_peer_on_return, &mut 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 @@ -77,6 +77,106 @@ impl AgentNodeExecutor { } } +/// One boxed attempt against the ctx. Boxed (rather than an opaque +/// `AsyncFnMut` future) because agent nodes run inside `tokio::spawn`ed map +/// branches, where higher-ranked opaque futures trip the `Send` auto-trait +/// solver ("implementation of `Send` is not general enough"). +type AttemptFuture<'a> = Pin> + Send + 'a>>; + +/// The per-attempt runner the retry loop drives. The attempt future may only +/// borrow the ctx it is handed, so runners move owned clones of everything +/// else into the future. +type AttemptRunner<'f> = dyn for<'a> FnMut(&'a mut RequestContext) -> AttemptFuture<'a> + Send + 'f; + +/// Identity funnel that pins a closure to the runner's higher-ranked +/// signature, so inline closures at call sites infer the right lifetimes. +fn attempt_runner(f: F) -> F +where + F: for<'a> FnMut(&'a mut RequestContext) -> AttemptFuture<'a> + Send, +{ + f +} + +fn boxed_attempt<'a>(fut: impl Future> + Send + 'a) -> AttemptFuture<'a> { + Box::pin(fut) +} + +/// Runs the node's agent up to `node.max_attempts` times, retrying only +/// failures that `is_transient_error` recognizes. +/// +/// `run_agent_for_graph` takes the peer identity off the ctx and never +/// restores it, so BOTH halves — the (id, inbox) assignment and the registry +/// — are captured up front and re-armed before every retry; re-arming only +/// the assignment would give a retried attempt its identity back but no +/// roster and a broken `agent__send_message`. A frontier peer is retired +/// exactly once, after the final attempt, the moment the agent stops — not +/// after extraction. Chains re-arm the same identity for later steps and +/// leave retirement to the chain runner. +async fn run_with_retries( + node: &AgentNode, + parent_ctx: &mut RequestContext, + retire_peer_on_return: bool, + run_attempt: &mut AttemptRunner<'_>, +) -> Result { + 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; + + if retire_peer_on_return && let (Some(registry), Some((id, _))) = (®istry, &assignment) { + registry.mark_finished(id); + } + + result +} + +async fn retry_transient( + node: &AgentNode, + parent_ctx: &mut RequestContext, + assignment: &Option<(String, Arc)>, + registry: &Option>, + run_attempt: &mut AttemptRunner<'_>, +) -> Result { + let mut last_err: Option = None; + for attempt in 1..=node.max_attempts { + if attempt > 1 { + if let Some(assignment) = assignment { + parent_ctx.peer_assignment = Some(assignment.clone()); + } + if let Some(registry) = registry { + parent_ctx.peer_registry = Some(Arc::clone(registry)); + } + } + 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"); + last_err = Some(e); + } + Err(e) => return Err(e), + } + } + Err(last_err.unwrap_or_else(|| anyhow!("agent node exhausted retries"))) +} + +/// Applies the wall-clock bound and the human-readable failure contexts to a +/// single attempt. The contexts land here — before the caller inspects the +/// error — because `tokio::time::error::Elapsed` renders as "deadline has +/// elapsed", which the transient matcher would not recognize as a timeout. +async fn bounded_attempt( + agent_name: &str, + secs: u64, + fut: impl Future>, +) -> Result { + let raw_result = match wall_clock(secs) { + Some(d) => timeout(d, fut).await, + None => Ok(fut.await), + }; + raw_result + .with_context(|| format!("Agent '{agent_name}' timed out after {secs}s"))? + .with_context(|| format!("Agent '{agent_name}' failed")) +} + /// Resolves each `inputs` template against the parent state. A lone /// `{{key}}` yields the state value as-is (numbers, arrays, objects and /// `null` all survive); anything else renders to a string. Every key is @@ -135,6 +235,7 @@ mod tests { state_updates: updates, output_schema: None, timeout: None, + max_attempts: 1, inputs: None, teammates: false, } @@ -184,6 +285,254 @@ mod tests { assert!(!registry.is_finished(&id)); } + fn plain_ctx() -> RequestContext { + RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd) + } + + fn retryable_node(max_attempts: u32) -> AgentNode { + let mut node = node_with("hi", None); + node.max_attempts = max_attempts; + node + } + + #[tokio::test] + async fn run_with_retries_retries_transient_failure_and_succeeds() { + let mut ctx = plain_ctx(); + let node = retryable_node(2); + let mut attempts = 0u32; + + let out = run_with_retries( + &node, + &mut ctx, + false, + &mut attempt_runner(|_ctx| { + attempts += 1; + let result = if attempts == 1 { + Err(anyhow!("error sending request for url")) + } else { + Ok("recovered".to_string()) + }; + boxed_attempt(async move { result }) + }), + ) + .await + .unwrap(); + + assert_eq!(out, "recovered"); + assert_eq!(attempts, 2); + } + + #[tokio::test] + async fn run_with_retries_does_not_retry_non_transient_failures() { + let mut ctx = plain_ctx(); + let node = retryable_node(3); + let mut attempts = 0u32; + + let err = run_with_retries( + &node, + &mut ctx, + false, + &mut attempt_runner(|_ctx| { + attempts += 1; + boxed_attempt(async move { Err::(anyhow!("Unknown model 'foo'")) }) + }), + ) + .await + .expect_err("non-transient failure must propagate immediately"); + + assert_eq!(attempts, 1); + assert!(format!("{err:#}").contains("Unknown model 'foo'")); + } + + #[tokio::test] + async fn run_with_retries_stops_at_max_attempts_and_keeps_failure_context() { + let mut ctx = plain_ctx(); + let node = retryable_node(2); + let mut attempts = 0u32; + + let err = run_with_retries( + &node, + &mut ctx, + false, + &mut attempt_runner(|_ctx| { + attempts += 1; + boxed_attempt(bounded_attempt("test_agent", 0, async { + Err(anyhow!("connection error: unexpected end of stream")) + })) + }), + ) + .await + .expect_err("exhausted retries must propagate the last error"); + + assert_eq!(attempts, 2); + let chain = format!("{err:#}"); + assert!(chain.contains("Agent 'test_agent' failed"), "{chain}"); + assert!(chain.contains("connection error"), "{chain}"); + } + + #[tokio::test(start_paused = true)] + async fn run_with_retries_retries_a_timeout_and_succeeds() { + let mut ctx = plain_ctx(); + let node = retryable_node(2); + let mut attempts = 0u32; + + let out = run_with_retries( + &node, + &mut ctx, + false, + &mut attempt_runner(|_ctx| { + attempts += 1; + if attempts == 1 { + boxed_attempt(bounded_attempt( + "test_agent", + 1, + std::future::pending::>(), + )) + } else { + boxed_attempt(async move { Ok("recovered".to_string()) }) + } + }), + ) + .await + .unwrap(); + + assert_eq!(out, "recovered"); + assert_eq!(attempts, 2); + } + + #[tokio::test(start_paused = true)] + async fn bounded_attempt_timeout_context_is_transient() { + let err = bounded_attempt("test_agent", 1, std::future::pending::>()) + .await + .expect_err("pending future must hit the wall clock"); + + let chain = format!("{err:#}"); + assert!( + chain.contains("Agent 'test_agent' timed out after 1s"), + "{chain}" + ); + assert!(is_transient_error(&err)); + } + + #[tokio::test] + async fn run_with_retries_rearms_the_same_peer_identity_for_each_retry() { + let (mut ctx, registry, id) = ctx_at_max_depth_with_peer(); + let expected_inbox = Arc::clone(&ctx.peer_assignment.as_ref().unwrap().1); + let node = retryable_node(2); + let mut attempts = 0u32; + let mut observations: Vec<(String, bool, bool)> = Vec::new(); + + let out = run_with_retries( + &node, + &mut ctx, + true, + &mut attempt_runner(|ctx| { + attempts += 1; + // Mirror run_agent_for_graph: consume both identity halves. + let (peer_id, inbox) = ctx.peer_assignment.take().expect("assignment armed"); + let reg = ctx.peer_registry.take().expect("registry armed"); + observations.push(( + peer_id, + Arc::ptr_eq(&inbox, &expected_inbox), + Arc::ptr_eq(®, ®istry), + )); + assert!( + !registry.is_finished(&id), + "peer must not be retired between attempts" + ); + let result = if attempts == 1 { + Err(anyhow!("stream closed because of a broken pipe")) + } else { + Ok("done".to_string()) + }; + boxed_attempt(async move { result }) + }), + ) + .await + .unwrap(); + + assert_eq!(out, "done"); + assert_eq!(observations.len(), 2); + for (peer_id, same_inbox, same_registry) in &observations { + assert_eq!(peer_id, &id); + assert!(same_inbox, "retry must reuse the same inbox Arc"); + assert!(same_registry, "retry must reuse the same registry Arc"); + } + assert!(registry.is_finished(&id)); + } + + #[tokio::test] + async fn run_with_retries_retires_frontier_peer_after_the_final_failed_attempt() { + let (mut ctx, registry, id) = ctx_at_max_depth_with_peer(); + let node = retryable_node(2); + let mut attempts = 0u32; + + let err = run_with_retries( + &node, + &mut ctx, + true, + &mut attempt_runner(|ctx| { + attempts += 1; + ctx.peer_assignment.take(); + ctx.peer_registry.take(); + assert!( + !registry.is_finished(&id), + "peer must not be retired between attempts" + ); + boxed_attempt(async move { Err::(anyhow!("Connection reset by peer")) }) + }), + ) + .await + .expect_err("both attempts fail"); + + assert_eq!(attempts, 2); + assert!(registry.is_finished(&id)); + assert!(format!("{err:#}").contains("Connection reset")); + } + + #[tokio::test] + async fn run_with_retries_leaves_peer_live_when_not_retiring() { + let (mut ctx, registry, id) = ctx_at_max_depth_with_peer(); + let node = retryable_node(1); + + run_with_retries( + &node, + &mut ctx, + false, + &mut attempt_runner(|ctx| { + ctx.peer_assignment.take(); + ctx.peer_registry.take(); + boxed_attempt(async move { Ok("done".to_string()) }) + }), + ) + .await + .unwrap(); + + assert!(!registry.is_finished(&id)); + } + + #[tokio::test] + async fn run_with_retries_zero_attempts_reports_exhausted_retries() { + let mut ctx = plain_ctx(); + let node = retryable_node(0); + let mut attempts = 0u32; + + let err = run_with_retries( + &node, + &mut ctx, + false, + &mut attempt_runner(|_ctx| { + attempts += 1; + boxed_attempt(async move { Ok("never".to_string()) }) + }), + ) + .await + .expect_err("zero attempts cannot succeed"); + + assert_eq!(attempts, 0); + assert!(format!("{err:#}").contains("agent node exhausted retries")); + } + fn node_with_inputs(pairs: &[(&str, &str)]) -> AgentNode { let mut node = node_with("hi", None); node.inputs = Some( diff --git a/src/graph/executor.rs b/src/graph/executor.rs index 127aa95..1367c88 100644 --- a/src/graph/executor.rs +++ b/src/graph/executor.rs @@ -827,6 +827,7 @@ mod tests { state_updates: None, output_schema: None, timeout: None, + max_attempts: 1, inputs: None, teammates, }), diff --git a/src/graph/llm.rs b/src/graph/llm.rs index a409469..7623c27 100644 --- a/src/graph/llm.rs +++ b/src/graph/llm.rs @@ -2,7 +2,7 @@ use super::state::StateManager; use super::state_updates; use super::structured; use super::types::LlmNode; -use super::wall_clock; +use super::{is_transient_error, wall_clock}; use crate::client::{Model, ModelType, call_chat_completions}; use crate::config::prompts::DEFAULT_SKILL_INSTRUCTIONS; use crate::config::{ @@ -250,7 +250,7 @@ async fn run_with_retries( for attempt in 1..=node.max_attempts { match run_chat_loop(node, prompt, ctx, abort).await { Ok(out) => return Ok(out), - Err(e) if is_transient(&e) && attempt < node.max_attempts => { + Err(e) if is_transient_error(&e) && attempt < node.max_attempts => { warn!("llm node attempt {attempt} failed (transient): {e}; retrying"); last_err = Some(e); } @@ -449,16 +449,6 @@ fn validate_tools_subset( Ok(()) } -fn is_transient(err: &Error) -> bool { - let s = format!("{err:#}"); - s.contains("timed out") - || s.contains("rate limit") - || s.contains("429") - || s.contains("Connection reset") - || s.contains("Connection refused") - || s.contains("produced no output") -} - fn apply_state_updates_with_output( node: &LlmNode, state_manager: &mut StateManager, @@ -724,26 +714,6 @@ mod tests { assert!(mcp.is_empty()); } - #[test] - fn is_transient_matches_expected_signatures() { - assert!(is_transient(&anyhow!("request timed out after 30s"))); - assert!(is_transient(&anyhow!("rate limit reached"))); - assert!(is_transient(&anyhow!("429 too many requests"))); - assert!(is_transient(&anyhow!("Connection reset by peer"))); - assert!(is_transient(&anyhow!("Connection refused"))); - assert!(is_transient(&anyhow!("llm produced no output"))); - } - - #[test] - fn is_transient_rejects_non_transient_errors() { - assert!(!is_transient(&anyhow!("Unknown model 'foo'"))); - assert!(!is_transient(&anyhow!( - "llm node references unknown tool 'bad'" - ))); - assert!(!is_transient(&anyhow!("hit max_iterations"))); - assert!(!is_transient(&anyhow!("authentication failed"))); - } - #[test] fn zero_timeout_resolves_to_no_bound() { assert!(Some(0u64).and_then(wall_clock).is_none()); diff --git a/src/graph/map.rs b/src/graph/map.rs index 29ad7de..6d74543 100644 --- a/src/graph/map.rs +++ b/src/graph/map.rs @@ -464,6 +464,7 @@ mod tests { state_updates: None, output_schema: None, timeout: None, + max_attempts: 1, inputs: None, teammates, }) diff --git a/src/graph/mod.rs b/src/graph/mod.rs index 3b1d72d..1d7ae9e 100644 --- a/src/graph/mod.rs +++ b/src/graph/mod.rs @@ -34,6 +34,22 @@ pub(crate) fn wall_clock(secs: u64) -> Option { (secs != 0).then(|| Duration::from_secs(secs)) } +/// 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. +pub(crate) fn is_transient_error(err: &anyhow::Error) -> bool { + let s = format!("{err:#}"); + s.contains("timed out") + || s.contains("rate limit") + || s.contains("429") + || s.contains("Connection reset") + || s.contains("Connection refused") + || s.contains("produced no output") + || s.contains("broken pipe") + || s.contains("connection error") + || s.contains("error sending request") +} + pub const MAX_STATE_SIZE_BYTES: usize = 32 * 1024; pub(in crate::graph) fn type_name(value: &Value) -> &'static str { @@ -50,6 +66,7 @@ pub(in crate::graph) fn type_name(value: &Value) -> &'static str { #[cfg(test)] mod tests { use super::*; + use anyhow::anyhow; #[test] fn wall_clock_zero_is_no_bound() { @@ -60,4 +77,41 @@ mod tests { fn wall_clock_nonzero_is_that_many_seconds() { assert_eq!(wall_clock(7), Some(Duration::from_secs(7))); } + + #[test] + 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!("Connection reset by peer"))); + assert!(is_transient_error(&anyhow!("Connection refused"))); + assert!(is_transient_error(&anyhow!("llm produced no output"))); + assert!(is_transient_error(&anyhow!( + "stream closed because of a broken pipe" + ))); + assert!(is_transient_error(&anyhow!( + "connection error: unexpected end of stream" + ))); + assert!(is_transient_error(&anyhow!( + "error sending request for url" + ))); + } + + #[test] + fn is_transient_error_sees_through_context_chains() { + let err = anyhow!("stream closed because of a broken pipe") + .context("Failed to call chat-completions api") + .context("Agent 'domain-reviewer' failed"); + assert!(is_transient_error(&err)); + } + + #[test] + fn is_transient_error_rejects_non_transient_errors() { + assert!(!is_transient_error(&anyhow!("Unknown model 'foo'"))); + assert!(!is_transient_error(&anyhow!( + "llm node references unknown tool 'bad'" + ))); + assert!(!is_transient_error(&anyhow!("hit max_iterations"))); + assert!(!is_transient_error(&anyhow!("authentication failed"))); + } } diff --git a/src/graph/types.rs b/src/graph/types.rs index 6e00d04..b69d442 100644 --- a/src/graph/types.rs +++ b/src/graph/types.rs @@ -240,6 +240,9 @@ pub struct AgentNode { #[serde(default, skip_serializing_if = "Option::is_none")] pub timeout: Option, + #[serde(default = "default_max_attempts")] + pub max_attempts: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] pub inputs: Option>, @@ -321,7 +324,7 @@ pub struct LlmNode { #[serde(default, skip_serializing_if = "Option::is_none")] pub fallback: Option, - #[serde(default = "default_llm_max_attempts")] + #[serde(default = "default_max_attempts")] pub max_attempts: u32, #[serde(default = "default_llm_max_iterations")] @@ -349,7 +352,7 @@ pub struct LlmNode { pub skill_instructions: Option, } -fn default_llm_max_attempts() -> u32 { +fn default_max_attempts() -> u32 { 1 } @@ -640,6 +643,41 @@ nodes: } } + #[test] + fn agent_node_max_attempts_parses_and_defaults_to_one() { + let yaml = r#" +name: g +start: a +nodes: + a: + id: a + type: agent + agent: helper + prompt: hi + max_attempts: 2 + next: b + b: + id: b + type: agent + agent: helper + prompt: hi + next: e + e: + id: e + type: end + output: done +"#; + let graph: Graph = serde_yaml::from_str(yaml).unwrap(); + match &graph.get_node("a").unwrap().node_type { + NodeType::Agent(n) => assert_eq!(n.max_attempts, 2), + _ => panic!("expected Agent variant"), + } + match &graph.get_node("b").unwrap().node_type { + NodeType::Agent(n) => assert_eq!(n.max_attempts, 1), + _ => panic!("expected Agent variant"), + } + } + #[test] fn agent_node_inputs_parse_optionally_and_empty_map_is_some() { let yaml = r#" diff --git a/src/graph/validator.rs b/src/graph/validator.rs index 08188fe..c190b1f 100644 --- a/src/graph/validator.rs +++ b/src/graph/validator.rs @@ -1924,6 +1924,7 @@ mod tests { state_updates: None, output_schema: None, timeout: None, + max_attempts: 1, inputs: None, teammates: false, }), From be0c5462edeeed624aa5814bb0b4df2651f5993a Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Sat, 12 Sep 2026 12:03:50 -0600 Subject: [PATCH 02/45] feat(TASK-009): add fallback routing to agent nodes Agent nodes gain an optional fallback target with llm-node parity: on any failure (retries exhausted, hard error, or structured-extraction failure) the failure chain is written into state as 'Agent node failed: ' and the chain routes to the fallback node. Without a fallback the error propagates unchanged. Validator collects agent fallback edges for missing-target checks, reachability, and map-branch locality. --- src/graph/agent.rs | 253 +++++++++++++++++++++++++++++++++-------- src/graph/executor.rs | 13 ++- src/graph/map.rs | 50 ++++++++ src/graph/types.rs | 38 +++++++ src/graph/validator.rs | 119 ++++++++++++++++++- 5 files changed, 416 insertions(+), 57 deletions(-) diff --git a/src/graph/agent.rs b/src/graph/agent.rs index 85a6a0b..974d8b8 100644 --- a/src/graph/agent.rs +++ b/src/graph/agent.rs @@ -18,65 +18,111 @@ use tokio::time::timeout; const DEFAULT_TIMEOUT_SECS: u64 = 300; +#[derive(Debug, PartialEq, Eq)] +pub(super) enum AgentExecutionOutcome { + Continue(String), + FellBack(String), +} + pub struct AgentNodeExecutor; impl AgentNodeExecutor { - pub async fn execute( + pub(super) async fn execute( node: &AgentNode, state_manager: &mut StateManager, parent_ctx: &mut RequestContext, retire_peer_on_return: bool, - ) -> Result { - let prompt = state_manager - .interpolate(&node.prompt) - .with_context(|| format!("Failed to interpolate prompt for agent '{}'", node.agent))?; - - let graph_inputs = resolve_inputs(node, state_manager)?; - if let Some(inputs) = &graph_inputs { - let mut keys: Vec<&String> = inputs.keys().collect(); - keys.sort_unstable(); - debug!("Agent '{}' graph inputs: {keys:?}", node.agent); - } - - let secs = node.timeout.unwrap_or(DEFAULT_TIMEOUT_SECS); - let agent_name = node.agent.clone(); - - let mut run_attempt = attempt_runner(move |ctx| { - let agent_name = agent_name.clone(); - let prompt = prompt.clone(); - let graph_inputs = graph_inputs.clone(); - boxed_attempt(async move { - bounded_attempt( - &agent_name, - secs, - run_agent_for_graph(ctx, &agent_name, &prompt, graph_inputs), - ) - .await - }) - }); - let raw = - run_with_retries(node, parent_ctx, retire_peer_on_return, &mut 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 - .with_context(|| { - format!( - "Agent '{}' output failed structured-output extraction", - node.agent - ) - })?, - None => Value::String(raw.clone()), - }; - - apply_state_updates(node, state_manager, &output_value); - - Ok(raw) + ) -> Result { + let result = run(node, state_manager, parent_ctx, retire_peer_on_return).await; + outcome_from(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. +fn outcome_from( + node: &AgentNode, + state_manager: &mut StateManager, + result: Result, +) -> Result { + match result { + Ok(raw) => Ok(AgentExecutionOutcome::Continue(raw)), + Err(e) => match &node.fallback { + Some(fb) => { + warn!("agent node failed, routing to fallback '{fb}': {e:#}"); + state_updates::apply( + state_manager, + &Value::String(format!("Agent node failed: {e:#}")), + node.output_schema.is_some(), + node.state_updates.as_ref(), + ); + Ok(AgentExecutionOutcome::FellBack(fb.clone())) + } + None => Err(e), + }, + } +} + +async fn run( + node: &AgentNode, + state_manager: &mut StateManager, + parent_ctx: &mut RequestContext, + retire_peer_on_return: bool, +) -> Result { + let prompt = state_manager + .interpolate(&node.prompt) + .with_context(|| format!("Failed to interpolate prompt for agent '{}'", node.agent))?; + + let graph_inputs = resolve_inputs(node, state_manager)?; + if let Some(inputs) = &graph_inputs { + let mut keys: Vec<&String> = inputs.keys().collect(); + keys.sort_unstable(); + debug!("Agent '{}' graph inputs: {keys:?}", node.agent); + } + + let secs = node.timeout.unwrap_or(DEFAULT_TIMEOUT_SECS); + let agent_name = node.agent.clone(); + + let mut run_attempt = attempt_runner(move |ctx| { + let agent_name = agent_name.clone(); + let prompt = prompt.clone(); + let graph_inputs = graph_inputs.clone(); + boxed_attempt(async move { + bounded_attempt( + &agent_name, + secs, + run_agent_for_graph(ctx, &agent_name, &prompt, graph_inputs), + ) + .await + }) + }); + let raw = run_with_retries(node, parent_ctx, retire_peer_on_return, &mut 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 + .with_context(|| { + format!( + "Agent '{}' output failed structured-output extraction", + node.agent + ) + })?, + None => Value::String(raw.clone()), + }; + + apply_state_updates(node, state_manager, &output_value); + + Ok(raw) +} + /// One boxed attempt against the ctx. Boxed (rather than an opaque /// `AsyncFnMut` future) because agent nodes run inside `tokio::spawn`ed map /// branches, where higher-ranked opaque futures trip the `Send` auto-trait @@ -236,6 +282,7 @@ mod tests { output_schema: None, timeout: None, max_attempts: 1, + fallback: None, inputs: None, teammates: false, } @@ -533,6 +580,114 @@ mod tests { assert!(format!("{err:#}").contains("agent node exhausted retries")); } + fn failure_capture_updates() -> HashMap { + let mut u = HashMap::new(); + u.insert("failure".into(), "{{output}}".into()); + u + } + + #[tokio::test] + async fn execute_failure_with_fallback_routes_and_records_the_failure() { + let mut ctx = plain_ctx(); + ctx.current_depth = default_max_agent_depth(); + let mut node = node_with("hi", Some(failure_capture_updates())); + node.fallback = Some("recover".into()); + let mut state = manager_with(&[]); + + let outcome = AgentNodeExecutor::execute(&node, &mut state, &mut ctx, false) + .await + .unwrap(); + + assert_eq!(outcome, AgentExecutionOutcome::FellBack("recover".into())); + let failure = state + .state() + .get("failure") + .and_then(Value::as_str) + .expect("failure text must land in state for the fallback node") + .to_string(); + assert!(failure.starts_with("Agent node failed: "), "{failure}"); + assert!(failure.contains("Agent 'test_agent' failed"), "{failure}"); + assert!(failure.contains("Max agent depth exceeded"), "{failure}"); + } + + fn extraction_error() -> Error { + anyhow!("no JSON object found in output") + .context("Agent 'test_agent' output failed structured-output extraction") + } + + #[test] + fn outcome_from_success_is_continue() { + let node = node_with("hi", None); + let mut state = manager_with(&[]); + + let outcome = outcome_from(&node, &mut state, Ok("raw".into())).unwrap(); + + assert_eq!(outcome, AgentExecutionOutcome::Continue("raw".into())); + } + + #[test] + fn outcome_from_extraction_failure_with_fallback_falls_back() { + let mut node = node_with("hi", Some(failure_capture_updates())); + node.fallback = Some("recover".into()); + let mut state = manager_with(&[]); + + let outcome = outcome_from(&node, &mut state, Err(extraction_error())).unwrap(); + + assert_eq!(outcome, AgentExecutionOutcome::FellBack("recover".into())); + let failure = state + .state() + .get("failure") + .and_then(Value::as_str) + .unwrap(); + assert!( + failure.contains("structured-output extraction"), + "{failure}" + ); + assert!(failure.contains("no JSON object found"), "{failure}"); + } + + #[test] + fn outcome_from_failure_without_fallback_propagates_the_error_unchanged() { + let node = node_with("hi", Some(failure_capture_updates())); + let mut state = manager_with(&[]); + + let err = outcome_from(&node, &mut state, Err(extraction_error())) + .expect_err("no fallback must propagate"); + + assert_eq!( + format!("{err:#}"), + "Agent 'test_agent' output failed structured-output extraction: \ + no JSON object found in output" + ); + assert!(state.state().get("failure").is_none()); + } + + #[tokio::test] + async fn fallback_engages_only_after_transient_retries_exhaust() { + let mut ctx = plain_ctx(); + let mut node = retryable_node(2); + node.fallback = Some("recover".into()); + let mut attempts = 0u32; + + let result = run_with_retries( + &node, + &mut ctx, + false, + &mut attempt_runner(|_ctx| { + attempts += 1; + boxed_attempt(async move { + Err::(anyhow!("connection error: reset by peer")) + }) + }), + ) + .await; + let mut state = manager_with(&[]); + let outcome = outcome_from(&node, &mut state, result).unwrap(); + + assert_eq!(attempts, 2, "fallback must not preempt the retry budget"); + assert_eq!(outcome, AgentExecutionOutcome::FellBack("recover".into())); + } + fn node_with_inputs(pairs: &[(&str, &str)]) -> AgentNode { let mut node = node_with("hi", None); node.inputs = Some( diff --git a/src/graph/executor.rs b/src/graph/executor.rs index 1367c88..805516d 100644 --- a/src/graph/executor.rs +++ b/src/graph/executor.rs @@ -1,4 +1,4 @@ -use super::agent::AgentNodeExecutor; +use super::agent::{AgentExecutionOutcome, AgentNodeExecutor}; use super::llm::{LlmExecutionOutcome, LlmNodeExecutor}; use super::logging::{GraphLogger, narrate_node_complete, narrate_node_failed}; use super::map::MapNodeExecutor; @@ -574,8 +574,14 @@ pub(super) async fn step( ) -> Result { match &node.node_type { NodeType::Agent(agent_node) => { - AgentNodeExecutor::execute(agent_node, state, ctx, !step_ctx.branch_mode).await?; - let targets = static_next_targets(node, current, "agent", step_ctx.branch_mode)?; + let outcome = + AgentNodeExecutor::execute(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)? + } + AgentExecutionOutcome::FellBack(target) => vec![target], + }; Ok(StepResult::Continue(targets)) } NodeType::Script(script_node) => { @@ -828,6 +834,7 @@ mod tests { output_schema: None, timeout: None, max_attempts: 1, + fallback: None, inputs: None, teammates, }), diff --git a/src/graph/map.rs b/src/graph/map.rs index 6d74543..2f982fc 100644 --- a/src/graph/map.rs +++ b/src/graph/map.rs @@ -465,6 +465,7 @@ mod tests { output_schema: None, timeout: None, max_attempts: 1, + fallback: None, inputs: None, teammates, }) @@ -1609,6 +1610,55 @@ nodes: assert_eq!(collected(&result), vec![json!("fb")]); } + #[tokio::test] + async fn map_chain_agent_failure_routes_to_fallback() { + if !cmd_available("python3") { + eprintln!("skipping: python3 not available"); + return; + } + let ws = TestWorkspace::new(); + ws.write_py( + "recover.py", + r#"print(json.dumps({"output": "recovered"}))"#, + ); + + let yaml = r#" +name: chain +start: fan_out +settings: + validate_before_run: false +initial_state: + items: [1, 2] +nodes: + fan_out: + type: map + over: "{{items}}" + as: item + branch: worker + collect_into: results + next: done + worker: + type: agent + agent: no-such-agent + prompt: "hi" + fallback: recover + recover: + type: script + script: recover.py + done: + type: end + output: "{{results}}" +"#; + let result = run_graph(yaml, &ws) + .await + .unwrap_or_else(|e| panic!("executor failed: {e:#}")); + + assert_eq!( + collected(&result), + vec![json!("recovered"), json!("recovered")] + ); + } + struct Harness { ws: TestWorkspace, graph: Arc, diff --git a/src/graph/types.rs b/src/graph/types.rs index b69d442..6a8c02f 100644 --- a/src/graph/types.rs +++ b/src/graph/types.rs @@ -243,6 +243,9 @@ pub struct AgentNode { #[serde(default = "default_max_attempts")] pub max_attempts: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub fallback: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub inputs: Option>, @@ -678,6 +681,41 @@ nodes: } } + #[test] + fn agent_node_fallback_parses_and_defaults_to_none() { + let yaml = r#" +name: g +start: a +nodes: + a: + id: a + type: agent + agent: helper + prompt: hi + fallback: b + next: b + b: + id: b + type: agent + agent: helper + prompt: hi + next: e + e: + id: e + type: end + output: done +"#; + let graph: Graph = serde_yaml::from_str(yaml).unwrap(); + match &graph.get_node("a").unwrap().node_type { + NodeType::Agent(n) => assert_eq!(n.fallback.as_deref(), Some("b")), + _ => panic!("expected Agent variant"), + } + match &graph.get_node("b").unwrap().node_type { + NodeType::Agent(n) => assert!(n.fallback.is_none()), + _ => panic!("expected Agent variant"), + } + } + #[test] fn agent_node_inputs_parse_optionally_and_empty_map_is_some() { let yaml = r#" diff --git a/src/graph/validator.rs b/src/graph/validator.rs index c190b1f..0fb3fde 100644 --- a/src/graph/validator.rs +++ b/src/graph/validator.rs @@ -719,6 +719,7 @@ impl GraphValidator { .filter_map(|(_, n)| match &n.node_type { NodeType::Script(s) => s.fallback.as_deref(), NodeType::Llm(l) => l.fallback.as_deref(), + NodeType::Agent(a) => a.fallback.as_deref(), _ => None, }) .collect(); @@ -982,12 +983,17 @@ fn declared_targets(node: &Node) -> Vec<(String, &'static str)> { out.push((t.clone(), "llm 'fallback'")); } } + NodeType::Agent(a) => { + if let Some(t) = &a.fallback { + out.push((t.clone(), "agent 'fallback'")); + } + } NodeType::Map(m) => { out.push((m.branch.clone(), "map 'branch'")); } - // `agent`/`input`/`rag` route only via `next` (already collected - // above); `end` is terminal. No type-specific routing edges to add. - NodeType::Agent(_) | NodeType::Input(_) | NodeType::Rag(_) | NodeType::End(_) => {} + // `input`/`rag` route only via `next` (already collected above); + // `end` is terminal. No type-specific routing edges to add. + NodeType::Input(_) | NodeType::Rag(_) | NodeType::End(_) => {} } out } @@ -1056,8 +1062,8 @@ fn main_flow_reachable(graph: &Graph) -> HashSet { } /// Nodes reachable from `entry` over the edges a chain can follow at run -/// time: `next` targets and script/llm `fallback`s. Approval routes and a -/// nested map's `branch` are not followed. Neither may appear inside a +/// time: `next` targets and script/llm/agent `fallback`s. Approval routes +/// and a nested map's `branch` are not followed. Neither may appear inside a /// branch, and the validator reports them separately. pub(super) fn branch_subgraph(graph: &Graph, entry: &str) -> HashSet { let mut reachable: HashSet = HashSet::new(); @@ -1082,6 +1088,7 @@ pub(super) fn branch_subgraph(graph: &Graph, entry: &str) -> HashSet { match &node.node_type { NodeType::Script(s) => edges.extend(s.fallback.as_ref()), NodeType::Llm(l) => edges.extend(l.fallback.as_ref()), + NodeType::Agent(a) => edges.extend(a.fallback.as_ref()), _ => {} } for next in edges { @@ -1925,6 +1932,7 @@ mod tests { output_schema: None, timeout: None, max_attempts: 1, + fallback: None, inputs: None, teammates: false, }), @@ -2018,6 +2026,51 @@ mod tests { ); } + #[test] + fn flags_missing_agent_fallback_target() { + let mut a = agent_node("a", "helper", Some("end")); + if let NodeType::Agent(ref mut an) = a.node_type { + an.fallback = Some("ghost".into()); + } + let graph = graph_with(vec![("a", a), ("end", end_node("end"))], "a"); + + let result = validator().validate(&graph); + + assert!(!result.is_valid()); + assert!( + result + .errors + .iter() + .any(|e| e.message.contains("non-existent node 'ghost'") + && e.message.contains("agent 'fallback'")) + ); + } + + #[test] + fn agent_fallback_target_is_not_flagged_unreachable() { + let mut a = agent_node("a", "helper", Some("end")); + if let NodeType::Agent(ref mut an) = a.node_type { + an.fallback = Some("recover".into()); + } + let recover = agent_node("recover", "helper", Some("end")); + let graph = graph_with( + vec![("a", a), ("recover", recover), ("end", end_node("end"))], + "a", + ); + + let result = validator().validate(&graph); + + assert!( + !result + .warnings + .iter() + .any(|w| w.node_id.as_deref() == Some("recover") + && w.message.contains("unreachable")), + "fallback target 'recover' incorrectly marked unreachable: {:?}", + result.warnings + ); + } + #[test] fn detects_two_node_cycle() { let mut a = end_node("a"); @@ -2857,6 +2910,62 @@ mod tests { ); } + #[test] + fn map_branch_agent_fallback_escaping_the_branch_gets_hint() { + let map = map_node_basic("m", "br", Some("end")); + let mut branch = agent_node("br", "helper", None); + if let NodeType::Agent(ref mut an) = branch.node_type { + an.fallback = Some("end".into()); + } + 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("'end'") + && e.message.contains("end node") + && e.message.contains("never honored") + && e.node_id.as_deref() == Some("m")), + "expected end-node error with the fallback hint: {:?}", + result.errors + ); + } + + #[test] + fn map_branch_agent_fallback_to_branch_local_node_passes() { + let map = map_node_basic("m", "br", Some("end")); + let mut branch = agent_node("br", "helper", None); + if let NodeType::Agent(ref mut an) = branch.node_type { + an.fallback = Some("recover".into()); + } + let recover = agent_node("recover", "helper", None); + let graph = graph_with( + vec![ + ("m", map), + ("br", branch), + ("recover", recover), + ("end", end_node("end")), + ], + "m", + ); + + let result = validator().validate(&graph); + + // Agent-existence errors on 'br'/'recover' are expected (no agent + // dirs in the test cwd); branch-locality errors land on the map. + assert!( + !result + .errors + .iter() + .any(|e| e.node_id.as_deref() == Some("m")), + "a branch-local agent fallback is allowed: {:?}", + result.errors + ); + } + #[test] fn map_branch_deep_approval_node_errors_on_the_map() { let map = map_node_basic("m", "br", Some("end")); From 75cc2a4b9cde9d91105563ff135c8b8e2e6bbcc5 Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Sat, 12 Sep 2026 12:14:41 -0600 Subject: [PATCH 03/45] feat(TASK-010): add max_attempts to read-only agent lanes in bundled graphs --- assets/agents/code-reviewer/graph.yaml | 2 ++ assets/agents/deep-research/graph.yaml | 1 + assets/agents/review-gauntlet/graph.yaml | 4 ++++ assets/agents/step-runner/graph.yaml | 1 + 4 files changed, 8 insertions(+) diff --git a/assets/agents/code-reviewer/graph.yaml b/assets/agents/code-reviewer/graph.yaml index 6f4eb8a..464628a 100644 --- a/assets/agents/code-reviewer/graph.yaml +++ b/assets/agents/code-reviewer/graph.yaml @@ -272,6 +272,7 @@ nodes: Linter output (extract the lines for your files): {{linter_notes}} {{extra_context}} + max_attempts: 2 state_updates: domain_report: '{{output}}' next: completeness_gate @@ -472,6 +473,7 @@ nodes: Findings (JSON; verify every 🔴 and 🟡 entry — id, severity, path, lines, claim are the fields; `title` + `block` together are the claim): {{findings}} + max_attempts: 2 state_updates: verifier_output: '{{output}}' next: verdict diff --git a/assets/agents/deep-research/graph.yaml b/assets/agents/deep-research/graph.yaml index 4aa2b25..f8be351 100644 --- a/assets/agents/deep-research/graph.yaml +++ b/assets/agents/deep-research/graph.yaml @@ -242,6 +242,7 @@ nodes: Produce the final report following your instructions. timeout: 300 + max_attempts: 2 state_updates: report: '{{output}}' next: verify_sources diff --git a/assets/agents/review-gauntlet/graph.yaml b/assets/agents/review-gauntlet/graph.yaml index 54b3e74..e975f17 100644 --- a/assets/agents/review-gauntlet/graph.yaml +++ b/assets/agents/review-gauntlet/graph.yaml @@ -276,6 +276,7 @@ nodes: {{extra_context}} Close your report with your Review Verdict line. + max_attempts: 2 state_updates: lane_out: '{{output}}' @@ -296,6 +297,7 @@ nodes: {{extra_context}} Return your ADVERSARIAL_REVIEW verdict. + max_attempts: 2 state_updates: lane_out: '{{output}}' @@ -313,6 +315,7 @@ nodes: {{extra_context}} Return your SECURITY_REVIEW verdict. + max_attempts: 2 state_updates: lane_out: '{{output}}' @@ -335,6 +338,7 @@ nodes: Derive expected behaviors from the plan/contract BEFORE reading implementation code, run existing suites first, then author spec-first tests for the uncovered delta. Return your USAGE_PROBE verdict. + max_attempts: 2 state_updates: lane_out: '{{output}}' diff --git a/assets/agents/step-runner/graph.yaml b/assets/agents/step-runner/graph.yaml index dc8cbe0..5c3c85b 100644 --- a/assets/agents/step-runner/graph.yaml +++ b/assets/agents/step-runner/graph.yaml @@ -405,6 +405,7 @@ nodes: Review the changed files against the step plan's acceptance criteria. Preserve severity tags in your findings. timeout: 1200 + max_attempts: 2 state_updates: review_report: '{{output}}' next: route_review From 18875a4cc9596c7977c81520da5709832b5f851c Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Sat, 12 Sep 2026 12:22:13 -0600 Subject: [PATCH 04/45] feat(TASK-011): traverse agent fallback edges in map peer-discovery BFS --- src/graph/map.rs | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/src/graph/map.rs b/src/graph/map.rs index 2f982fc..56ec9cf 100644 --- a/src/graph/map.rs +++ b/src/graph/map.rs @@ -440,6 +440,7 @@ fn first_flagged_agent_by_bfs<'g>( match &node.node_type { NodeType::Script(s) => edges.extend(s.fallback.as_ref()), NodeType::Llm(l) => edges.extend(l.fallback.as_ref()), + NodeType::Agent(a) => edges.extend(a.fallback.as_ref()), _ => {} } for next in edges { @@ -2484,6 +2485,33 @@ nodes: } } + #[test] + fn provision_map_peers_finds_flagged_agent_behind_agent_fallback_edge() { + let h = Harness::new( + r#" +name: t +start: gate +nodes: + gate: + type: agent + agent: gatekeeper + prompt: "p" + fallback: rescue + rescue: + type: agent + agent: rescue-agent + prompt: "p" + teammates: true +"#, + ); + + let (_, assignments) = h.provision("gate", 2); + + for (id, _) in &assignments { + assert!(id.starts_with("graph_agent_rescue-agent_"), "{id}"); + } + } + #[test] fn provision_map_peers_ignores_flagged_agent_outside_subgraph() { let h = Harness::new(GATE_WORKER_FINISH_GRAPH); From 80ace77e4402eabd29cd810d301d2c269896997b Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Sat, 12 Sep 2026 14:42:42 -0600 Subject: [PATCH 05/45] 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); + }); +} From 7dc950a09274dd600f231e12bd81efd0789fb327 Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Sat, 12 Sep 2026 14:50:40 -0600 Subject: [PATCH 06/45] test(TASK-016): assert the fallback warn names the node id and fallback target MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The adversary conformance review flagged that the fallback warn! at outcome_from (src/graph/agent.rs:71) was never asserted — only the retry warn had captured-text coverage. Adds one test installing the shared warn collector, filtered by the unique node id 'fallback_warn_capture' for parallel safety, asserting the warn names both the node and the fallback target and carries the error chain. --- src/graph/agent.rs | 38 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/src/graph/agent.rs b/src/graph/agent.rs index 751bae7..23a94f4 100644 --- a/src/graph/agent.rs +++ b/src/graph/agent.rs @@ -796,6 +796,44 @@ mod tests { assert!(failure.contains("no JSON object found"), "{failure}"); } + #[test] + fn outcome_from_fallback_warn_names_the_node_and_the_fallback_target() { + install_warn_collector(); + let mut node = node_with("hi", None); + node.fallback = Some("recover".into()); + let mut state = manager_with(&[]); + + let outcome = outcome_from( + "fallback_warn_capture", + &node, + &mut state, + Err(extraction_error()), + ) + .unwrap(); + + assert_eq!(outcome, AgentExecutionOutcome::FellBack("recover".into())); + let warns: Vec = warn_messages() + .lock() + .unwrap() + .iter() + .filter(|m| m.contains("'fallback_warn_capture'")) + .cloned() + .collect(); + assert_eq!(warns.len(), 1, "exactly one fallback warn: {warns:?}"); + assert!( + warns[0].contains( + "agent node 'fallback_warn_capture' failed, routing to fallback 'recover'" + ), + "{}", + warns[0] + ); + assert!( + warns[0].contains("Agent 'test_agent' output failed structured-output extraction"), + "warn must carry the error chain: {}", + warns[0] + ); + } + #[test] fn outcome_from_failure_without_fallback_propagates_the_error_unchanged() { let node = node_with("hi", Some(failure_capture_updates())); From 3dcb9541b1186f6e395fc9e370fe321d8e6a9edf Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Sat, 12 Sep 2026 14:54:33 -0600 Subject: [PATCH 07/45] test(TASK-016): assert fallback warn carries the error root cause --- src/graph/agent.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/graph/agent.rs b/src/graph/agent.rs index 23a94f4..fc8edde 100644 --- a/src/graph/agent.rs +++ b/src/graph/agent.rs @@ -832,6 +832,11 @@ mod tests { "warn must carry the error chain: {}", warns[0] ); + assert!( + warns[0].contains("no JSON object found in output"), + "fallback warn must carry the full error chain: {}", + warns[0] + ); } #[test] From d50059a56f90dcfb977a224f2751b1c749131dc7 Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Sat, 12 Sep 2026 15:06:28 -0600 Subject: [PATCH 08/45] feat(TASK-017): adversary execution evidence + fail-closed routing - parse extracts caller-declared verification_commands (schema + initial_state); never invents commands. parse gains fallback + state_updates capturing the failure string (TASK-016 fallback-capture warning stays silent). - new run_checks script node (facts -> run_checks -> map_criteria, timeout 3600): runs each declared command in the project dir (900s per-command timeout), records exec_results {cmd, exit, duration_s, tail~50}; 'none declared' / ENVIRONMENT markers instead of guessing or failing the node. - check_criterion judges execution criteria against {{exec_results}} only: recorded green run = MET citing it, recorded failure = UNMET, markers = PARTIAL with an ENVIRONMENT note. - fail-closed routing (R1/R2): parse/facts fallback -> pipeline_fault marker script -> verdict; holistic fallback -> verdict with holistic_failure captured. verdict.py: any PIPELINE-FAULT forces DIVERGES with the fault as complaint #1, per-criterion results still reported; report gains a 'Verification runs:' section. Non-degraded semantics unchanged. - R5 rust tests (python3-guarded, local cmd_available): verdict fault path, holistic-failure fault, CONFORMS regression, run_checks markers + recorded runs, crit_gate contract regression. --- assets/agents/adversary/graph.yaml | 61 ++++- .../adversary/scripts/pipeline_fault.py | 58 +++++ assets/agents/adversary/scripts/run_checks.py | 113 +++++++++ assets/agents/adversary/scripts/verdict.py | 136 +++++++++-- src/config/request_context.rs | 222 ++++++++++++++++++ 5 files changed, 564 insertions(+), 26 deletions(-) create mode 100644 assets/agents/adversary/scripts/pipeline_fault.py create mode 100644 assets/agents/adversary/scripts/run_checks.py diff --git a/assets/agents/adversary/graph.yaml b/assets/agents/adversary/graph.yaml index 9ad6001..e267aee 100644 --- a/assets/agents/adversary/graph.yaml +++ b/assets/agents/adversary/graph.yaml @@ -12,7 +12,11 @@ description: | tests — ADDITIVE-only: it can add complaints, never flip a criterion verdict. The final CONFORMS/DIVERGES sentinel is computed by a script: CONFORMS requires every criterion MET and zero holistic complaints. Missing plan = - DIVERGES (fail-closed) naming the missing spec. + DIVERGES (fail-closed) naming the missing spec. Caller-declared verification + commands are executed ONCE by a deterministic stage and recorded — criterion + branches cite the record and never run anything themselves. Any pipeline + fault (parse, diff resolution, or holistic failure) lands a PIPELINE-FAULT + marker and fails closed into DIVERGES: degraded is never passing. version: '1.0' @@ -42,12 +46,17 @@ initial_state: changed_files: [] diff_text: '' diff_note: '' + verification_commands: [] + exec_results: '' crit_verdicts: [] extra_complaints: [] observations: '' adv_report: '' gate_feedback: '' gate_attempts: 0 + parse_failure: '' + holistic_failure: '' + pipeline_faults: [] start: parse @@ -75,6 +84,10 @@ nodes: verbatim — the holistic pass checks the diff didn't silently override them. - `scope_notes`: what the plan says NOT to touch / explicit boundaries. - `extra_context`: anything else conformance-relevant, condensed. + - `verification_commands`: the exact build/test/lint shell commands the + caller explicitly declares for verification (verbatim strings, in the + order given). Empty array when the caller declares none — NEVER invent, + guess, or auto-detect commands; undeclared stays undeclared. prompt: | Extract the conformance-review facts from this request: @@ -103,9 +116,16 @@ nodes: type: string project_dir_in: type: string - required: [criteria, diff_spec, plan_notes, scope_notes, extra_context, project_dir_in] + verification_commands: + type: array + items: + type: string + required: [criteria, diff_spec, plan_notes, scope_notes, extra_context, project_dir_in, verification_commands] max_iterations: 3 max_attempts: 2 + state_updates: + parse_failure: '{{output}}' + fallback: pipeline_fault next: facts facts: @@ -114,6 +134,15 @@ nodes: description: Deterministic diff resolution — changed files and the (bounded) diff text every criterion branch verifies against. script: scripts/diff_facts.py timeout: 120 + fallback: pipeline_fault + next: run_checks + + run_checks: + id: run_checks + type: script + description: Deterministic execution evidence — runs ONLY the caller-declared verification commands in the project dir and records exec_results; criterion branches cite the record and never execute anything. Undeclared commands degrade to a "none declared" marker, runner errors to an ENVIRONMENT marker; never fails the node. + script: scripts/run_checks.py + timeout: 3600 next: map_criteria map_criteria: @@ -158,6 +187,18 @@ nodes: values shaped to the criterion's example, TODO/stub presented as done — those are PARTIAL or DIVERGED with the gaming named in the complaint. + Execution criteria ("tests pass", "lint clean", "build succeeds") are + judged ONLY against the recorded verification runs in your prompt — you + have no execution tools BY DESIGN and must never claim to have run + anything yourself: + - a recorded green run (exit 0) of a command covering the criterion is + the proving evidence — MET, citing that run (the command + exit 0). + - a recorded failing run covering the criterion ⇒ UNMET, citing the + recorded exit code and output tail. + - "none declared" or an ENVIRONMENT marker ⇒ at best PARTIAL, with an + ENVIRONMENT note in the complaint naming the exact command the caller + should declare via verification_commands (e.g. "cargo test --all"). + Output format — respond with ONLY this JSON object, no prose around it: {"id": "", "status": "MET" | "PARTIAL" | "UNMET" | "DIVERGED", @@ -181,6 +222,9 @@ nodes: The diff{{diff_note}}: {{diff_text}} + + Verification runs (recorded by the deterministic run_checks stage): + {{exec_results}} {{gate_feedback}} tools: - fs_read @@ -273,12 +317,23 @@ nodes: required: [extra_complaints, observations] max_iterations: 10 max_attempts: 2 + state_updates: + holistic_failure: '{{output}}' + fallback: verdict + next: verdict + + pipeline_fault: + id: pipeline_fault + type: script + description: Fail-closed fault marker (reached only via fallback routes from parse/facts) — records a PIPELINE-FAULT in state and continues to verdict, which turns any fault into DIVERGES. The graph never dies without its sentinel. + script: scripts/pipeline_fault.py + timeout: 30 next: verdict verdict: id: verdict type: script - description: Deterministic verdict — CONFORMS requires every criterion MET and zero holistic complaints; no criteria at all fails closed. Assembles the exact sentinel format. + description: Deterministic verdict — CONFORMS requires every criterion MET, zero holistic complaints, and zero pipeline faults; no criteria at all fails closed and any recorded PIPELINE-FAULT forces DIVERGES (degraded is never passing). Assembles the exact sentinel format. script: scripts/verdict.py timeout: 30 next: done diff --git a/assets/agents/adversary/scripts/pipeline_fault.py b/assets/agents/adversary/scripts/pipeline_fault.py new file mode 100644 index 0000000..a16bfc7 --- /dev/null +++ b/assets/agents/adversary/scripts/pipeline_fault.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +"""Fail-closed fault marker for the adversary graph. + +Reached ONLY via `fallback:` routes (parse, facts). Appends a machine-readable +"PIPELINE-FAULT:" marker to `pipeline_faults` and continues to `verdict`, +which turns any recorded fault into DIVERGES — the graph never dies without +its sentinel, and a degraded run can never read as CONFORMS. + +Stage attribution: an llm-node failure writes an "LLM node …failed: " +string into `parse_failure` via parse's `state_updates`; anything else means +the deterministic facts (diff resolution) script failed, which leaves no +failure text in state by design. +""" + +import json +import os + +MAX_DETAIL_CHARS = 500 + + +def load_state(): + if path := os.environ.get("GRAPH_STATE_FILE"): + with open(path) as f: + return json.load(f) + return json.loads(os.environ.get("GRAPH_STATE", "{}")) + + +def main(): + state = load_state() + faults = [f for f in (state.get("pipeline_faults") or []) if isinstance(f, str)] + parse_failure = state.get("parse_failure") + detail = parse_failure.strip() if isinstance(parse_failure, str) else "" + if detail.startswith("LLM node"): + detail = detail.replace("\n", " ") + if len(detail) > MAX_DETAIL_CHARS: + detail = detail[: MAX_DETAIL_CHARS - 1] + "…" + faults.append(f"PIPELINE-FAULT: parse failed — cannot review: {detail}") + else: + faults.append( + "PIPELINE-FAULT: diff resolution failed — changed files and diff text " + "could not be resolved; criterion verification never ran" + ) + print(json.dumps({"pipeline_faults": faults})) + + +try: + main() +except Exception as e: # noqa: BLE001 — the fault marker itself must never crash + print( + json.dumps( + { + "pipeline_faults": [ + f"PIPELINE-FAULT: fault-marker script error: {e} — treat the " + "run as degraded" + ] + } + ) + ) diff --git a/assets/agents/adversary/scripts/run_checks.py b/assets/agents/adversary/scripts/run_checks.py new file mode 100644 index 0000000..c82e225 --- /dev/null +++ b/assets/agents/adversary/scripts/run_checks.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +"""Deterministic execution-evidence recorder for the adversary graph. + +Runs ONLY the caller-declared verification commands (verification_commands) +in the project dir and records the outcomes as `exec_results` — a list of +{cmd, exit, duration_s, tail} records. Criterion branches cite this record; +they never execute anything themselves. + +NO auto-detection: undeclared commands mean a "none declared" marker +(fail-visible, not fail-guessed). Any runner error degrades into an +ENVIRONMENT marker — this script never fails the node. +""" + +import json +import os +import subprocess +import time + +PER_COMMAND_TIMEOUT_SECS = 900 +TAIL_LINES = 50 +MAX_TAIL_CHARS = 8000 + + +def load_state(): + if path := os.environ.get("GRAPH_STATE_FILE"): + with open(path) as f: + return json.load(f) + return json.loads(os.environ.get("GRAPH_STATE", "{}")) + + +def tail_of(text): + tail = "\n".join(text.splitlines()[-TAIL_LINES:]) + if len(tail) > MAX_TAIL_CHARS: + tail = "…" + tail[-MAX_TAIL_CHARS:] + return tail + + +def main(): + state = load_state() + declared = state.get("verification_commands") + cmds = ( + [c.strip() for c in declared if isinstance(c, str) and c.strip()] + if isinstance(declared, list) + else [] + ) + if not cmds: + print( + json.dumps( + { + "exec_results": ( + "none declared — the caller supplied no verification_commands; " + "execution criteria (build/test/lint) cannot be independently " + "confirmed. Declare the exact commands in the spawn prompt to " + "get recorded runs." + ) + } + ) + ) + return + + proj = os.path.expanduser( + (state.get("project_dir") or "").strip() or "." + ) + results = [] + for cmd in cmds: + start = time.monotonic() + try: + r = subprocess.run( + cmd, + shell=True, + cwd=proj, + capture_output=True, + text=True, + timeout=PER_COMMAND_TIMEOUT_SECS, + ) + combined = (r.stdout or "") + (("\n" + r.stderr) if r.stderr else "") + results.append( + { + "cmd": cmd, + "exit": r.returncode, + "duration_s": round(time.monotonic() - start, 1), + "tail": tail_of(combined), + } + ) + except subprocess.TimeoutExpired: + results.append( + { + "cmd": cmd, + "exit": -1, + "duration_s": round(time.monotonic() - start, 1), + "tail": ( + f"TIMEOUT: no exit within {PER_COMMAND_TIMEOUT_SECS}s — " + "recorded as a failing run" + ), + } + ) + print(json.dumps({"exec_results": results})) + + +try: + main() +except Exception as e: # noqa: BLE001 — never fail the node; degrade visibly + print( + json.dumps( + { + "exec_results": ( + f"ENVIRONMENT: verification runner error: {e} — the declared " + "commands could not be executed; treat execution criteria as " + "PARTIAL (unproven), not as failed" + ) + } + ) + ) diff --git a/assets/agents/adversary/scripts/verdict.py b/assets/agents/adversary/scripts/verdict.py index f604e0b..524baca 100644 --- a/assets/agents/adversary/scripts/verdict.py +++ b/assets/agents/adversary/scripts/verdict.py @@ -1,15 +1,21 @@ #!/usr/bin/env python3 """Deterministic conformance verdict for the adversary graph. -CONFORMS requires: at least one criterion, every criterion MET, and zero -holistic complaints. Everything else — including a missing plan — is -DIVERGES (fail-closed). Assembles the exact sentinel format the callers -route on. Never crashes into a silent verdict. +CONFORMS requires: at least one criterion, every criterion MET, zero +holistic complaints, and zero pipeline faults. Everything else — including +a missing plan or a degraded (faulted) pipeline — is DIVERGES (fail-closed). +Any "PIPELINE-FAULT:" marker recorded in `pipeline_faults` (or a holistic +llm-node failure captured in `holistic_failure`) becomes complaint #1; +per-criterion results are still reported. Assembles the exact sentinel +format the callers route on. Never crashes into a silent verdict. """ import json import os +MAX_FAULT_DETAIL_CHARS = 300 + + def load_state(): if path := os.environ.get("GRAPH_STATE_FILE"): with open(path) as f: @@ -17,11 +23,85 @@ def load_state(): return json.loads(os.environ.get("GRAPH_STATE", "{}")) +def collect_pipeline_faults(state): + """Recorded fault markers, plus a holistic llm-node failure if one occurred. + + `holistic_failure` holds the "LLM node …failed: " string when the + holistic pass fell back to verdict; on a successful pass it holds the + rendered schema output (never that prefix). + """ + faults = [ + f.strip() + for f in (state.get("pipeline_faults") or []) + if isinstance(f, str) and f.strip() + ] + hf = state.get("holistic_failure") + if isinstance(hf, str) and hf.strip().startswith("LLM node"): + detail = hf.strip().replace("\n", " ") + if len(detail) > MAX_FAULT_DETAIL_CHARS: + detail = detail[: MAX_FAULT_DETAIL_CHARS - 1] + "…" + faults.append( + "PIPELINE-FAULT: holistic pass failed — criterion verdicts stand " + f"but cross-cutting hunt did not run ({detail})" + ) + return faults + + +def append_criterion_complaints(lines, verdicts, extra, start): + """Numbered complaints for non-MET verdicts and holistic extras.""" + i = start + for v in verdicts: + if v.get("status") == "MET": + continue + i += 1 + text = (v.get("text") or "").strip().replace("\n", " ") + if len(text) > 140: + text = text[:137] + "…" + lines.append( + f'{i}. Acceptance criterion "{text}" — {v.get("status", "?").title()} — ' + f'{(v.get("complaint") or "").strip()}' + ) + for c in extra: + i += 1 + lines.append( + f"{i}. {c.get('kind', 'violation')} — {c.get('location', '?')} — " + f"{c.get('violation', '')} — {c.get('fix', '')}" + ) + return i + + +def render_exec_results(exec_results): + """Lines for the "Verification runs:" report section. + + exec_results is a list of {cmd, exit, duration_s, tail} records from + run_checks.py, or a string marker ("none declared —…" / "ENVIRONMENT: …"), + or empty when the run never reached the verification stage. + """ + if isinstance(exec_results, list): + out = [] + for r in exec_results: + if not isinstance(r, dict): + continue + status = "PASS" if r.get("exit") == 0 else "FAIL" + out.append( + f"- [{status}] `{r.get('cmd', '?')}` — exit {r.get('exit', '?')}, " + f"{r.get('duration_s', '?')}s" + ) + if status == "FAIL": + tail = (r.get("tail") or "").strip() + out.extend(" " + t for t in tail.splitlines()[-15:]) + return out or ["- (no verification commands were run)"] + if isinstance(exec_results, str) and exec_results.strip(): + return [f"- {exec_results.strip()}"] + return ["- none recorded (the run did not reach the verification stage)"] + + def main(): state = load_state() verdicts = [v for v in (state.get("crit_verdicts") or []) if isinstance(v, dict)] extra = [c for c in (state.get("extra_complaints") or []) if isinstance(c, dict)] observations = (state.get("observations") or "").strip() + faults = collect_pipeline_faults(state) n = len(verdicts) met = [v for v in verdicts if v.get("status") == "MET"] @@ -29,7 +109,30 @@ def main(): bad = [v for v in verdicts if v.get("status") in ("UNMET", "DIVERGED")] lines = [] - if n == 0: + if faults: + lines.append("ADVERSARIAL_REVIEW: DIVERGES") + if n: + lines.append( + f"Criteria: {len(met)}/{n} met, {len(partial)} partial, " + f"{len(bad)} unmet/diverged — degraded run: pipeline fault recorded " + "(fail-closed)." + ) + else: + lines.append( + "Criteria: none verified — degraded run: pipeline fault recorded " + "(fail-closed)." + ) + lines.append("Complaints:") + i = 0 + for f in faults: + i += 1 + lines.append(f"{i}. {f}") + append_criterion_complaints(lines, verdicts, extra, i) + if observations: + lines.append("") + lines.append("Non-blocking observations:") + lines.append(observations) + elif n == 0: lines.append("ADVERSARIAL_REVIEW: DIVERGES") lines.append("Criteria: none provided.") lines.append("Complaints:") @@ -51,24 +154,7 @@ def main(): f"Criteria: {len(met)}/{n} met, {len(partial)} partial, {len(bad)} unmet/diverged." ) lines.append("Complaints:") - i = 0 - for v in verdicts: - if v.get("status") == "MET": - continue - i += 1 - text = (v.get("text") or "").strip().replace("\n", " ") - if len(text) > 140: - text = text[:137] + "…" - lines.append( - f'{i}. Acceptance criterion "{text}" — {v.get("status", "?").title()} — ' - f'{(v.get("complaint") or "").strip()}' - ) - for c in extra: - i += 1 - lines.append( - f"{i}. {c.get('kind', 'violation')} — {c.get('location', '?')} — " - f"{c.get('violation', '')} — {c.get('fix', '')}" - ) + append_criterion_complaints(lines, verdicts, extra, 0) if observations: lines.append("") lines.append("Non-blocking observations:") @@ -84,6 +170,10 @@ def main(): text = text[:97] + "…" lines.append(f'- "{text}" — {(v.get("evidence") or "").strip()}') + lines.append("") + lines.append("Verification runs:") + lines.extend(render_exec_results(state.get("exec_results"))) + print(json.dumps({"adv_report": "\n".join(lines)})) diff --git a/src/config/request_context.rs b/src/config/request_context.rs index 27ce19b..be469cd 100644 --- a/src/config/request_context.rs +++ b/src/config/request_context.rs @@ -8348,6 +8348,228 @@ mod tests { ); } + // ---- adversary suite-script regression tests (R5) ---- + // + // Fault paths of the adversary's verdict/gate scripts, exercised by + // invoking `python3