diff --git a/Cargo.lock b/Cargo.lock index 1e5a85662..8f9905ec9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -268,6 +268,7 @@ dependencies = [ "nom", "predicates", "rand 0.8.5", + "regex", "scopeguard", "serde", "serde_json", diff --git a/crates/arc-workflows/Cargo.toml b/crates/arc-workflows/Cargo.toml index d1d84daa6..136594a31 100644 --- a/crates/arc-workflows/Cargo.toml +++ b/crates/arc-workflows/Cargo.toml @@ -37,6 +37,7 @@ dialoguer.workspace = true daytona-sdk.workspace = true daytona-api-client.workspace = true base64.workspace = true +regex.workspace = true scopeguard = "1" git2.workspace = true tokio-util.workspace = true diff --git a/crates/arc-workflows/src/checkpoint.rs b/crates/arc-workflows/src/checkpoint.rs index 7779f321e..c455aeb17 100644 --- a/crates/arc-workflows/src/checkpoint.rs +++ b/crates/arc-workflows/src/checkpoint.rs @@ -6,7 +6,7 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use crate::context::Context; -use crate::error::{ArcError, Result}; +use crate::error::{ArcError, FailureSignature, Result}; use crate::outcome::Outcome; /// Serializable snapshot of execution state for crash recovery and resume. @@ -27,10 +27,17 @@ pub struct Checkpoint { /// SHA of the git commit created at this checkpoint (when running in a worktree). #[serde(default, skip_serializing_if = "Option::is_none")] pub git_commit_sha: Option, + /// Failure signature counts within the main loop (deterministic/structural failures). + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub loop_failure_signatures: HashMap, + /// Failure signature counts across loop_restart edges. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub restart_failure_signatures: HashMap, } impl Checkpoint { /// Create a checkpoint from the current execution state. + #[allow(clippy::too_many_arguments)] pub fn from_context( context: &Context, current_node: impl Into, @@ -38,6 +45,8 @@ impl Checkpoint { node_retries: HashMap, node_outcomes: HashMap, next_node_id: Option, + loop_failure_signatures: HashMap, + restart_failure_signatures: HashMap, ) -> Self { Self { timestamp: Utc::now(), @@ -49,6 +58,8 @@ impl Checkpoint { node_outcomes, next_node_id, git_commit_sha: None, + loop_failure_signatures, + restart_failure_signatures, } } @@ -94,6 +105,8 @@ mod tests { HashMap::new(), HashMap::new(), None, + HashMap::new(), + HashMap::new(), ); assert_eq!(cp.current_node, "node_a"); @@ -131,6 +144,8 @@ mod tests { retries, outcomes, Some("next_step".to_string()), + HashMap::new(), + HashMap::new(), ); cp.save(&path).unwrap(); @@ -170,10 +185,83 @@ mod tests { #[test] fn serialization_roundtrip() { let ctx = Context::new(); - let cp = Checkpoint::from_context(&ctx, "n1", vec![], HashMap::new(), HashMap::new(), None); + let cp = Checkpoint::from_context( + &ctx, + "n1", + vec![], + HashMap::new(), + HashMap::new(), + None, + HashMap::new(), + HashMap::new(), + ); let json = serde_json::to_string(&cp).unwrap(); let deserialized: Checkpoint = serde_json::from_str(&json).unwrap(); assert_eq!(deserialized.current_node, "n1"); } + + #[test] + fn signature_maps_roundtrip() { + use crate::error::{FailureClass, FailureSignature}; + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("checkpoint.json"); + + let ctx = Context::new(); + let mut loop_sigs = HashMap::new(); + loop_sigs.insert( + FailureSignature::new( + "verify", + FailureClass::Deterministic, + None, + Some("test failed"), + ), + 2, + ); + let mut restart_sigs = HashMap::new(); + restart_sigs.insert( + FailureSignature::new("build", FailureClass::Structural, None, Some("scope error")), + 1, + ); + + let cp = Checkpoint::from_context( + &ctx, + "verify", + vec![], + HashMap::new(), + HashMap::new(), + None, + loop_sigs, + restart_sigs, + ); + cp.save(&path).unwrap(); + + let loaded = Checkpoint::load(&path).unwrap(); + assert_eq!(loaded.loop_failure_signatures.len(), 1); + assert_eq!(loaded.restart_failure_signatures.len(), 1); + let sig = FailureSignature::new( + "verify", + FailureClass::Deterministic, + None, + Some("test failed"), + ); + assert_eq!(loaded.loop_failure_signatures.get(&sig), Some(&2)); + } + + #[test] + fn backward_compat_missing_signature_fields() { + // A checkpoint saved before signatures were added should deserialize with empty maps + let json = r#"{ + "timestamp": "2025-01-01T00:00:00Z", + "current_node": "work", + "completed_nodes": ["start"], + "node_retries": {}, + "context_values": {}, + "logs": [] + }"#; + let cp: Checkpoint = serde_json::from_str(json).unwrap(); + assert!(cp.loop_failure_signatures.is_empty()); + assert!(cp.restart_failure_signatures.is_empty()); + } } diff --git a/crates/arc-workflows/src/engine.rs b/crates/arc-workflows/src/engine.rs index 38e005969..0d0b20c94 100644 --- a/crates/arc-workflows/src/engine.rs +++ b/crates/arc-workflows/src/engine.rs @@ -16,7 +16,7 @@ use crate::artifact::{offload_large_values, sync_artifacts_to_env, ArtifactStore use crate::checkpoint::Checkpoint; use crate::condition::evaluate_condition; use crate::context::Context; -use crate::error::{classify_failure_reason, ArcError, FailureClass, Result}; +use crate::error::{classify_failure_reason, ArcError, FailureClass, FailureSignature, Result}; use crate::event::{EventEmitter, PipelineEvent}; use crate::graph::{Edge, Graph, Node}; use crate::handler::{EngineServices, HandlerRegistry}; @@ -59,6 +59,17 @@ fn classify_outcome(outcome: &Outcome) -> Option { } } +/// Mutable state carried across loop restarts and recursive `run_internal` calls. +#[derive(Default)] +struct LoopState { + node_visits: HashMap, + /// Tracks deterministic/structural failure signatures across main-loop stages. + /// Never reset on success — prevents impl-succeeds/verify-fails cycles. + loop_failure_signatures: HashMap, + /// Tracks failure signatures across loop_restart edges. + restart_failure_signatures: HashMap, +} + // --- Retry policy types --- /// Configuration for exponential backoff between retry attempts. @@ -666,10 +677,7 @@ pub async fn git_add_worktree_remote( } /// Remove a git worktree inside a remote execution environment. -pub async fn git_remove_worktree_remote( - exec_env: &dyn ExecutionEnvironment, - path: &str, -) -> bool { +pub async fn git_remove_worktree_remote(exec_env: &dyn ExecutionEnvironment, path: &str) -> bool { let cmd = format!("{GIT_REMOTE} worktree remove --force {path}"); matches!( exec_env.exec_command(&cmd, 30_000, None, None, None).await, @@ -678,10 +686,7 @@ pub async fn git_remove_worktree_remote( } /// Fast-forward merge to a given SHA inside a remote execution environment. -pub async fn git_merge_ff_only_remote( - exec_env: &dyn ExecutionEnvironment, - sha: &str, -) -> bool { +pub async fn git_merge_ff_only_remote(exec_env: &dyn ExecutionEnvironment, sha: &str) -> bool { let cmd = format!("{GIT_REMOTE} merge --ff-only {sha}"); matches!( exec_env.exec_command(&cmd, 30_000, None, None, None).await, @@ -692,10 +697,7 @@ pub async fn git_merge_ff_only_remote( /// Get the current HEAD SHA from a remote execution environment. pub async fn git_head_sha_remote(exec_env: &dyn ExecutionEnvironment) -> Option { let cmd = format!("{GIT_REMOTE} rev-parse HEAD"); - match exec_env - .exec_command(&cmd, 10_000, None, None, None) - .await - { + match exec_env.exec_command(&cmd, 10_000, None, None, None).await { Ok(r) if r.exit_code == 0 => Some(r.stdout.trim().to_string()), _ => None, } @@ -929,7 +931,7 @@ impl PipelineEngine { /// Returns an error if no start node is found, a node is missing, or a goal gate fails /// without a retry target. pub async fn run(&self, graph: &Graph, config: &RunConfig) -> Result { - self.run_internal(graph, config, None, None, HashMap::new()) + self.run_internal(graph, config, None, None, LoopState::default()) .await } @@ -945,7 +947,12 @@ impl PipelineEngine { config: &RunConfig, checkpoint: &Checkpoint, ) -> Result { - self.run_internal(graph, config, Some(checkpoint), None, HashMap::new()) + let loop_state = LoopState { + node_visits: HashMap::new(), + loop_failure_signatures: checkpoint.loop_failure_signatures.clone(), + restart_failure_signatures: checkpoint.restart_failure_signatures.clone(), + }; + self.run_internal(graph, config, Some(checkpoint), None, loop_state) .await } @@ -956,7 +963,7 @@ impl PipelineEngine { config: &RunConfig, resume_checkpoint: Option<&Checkpoint>, start_at: Option<&str>, - mut node_visits: HashMap, + mut loop_state: LoopState, ) -> Result { let run_start = Instant::now(); let run_id = config.run_id.clone(); @@ -1045,7 +1052,7 @@ impl PipelineEngine { completed_nodes = cp.completed_nodes.clone(); // Rebuild visit counts from completed_nodes (which records every visit) for id in &completed_nodes { - *node_visits.entry(id.clone()).or_insert(0) += 1; + *loop_state.node_visits.entry(id.clone()).or_insert(0) += 1; } // Gap #5: Restore retry counters from checkpoint node_retries = cp.node_retries.clone(); @@ -1108,7 +1115,10 @@ impl PipelineEngine { .ok_or_else(|| ArcError::Engine(format!("node not found: {current_node_id}")))?; // Always track visit count (used for stage directory naming) - let count = node_visits.entry(current_node_id.clone()).or_insert(0); + let count = loop_state + .node_visits + .entry(current_node_id.clone()) + .or_insert(0); *count += 1; if max_node_visits > 0 && *count > max_node_visits { return Err(ArcError::Engine(format!( @@ -1173,7 +1183,7 @@ impl PipelineEngine { } // Step 2: Execute node handler with retry policy - let visit = *node_visits.get(¤t_node_id).unwrap_or(&1); + let visit = *loop_state.node_visits.get(¤t_node_id).unwrap_or(&1); context.set("internal.node_visit_count", serde_json::json!(visit)); context.set("current_node", serde_json::json!(&node.id)); let retry_policy = build_retry_policy(node, graph); @@ -1224,6 +1234,36 @@ impl PipelineEngine { let outcome_failure_class = classify_outcome(&outcome); + // Circuit breaker: track deterministic/structural failure signatures + let failure_sig = if let Some(fc) = outcome_failure_class { + let sig_hint = outcome + .context_updates + .get("failure_signature") + .and_then(|v| v.as_str()); + let sig = FailureSignature::new( + &node.id, + fc, + sig_hint, + outcome.failure_reason.as_deref(), + ); + if fc.is_signature_tracked() { + let count = loop_state + .loop_failure_signatures + .entry(sig.clone()) + .or_insert(0); + *count += 1; + let limit = graph.loop_restart_signature_limit(); + if *count >= limit { + return Err(ArcError::Engine(format!( + "deterministic failure cycle detected: signature {sig} repeated {count} times (limit {limit})" + ))); + } + } + Some(sig) + } else { + None + }; + if outcome.status == StageStatus::Fail { self.services.emitter.emit(&PipelineEvent::StageFailed { name: node.label().to_string(), @@ -1285,6 +1325,12 @@ impl PipelineEngine { "failure_class", serde_json::json!(outcome_failure_class.map_or(String::new(), |fc| fc.to_string())), ); + context.set( + "failure_signature", + serde_json::json!(failure_sig + .as_ref() + .map_or(String::new(), |s| s.to_string())), + ); if let Some(ref pref) = outcome.preferred_label { context.set("preferred_label", serde_json::json!(pref)); } @@ -1309,6 +1355,8 @@ impl PipelineEngine { node_retries.clone(), node_outcomes.clone(), next_node_id_for_checkpoint, + loop_state.loop_failure_signatures.clone(), + loop_state.restart_failure_signatures.clone(), ); let checkpoint_path = config.logs_root.join("checkpoint.json"); if let Err(e) = checkpoint.save(&checkpoint_path) { @@ -1457,6 +1505,20 @@ impl PipelineEngine { incoming_edge = Some(edge); // Gap #6: Handle loop_restart by recursively running from the target if edge.loop_restart() { + // Circuit breaker: check restart failure signatures + if let Some(ref sig) = failure_sig { + let count = loop_state + .restart_failure_signatures + .entry(sig.clone()) + .or_insert(0); + *count += 1; + let limit = graph.loop_restart_signature_limit(); + if *count >= limit { + return Err(ArcError::Engine(format!( + "loop_restart circuit breaker: signature {sig} repeated {count} times (limit {limit})" + ))); + } + } self.services.emitter.emit(&PipelineEvent::LoopRestart { from_node: node.id.clone(), to_node: edge.to.clone(), @@ -1466,7 +1528,7 @@ impl PipelineEngine { config, None, Some(&edge.to), - node_visits, + loop_state, )) .await; } @@ -3483,4 +3545,342 @@ mod tests { Some(FailureClass::TransientInfra) ); } + + // --- Circuit breaker tests --- + + /// Build a graph where `work` always fails deterministically, + /// and a fail edge loops back to `work`. + fn looping_fail_graph() -> Graph { + let mut g = Graph::new("loop_fail"); + g.attrs + .insert("goal".to_string(), AttrValue::String("test".to_string())); + g.attrs + .insert("default_max_retry".to_string(), AttrValue::Integer(0)); + + let mut start = Node::new("start"); + start.attrs.insert( + "shape".to_string(), + AttrValue::String("Mdiamond".to_string()), + ); + g.nodes.insert("start".to_string(), start); + + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("always_fail".to_string()), + ); + work.attrs + .insert("max_retries".to_string(), AttrValue::Integer(0)); + g.nodes.insert("work".to_string(), work); + + let mut exit = Node::new("exit"); + exit.attrs.insert( + "shape".to_string(), + AttrValue::String("Msquare".to_string()), + ); + g.nodes.insert("exit".to_string(), exit); + + g.edges.push(Edge::new("start", "work")); + // Fail loops back + let mut fail_edge = Edge::new("work", "work"); + fail_edge.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=fail".to_string()), + ); + g.edges.push(fail_edge); + // Success goes to exit (never taken) + let mut ok_edge = Edge::new("work", "exit"); + ok_edge.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=success".to_string()), + ); + g.edges.push(ok_edge); + g + } + + /// Handler that always returns transient_infra failure. + struct TransientFailHandler; + + #[async_trait] + impl HandlerTrait for TransientFailHandler { + async fn execute( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _logs_root: &Path, + _services: &crate::handler::EngineServices, + ) -> std::result::Result { + let mut outcome = Outcome::fail("connection refused"); + outcome.context_updates.insert( + "failure_class".to_string(), + serde_json::json!("transient_infra"), + ); + Ok(outcome) + } + } + + /// Handler that fails with a semantically different message each time. + /// Uses words instead of numbers to avoid normalization collapsing them. + struct VaryingFailHandler { + counter: std::sync::atomic::AtomicUsize, + } + + static VARYING_REASONS: &[&str] = &[ + "syntax error in module alpha", + "type mismatch in module beta", + "missing field in module gamma", + "undefined reference in module delta", + "assertion failed in module epsilon", + ]; + + #[async_trait] + impl HandlerTrait for VaryingFailHandler { + async fn execute( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _logs_root: &Path, + _services: &crate::handler::EngineServices, + ) -> std::result::Result { + let n = self.counter.fetch_add(1, Ordering::Relaxed); + let reason = VARYING_REASONS[n % VARYING_REASONS.len()]; + Ok(Outcome::fail(reason)) + } + } + + #[tokio::test] + async fn loop_circuit_breaker_aborts_on_repeated_deterministic_failure() { + let dir = tempfile::tempdir().unwrap(); + let g = looping_fail_graph(); + + let mut registry = make_registry(); + registry.register("always_fail", Box::new(AlwaysFailHandler)); + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "test-run".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + let result = engine.run(&g, &config).await; + assert!(result.is_err()); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("deterministic failure cycle detected"), + "expected circuit breaker error, got: {err}" + ); + } + + #[tokio::test] + async fn loop_circuit_breaker_ignores_transient_failures() { + let dir = tempfile::tempdir().unwrap(); + let mut g = looping_fail_graph(); + // Set a high visit limit so we don't trip it; we want to hit the visit limit, not circuit breaker + g.attrs + .insert("max_node_visits".to_string(), AttrValue::Integer(5)); + + let mut registry = make_registry(); + registry.register("always_fail", Box::new(TransientFailHandler)); + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "test-run".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + let result = engine.run(&g, &config).await; + assert!(result.is_err()); + let err = result.unwrap_err().to_string(); + // Should hit visit limit, NOT circuit breaker + assert!( + err.contains("exceeded max visit limit"), + "expected visit limit error (transient shouldn't trigger circuit breaker), got: {err}" + ); + } + + #[tokio::test] + async fn loop_circuit_breaker_different_reasons_get_separate_counters() { + let dir = tempfile::tempdir().unwrap(); + let mut g = looping_fail_graph(); + // Each failure has a different message, so no signature repeats. + // Should hit max_node_visits instead of circuit breaker. + g.attrs + .insert("max_node_visits".to_string(), AttrValue::Integer(5)); + + let mut registry = make_registry(); + registry.register( + "always_fail", + Box::new(VaryingFailHandler { + counter: std::sync::atomic::AtomicUsize::new(0), + }), + ); + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "test-run".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + let result = engine.run(&g, &config).await; + assert!(result.is_err()); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("exceeded max visit limit"), + "expected visit limit (each failure unique), got: {err}" + ); + } + + #[tokio::test] + async fn restart_circuit_breaker_aborts_on_repeated_failure() { + // In a pipeline with loop_restart edges, a repeating deterministic failure + // triggers a circuit breaker (either loop or restart, depending on topology). + let dir = tempfile::tempdir().unwrap(); + let mut g = Graph::new("restart_test"); + g.attrs + .insert("goal".to_string(), AttrValue::String("test".to_string())); + g.attrs + .insert("default_max_retry".to_string(), AttrValue::Integer(0)); + g.attrs + .insert("max_node_visits".to_string(), AttrValue::Integer(100)); + + let mut start = Node::new("start"); + start.attrs.insert( + "shape".to_string(), + AttrValue::String("Mdiamond".to_string()), + ); + g.nodes.insert("start".to_string(), start); + + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("always_fail".to_string()), + ); + work.attrs + .insert("max_retries".to_string(), AttrValue::Integer(0)); + g.nodes.insert("work".to_string(), work); + + let mut exit = Node::new("exit"); + exit.attrs.insert( + "shape".to_string(), + AttrValue::String("Msquare".to_string()), + ); + g.nodes.insert("exit".to_string(), exit); + + g.edges.push(Edge::new("start", "work")); + // loop_restart edge on failure + let mut restart_edge = Edge::new("work", "start"); + restart_edge.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=fail".to_string()), + ); + restart_edge + .attrs + .insert("loop_restart".to_string(), AttrValue::Boolean(true)); + g.edges.push(restart_edge); + // Success goes to exit + let mut ok_edge = Edge::new("work", "exit"); + ok_edge.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=success".to_string()), + ); + g.edges.push(ok_edge); + + let mut registry = make_registry(); + registry.register("always_fail", Box::new(AlwaysFailHandler)); + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "test-run".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + let result = engine.run(&g, &config).await; + assert!(result.is_err()); + let err = result.unwrap_err().to_string(); + // A circuit breaker fires (loop or restart) rather than looping indefinitely + assert!( + err.contains("failure cycle detected") || err.contains("circuit breaker"), + "expected circuit breaker error, got: {err}" + ); + } + + #[tokio::test] + async fn failure_signature_stored_in_context() { + let dir = tempfile::tempdir().unwrap(); + // Simple pipeline: start -> work (fails) -> exit (via fail edge) + let mut g = Graph::new("sig_context_test"); + g.attrs + .insert("goal".to_string(), AttrValue::String("test".to_string())); + g.attrs + .insert("default_max_retry".to_string(), AttrValue::Integer(0)); + + let mut start = Node::new("start"); + start.attrs.insert( + "shape".to_string(), + AttrValue::String("Mdiamond".to_string()), + ); + g.nodes.insert("start".to_string(), start); + + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("always_fail".to_string()), + ); + work.attrs + .insert("max_retries".to_string(), AttrValue::Integer(0)); + g.nodes.insert("work".to_string(), work); + + let mut exit = Node::new("exit"); + exit.attrs.insert( + "shape".to_string(), + AttrValue::String("Msquare".to_string()), + ); + g.nodes.insert("exit".to_string(), exit); + + g.edges.push(Edge::new("start", "work")); + g.edges.push(Edge::new("work", "exit")); + + let mut registry = make_registry(); + registry.register("always_fail", Box::new(AlwaysFailHandler)); + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "test-run".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + let _outcome = engine.run(&g, &config).await.unwrap(); + + // Check the checkpoint for the failure_signature context value + let checkpoint_path = dir.path().join("checkpoint.json"); + let cp = Checkpoint::load(&checkpoint_path).unwrap(); + let sig_value = cp.context_values.get("failure_signature").unwrap(); + let sig_str = sig_value.as_str().unwrap(); + assert!( + sig_str.contains("work|deterministic|"), + "expected failure signature in context, got: {sig_str}" + ); + } } diff --git a/crates/arc-workflows/src/error.rs b/crates/arc-workflows/src/error.rs index f9311b056..866ca917a 100644 --- a/crates/arc-workflows/src/error.rs +++ b/crates/arc-workflows/src/error.rs @@ -55,9 +55,15 @@ impl FromStr for FailureClass { "structural" => Self::Structural, // Aliases: transient_infra - "transient" | "transient-infra" | "infra_transient" | "transient infra" - | "infrastructure_transient" | "retryable" | "toolchain_workspace_io" - | "toolchain-workspace-io" | "toolchain_or_dependency_registry_unavailable" + "transient" + | "transient-infra" + | "infra_transient" + | "transient infra" + | "infrastructure_transient" + | "retryable" + | "toolchain_workspace_io" + | "toolchain-workspace-io" + | "toolchain_or_dependency_registry_unavailable" | "toolchain-dependency-registry-unavailable" => Self::TransientInfra, // Aliases: deterministic @@ -208,6 +214,80 @@ pub fn classify_failure_reason(reason: &str) -> FailureClass { FailureClass::Deterministic } +/// Normalize a failure reason for stable signature grouping. +/// +/// Replaces variable data (hex strings, digits) with placeholders so that +/// semantically identical errors produce the same signature regardless of +/// line numbers, commit hashes, or timestamps. +pub fn normalize_failure_reason(reason: &str) -> String { + use regex::Regex; + use std::sync::LazyLock; + + static HEX_RE: LazyLock = LazyLock::new(|| Regex::new(r"\b[0-9a-f]{7,64}\b").unwrap()); + static DIGITS_RE: LazyLock = LazyLock::new(|| Regex::new(r"\b\d+\b").unwrap()); + static COMMA_SPACE_RE: LazyLock = LazyLock::new(|| Regex::new(r",\s+").unwrap()); + static WHITESPACE_RE: LazyLock = LazyLock::new(|| Regex::new(r"\s+").unwrap()); + + let s = reason.trim().to_lowercase(); + if s.is_empty() { + return String::new(); + } + let s = HEX_RE.replace_all(&s, ""); + let s = DIGITS_RE.replace_all(&s, ""); + let s = COMMA_SPACE_RE.replace_all(&s, ","); + let s = WHITESPACE_RE.replace_all(&s, " "); + let s = s.trim(); + if s.len() > 240 { + s[..240].to_string() + } else { + s.to_string() + } +} + +/// Composite key that uniquely identifies a specific recurring failure. +/// +/// Format: `node_id|failure_class|normalized_reason` +/// +/// Used by circuit breakers to detect when the same failure keeps repeating, +/// e.g. "verify|deterministic|assertion failed in foo_test". +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct FailureSignature(String); + +impl FailureSignature { + /// Build a signature from failure context. + /// + /// The signature hint from `outcome.context_updates["failure_signature"]` takes + /// priority over the raw `failure_reason`, allowing handlers to provide explicit + /// grouping keys. + pub fn new( + node_id: &str, + failure_class: FailureClass, + signature_hint: Option<&str>, + failure_reason: Option<&str>, + ) -> Self { + let reason = signature_hint + .map(normalize_failure_reason) + .filter(|s| !s.is_empty()) + .or_else(|| failure_reason.map(normalize_failure_reason)) + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "unknown".to_string()); + Self(format!("{}|{}|{}", node_id.trim(), failure_class, reason)) + } +} + +impl fmt::Display for FailureSignature { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +impl FailureClass { + /// Whether this failure class should be tracked by the cycle breaker. + pub fn is_signature_tracked(self) -> bool { + matches!(self, Self::Deterministic | Self::Structural) + } +} + #[derive(Error, Debug, Clone)] pub enum ArcError { #[error("Parse error: {0}")] @@ -1189,4 +1269,133 @@ mod tests { FailureClass::Deterministic ); } + + // --- normalize_failure_reason tests --- + + #[test] + fn normalize_empty_and_whitespace_returns_empty() { + assert_eq!(normalize_failure_reason(""), ""); + assert_eq!(normalize_failure_reason(" "), ""); + assert_eq!(normalize_failure_reason("\n\t"), ""); + } + + #[test] + fn normalize_lowercases_and_trims() { + assert_eq!(normalize_failure_reason(" Hello World "), "hello world"); + } + + #[test] + fn normalize_replaces_hex_strings() { + assert_eq!( + normalize_failure_reason("commit abc123def0"), + "commit " + ); + // Short hex (< 7 chars) not replaced + assert_eq!(normalize_failure_reason("value abcdef"), "value abcdef"); + } + + #[test] + fn normalize_replaces_digit_sequences() { + assert_eq!(normalize_failure_reason("line 42"), "line "); + assert_eq!(normalize_failure_reason("error 0"), "error "); + } + + #[test] + fn normalize_collapses_comma_space_and_whitespace() { + assert_eq!(normalize_failure_reason("a, b, c"), "a,b,c"); + assert_eq!(normalize_failure_reason("a b"), "a b"); + } + + #[test] + fn normalize_truncates_to_240_chars() { + let long = "a".repeat(300); + let result = normalize_failure_reason(&long); + assert_eq!(result.len(), 240); + } + + #[test] + fn normalize_combined_example() { + assert_eq!( + normalize_failure_reason("Error at line 42 in abc123def"), + "error at line in " + ); + } + + // --- FailureSignature tests --- + + #[test] + fn failure_signature_format() { + let sig = FailureSignature::new( + "verify", + FailureClass::Deterministic, + None, + Some("test failed"), + ); + assert_eq!(sig.to_string(), "verify|deterministic|test failed"); + } + + #[test] + fn failure_signature_display() { + let sig = FailureSignature::new( + "build", + FailureClass::Structural, + None, + Some("scope violation"), + ); + assert_eq!(format!("{sig}"), "build|structural|scope violation"); + } + + #[test] + fn failure_signature_hint_takes_priority() { + let sig = FailureSignature::new( + "verify", + FailureClass::Deterministic, + Some("custom hint"), + Some("raw reason"), + ); + assert_eq!(sig.to_string(), "verify|deterministic|custom hint"); + } + + #[test] + fn failure_signature_missing_reason_falls_back_to_unknown() { + let sig = FailureSignature::new("node", FailureClass::Deterministic, None, None); + assert_eq!(sig.to_string(), "node|deterministic|unknown"); + } + + #[test] + fn failure_signature_equality_and_hash() { + let sig1 = FailureSignature::new( + "verify", + FailureClass::Deterministic, + None, + Some("test failed"), + ); + let sig2 = FailureSignature::new( + "verify", + FailureClass::Deterministic, + None, + Some("test failed"), + ); + assert_eq!(sig1, sig2); + + let mut map = std::collections::HashMap::new(); + map.insert(sig1.clone(), 1); + assert_eq!(map.get(&sig2), Some(&1)); + } + + // --- is_signature_tracked tests --- + + #[test] + fn is_signature_tracked_deterministic_and_structural() { + assert!(FailureClass::Deterministic.is_signature_tracked()); + assert!(FailureClass::Structural.is_signature_tracked()); + } + + #[test] + fn is_signature_tracked_false_for_others() { + assert!(!FailureClass::TransientInfra.is_signature_tracked()); + assert!(!FailureClass::BudgetExhausted.is_signature_tracked()); + assert!(!FailureClass::Canceled.is_signature_tracked()); + assert!(!FailureClass::CompilationLoop.is_signature_tracked()); + } } diff --git a/crates/arc-workflows/src/git.rs b/crates/arc-workflows/src/git.rs index d03408983..a26ee907d 100644 --- a/crates/arc-workflows/src/git.rs +++ b/crates/arc-workflows/src/git.rs @@ -737,6 +737,8 @@ mod tests { std::collections::HashMap::new(), std::collections::HashMap::new(), Some("node_b".to_string()), + std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let cp_json = serde_json::to_vec_pretty(&cp).unwrap(); store.write_checkpoint("RUN2", &cp_json, &[]).unwrap(); @@ -769,6 +771,8 @@ mod tests { std::collections::HashMap::new(), std::collections::HashMap::new(), None, + std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let cp1_json = serde_json::to_vec_pretty(&cp1).unwrap(); store.write_checkpoint("RUN3", &cp1_json, &[]).unwrap(); @@ -780,6 +784,8 @@ mod tests { std::collections::HashMap::new(), std::collections::HashMap::new(), Some("node_c".to_string()), + std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let cp2_json = serde_json::to_vec_pretty(&cp2).unwrap(); store.write_checkpoint("RUN3", &cp2_json, &[]).unwrap(); @@ -846,10 +852,7 @@ mod tests { #[test] fn sanitize_ref_component_mixed() { - assert_eq!( - sanitize_ref_component("My Node!@#123"), - "my-node-123" - ); + assert_eq!(sanitize_ref_component("My Node!@#123"), "my-node-123"); } #[test] diff --git a/crates/arc-workflows/src/graph/types.rs b/crates/arc-workflows/src/graph/types.rs index d6aad1eb0..e9152e8bd 100644 --- a/crates/arc-workflows/src/graph/types.rs +++ b/crates/arc-workflows/src/graph/types.rs @@ -419,6 +419,16 @@ impl Graph { self.attrs.get("default_thread").and_then(AttrValue::as_str) } + /// Graph-level `loop_restart_signature_limit` (default 3). + /// When the same failure signature repeats this many times, the pipeline aborts. + pub fn loop_restart_signature_limit(&self) -> usize { + self.attrs + .get("loop_restart_signature_limit") + .and_then(AttrValue::as_i64) + .filter(|&v| v >= 1) + .map_or(3, |v| v as usize) + } + /// Graph-level `max_node_visits` (default 0 = disabled). pub fn max_node_visits(&self) -> u64 { self.attrs @@ -746,4 +756,36 @@ mod tests { ); assert!(node.codergen_mode().is_err()); } + + #[test] + fn graph_loop_restart_signature_limit_default() { + let g = Graph::new("empty"); + assert_eq!(g.loop_restart_signature_limit(), 3); + } + + #[test] + fn graph_loop_restart_signature_limit_set() { + let mut g = Graph::new("test"); + g.attrs.insert( + "loop_restart_signature_limit".to_string(), + AttrValue::Integer(5), + ); + assert_eq!(g.loop_restart_signature_limit(), 5); + } + + #[test] + fn graph_loop_restart_signature_limit_invalid_falls_back() { + let mut g = Graph::new("test"); + g.attrs.insert( + "loop_restart_signature_limit".to_string(), + AttrValue::Integer(0), + ); + assert_eq!(g.loop_restart_signature_limit(), 3); + + g.attrs.insert( + "loop_restart_signature_limit".to_string(), + AttrValue::Integer(-1), + ); + assert_eq!(g.loop_restart_signature_limit(), 3); + } } diff --git a/crates/arc-workflows/src/retro.rs b/crates/arc-workflows/src/retro.rs index 797ce3b9c..5fd5cbee6 100644 --- a/crates/arc-workflows/src/retro.rs +++ b/crates/arc-workflows/src/retro.rs @@ -245,9 +245,7 @@ pub fn derive_retro( *total_cost.get_or_insert(0.0) += c; } - let files = outcome - .map(|o| o.files_touched.clone()) - .unwrap_or_default(); + let files = outcome.map(|o| o.files_touched.clone()).unwrap_or_default(); all_files.extend(files.iter().cloned()); stages.push(StageRetro { @@ -344,6 +342,8 @@ mod tests { node_outcomes, next_node_id: None, git_commit_sha: None, + loop_failure_signatures: HashMap::new(), + restart_failure_signatures: HashMap::new(), } } @@ -355,7 +355,16 @@ mod tests { .into_iter() .collect(); - let retro = derive_retro("run-1", "my_pipeline", "Fix the bug", &cp, false, None, 20000, &durations); + let retro = derive_retro( + "run-1", + "my_pipeline", + "Fix the bug", + &cp, + false, + None, + 20000, + &durations, + ); assert_eq!(retro.run_id, "run-1"); assert_eq!(retro.pipeline_name, "my_pipeline"); @@ -393,11 +402,19 @@ mod tests { }, next_node_id: None, git_commit_sha: None, + loop_failure_signatures: HashMap::new(), + restart_failure_signatures: HashMap::new(), }; let retro = derive_retro( - "run-2", "pipe", "goal", &cp, true, - Some("boom"), 5000, &HashMap::new(), + "run-2", + "pipe", + "goal", + &cp, + true, + Some("boom"), + 5000, + &HashMap::new(), ); assert_eq!(retro.stats.stages_failed, 1); @@ -428,7 +445,10 @@ mod tests { assert_eq!(retro.smoothness, Some(SmoothnessRating::Smooth)); assert_eq!(retro.intent.as_deref(), Some("Fix authentication bug")); - assert_eq!(retro.outcome.as_deref(), Some("Successfully fixed the login flow")); + assert_eq!( + retro.outcome.as_deref(), + Some("Successfully fixed the login flow") + ); assert_eq!(retro.learnings.as_ref().unwrap().len(), 1); assert!(retro.friction_points.is_none()); // empty vec -> None assert_eq!(retro.open_items.as_ref().unwrap().len(), 1); @@ -438,7 +458,16 @@ mod tests { fn save_and_load_roundtrip() { let dir = tempfile::tempdir().unwrap(); let cp = make_checkpoint_with_stages(); - let mut retro = derive_retro("r1", "pipe", "goal", &cp, false, None, 1000, &HashMap::new()); + let mut retro = derive_retro( + "r1", + "pipe", + "goal", + &cp, + false, + None, + 1000, + &HashMap::new(), + ); retro.smoothness = Some(SmoothnessRating::Bumpy); retro.intent = Some("Test intent".to_string()); diff --git a/crates/arc-workflows/tests/integration.rs b/crates/arc-workflows/tests/integration.rs index cc6c19c02..b198347a0 100644 --- a/crates/arc-workflows/tests/integration.rs +++ b/crates/arc-workflows/tests/integration.rs @@ -1095,6 +1095,8 @@ fn checkpoint_save_and_resume_roundtrip() { retries, std::collections::HashMap::new(), None, + std::collections::HashMap::new(), + std::collections::HashMap::new(), ); checkpoint.save(&path).expect("save should succeed"); @@ -1560,6 +1562,8 @@ async fn resume_from_checkpoint_completes_pipeline() { std::collections::HashMap::new(), outcomes, Some("step_b".to_string()), + std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let dir = tempfile::tempdir().unwrap(); @@ -1653,6 +1657,8 @@ async fn resume_from_checkpoint_preserves_goal_gate_outcomes() { std::collections::HashMap::new(), outcomes, Some("step_b".to_string()), + std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let dir = tempfile::tempdir().unwrap(); @@ -2520,6 +2526,8 @@ async fn scenario_crash_recovery() { std::collections::HashMap::new(), outcomes, Some("b".to_string()), + std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let dir = tempfile::tempdir().unwrap(); @@ -4260,6 +4268,8 @@ async fn fidelity_resume_degrades_full_to_summary_high() { std::collections::HashMap::new(), outcomes, Some("step_b".to_string()), + std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let captures = FidelityCaptures::new(); @@ -4350,6 +4360,8 @@ async fn fidelity_resume_degrade_only_affects_first_hop() { std::collections::HashMap::new(), outcomes, Some("step_b".to_string()), + std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let captures = FidelityCaptures::new(); @@ -4427,6 +4439,8 @@ async fn fidelity_resume_no_degrade_when_not_full() { std::collections::HashMap::new(), outcomes, Some("step_b".to_string()), + std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let captures = FidelityCaptures::new(); @@ -5223,6 +5237,8 @@ async fn fidelity_resume_preserves_context_values_across_checkpoint() { std::collections::HashMap::new(), outcomes, Some("step_b".to_string()), + std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let captures = FidelityCaptures::new(); @@ -9173,15 +9189,17 @@ async fn parallel_git_branching_host_e2e() { ); let mut start = Node::new("start"); - start - .attrs - .insert("shape".to_string(), AttrValue::String("Mdiamond".to_string())); + start.attrs.insert( + "shape".to_string(), + AttrValue::String("Mdiamond".to_string()), + ); graph.nodes.insert("start".to_string(), start); let mut fan_out = Node::new("fan_out"); - fan_out - .attrs - .insert("shape".to_string(), AttrValue::String("component".to_string())); + fan_out.attrs.insert( + "shape".to_string(), + AttrValue::String("component".to_string()), + ); graph.nodes.insert("fan_out".to_string(), fan_out); let branch_a = Node::new("branch_a"); @@ -9198,8 +9216,10 @@ async fn parallel_git_branching_host_e2e() { graph.nodes.insert("fan_in".to_string(), fan_in); let mut exit = Node::new("exit"); - exit.attrs - .insert("shape".to_string(), AttrValue::String("Msquare".to_string())); + exit.attrs.insert( + "shape".to_string(), + AttrValue::String("Msquare".to_string()), + ); graph.nodes.insert("exit".to_string(), exit); graph.edges.push(Edge::new("start", "fan_out")); @@ -9214,8 +9234,9 @@ async fn parallel_git_branching_host_e2e() { let mut emitter = EventEmitter::new(); let events = collect_events(&mut emitter); - let env: Arc = - Arc::new(arc_agent::LocalExecutionEnvironment::new(worktree_path.clone())); + let env: Arc = Arc::new( + arc_agent::LocalExecutionEnvironment::new(worktree_path.clone()), + ); let mut registry = HandlerRegistry::new(Box::new(FileWriterHandler)); registry.register("start", Box::new(StartHandler)); @@ -9252,8 +9273,8 @@ async fn parallel_git_branching_host_e2e() { ); // 6. Verify parallel.results has head_sha for each branch - let checkpoint = Checkpoint::load(&logs_dir.path().join("checkpoint.json")) - .expect("checkpoint should load"); + let checkpoint = + Checkpoint::load(&logs_dir.path().join("checkpoint.json")).expect("checkpoint should load"); let parallel_results = checkpoint .context_values .get("parallel.results") @@ -9297,7 +9318,10 @@ async fn parallel_git_branching_host_e2e() { .expect("fan_in should have set best_head_sha"); // Heuristic select with both success: lexical tiebreak picks "branch_a" - assert_eq!(best_id, "branch_a", "heuristic should pick branch_a (lexical)"); + assert_eq!( + best_id, "branch_a", + "heuristic should pick branch_a (lexical)" + ); // 8. Verify winner's file is in the main worktree, loser's is NOT let winner_file = worktree_path.join(format!("{best_id}.txt")); @@ -9401,4 +9425,987 @@ async fn parallel_git_branching_host_e2e() { .output(); } +// --------------------------------------------------------------------------- +// Failure Signatures & Circuit Breaker E2E Tests +// --------------------------------------------------------------------------- + +/// Handler that always fails with a fixed deterministic reason. +struct DeterministicFailHandler { + reason: String, +} + +impl DeterministicFailHandler { + fn new(reason: &str) -> Self { + Self { + reason: reason.to_string(), + } + } +} + +#[async_trait::async_trait] +impl Handler for DeterministicFailHandler { + async fn execute( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _logs_root: &Path, + _services: &arc_workflows::handler::EngineServices, + ) -> Result { + Ok(Outcome::fail(&self.reason)) + } +} + +/// Handler that always fails with a transient_infra classification hint. +struct TransientInfraFailHandler; + +#[async_trait::async_trait] +impl Handler for TransientInfraFailHandler { + async fn execute( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _logs_root: &Path, + _services: &arc_workflows::handler::EngineServices, + ) -> Result { + let mut outcome = Outcome::fail("connection refused"); + outcome.context_updates.insert( + "failure_class".to_string(), + serde_json::json!("transient_infra"), + ); + Ok(outcome) + } +} + +/// Handler that provides an explicit `failure_signature` hint in context_updates. +struct SignatureHintHandler; + +#[async_trait::async_trait] +impl Handler for SignatureHintHandler { + async fn execute( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _logs_root: &Path, + _services: &arc_workflows::handler::EngineServices, + ) -> Result { + let mut outcome = Outcome::fail("error at line 42 in commit abc123def0"); + outcome.context_updates.insert( + "failure_signature".to_string(), + serde_json::json!("custom-grouping-key"), + ); + Ok(outcome) + } +} + +/// Handler that fails with varying reasons each call (truly different after normalization). +struct VaryingReasonFailHandler { + counter: std::sync::atomic::AtomicU32, +} + +static E2E_VARYING_REASONS: &[&str] = &[ + "syntax error in module alpha", + "type mismatch in module beta", + "missing field in module gamma", + "undefined reference in module delta", + "assertion failed in module epsilon", + "panic in module zeta", + "out of bounds in module eta", + "null pointer in module theta", + "stack overflow in module iota", + "deadlock in module kappa", +]; + +#[async_trait::async_trait] +impl Handler for VaryingReasonFailHandler { + async fn execute( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _logs_root: &Path, + _services: &arc_workflows::handler::EngineServices, + ) -> Result { + let n = self + .counter + .fetch_add(1, std::sync::atomic::Ordering::SeqCst) as usize; + Ok(Outcome::fail( + E2E_VARYING_REASONS[n % E2E_VARYING_REASONS.len()], + )) + } +} + +/// Handler that succeeds on the Nth call (0-indexed). Fails deterministically before that. +struct SucceedOnNthHandler { + succeed_on: u32, + counter: std::sync::atomic::AtomicU32, +} + +#[async_trait::async_trait] +impl Handler for SucceedOnNthHandler { + async fn execute( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _logs_root: &Path, + _services: &arc_workflows::handler::EngineServices, + ) -> Result { + let n = self + .counter + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + if n >= self.succeed_on { + Ok(Outcome::success()) + } else { + Ok(Outcome::fail("not yet ready")) + } + } +} + +/// Build a pipeline: start -> work -> (fail loop back to work, success to exit) +/// This creates a self-loop where work keeps retrying via edge routing. +fn circuit_breaker_self_loop_graph(signature_limit: Option) -> Graph { + let mut graph = make_graph_with_start_exit("CircuitBreakerSelfLoop"); + graph + .attrs + .insert("default_max_retry".to_string(), AttrValue::Integer(0)); + // High visit limit so the circuit breaker fires first + graph + .attrs + .insert("max_node_visits".to_string(), AttrValue::Integer(100)); + if let Some(limit) = signature_limit { + graph.attrs.insert( + "loop_restart_signature_limit".to_string(), + AttrValue::Integer(limit), + ); + } + + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("test_handler".to_string()), + ); + work.attrs + .insert("max_retries".to_string(), AttrValue::Integer(0)); + graph.nodes.insert("work".to_string(), work); + + graph.edges.push(Edge::new("start", "work")); + let mut fail_edge = Edge::new("work", "work"); + fail_edge.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=fail".to_string()), + ); + graph.edges.push(fail_edge); + let mut ok_edge = Edge::new("work", "exit"); + ok_edge.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=success".to_string()), + ); + graph.edges.push(ok_edge); + graph +} + +/// Build a pipeline: start -> work -> (fail: loop_restart to start, success: exit) +/// This uses loop_restart edges for full pipeline restarts. +fn circuit_breaker_restart_graph(signature_limit: Option) -> Graph { + let mut graph = make_graph_with_start_exit("CircuitBreakerRestart"); + graph + .attrs + .insert("default_max_retry".to_string(), AttrValue::Integer(0)); + graph + .attrs + .insert("max_node_visits".to_string(), AttrValue::Integer(100)); + if let Some(limit) = signature_limit { + graph.attrs.insert( + "loop_restart_signature_limit".to_string(), + AttrValue::Integer(limit), + ); + } + + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("test_handler".to_string()), + ); + work.attrs + .insert("max_retries".to_string(), AttrValue::Integer(0)); + graph.nodes.insert("work".to_string(), work); + + graph.edges.push(Edge::new("start", "work")); + let mut restart_edge = Edge::new("work", "start"); + restart_edge.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=fail".to_string()), + ); + restart_edge + .attrs + .insert("loop_restart".to_string(), AttrValue::Boolean(true)); + graph.edges.push(restart_edge); + let mut ok_edge = Edge::new("work", "exit"); + ok_edge.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=success".to_string()), + ); + graph.edges.push(ok_edge); + graph +} + +// --- E2E Test: normalize_failure_reason produces stable signatures --- + +#[test] +fn e2e_normalize_failure_reason_strips_variable_data() { + use arc_workflows::error::normalize_failure_reason; + + // Two error messages that differ only in line numbers and hex hashes + // should normalize to the same string. + let reason_a = "Error at line 42 in commit abc123def0: assertion failed"; + let reason_b = "Error at line 999 in commit deadbeef01: assertion failed"; + assert_eq!( + normalize_failure_reason(reason_a), + normalize_failure_reason(reason_b), + "errors differing only in line numbers and hashes should normalize identically" + ); + + // Different semantic errors should NOT normalize to the same string. + let reason_c = "syntax error in module alpha"; + let reason_d = "type mismatch in module beta"; + assert_ne!( + normalize_failure_reason(reason_c), + normalize_failure_reason(reason_d), + "semantically different errors should produce different normalized forms" + ); +} + +// --- E2E Test: FailureSignature composite key format --- + +#[test] +fn e2e_failure_signature_composite_key() { + use arc_workflows::error::{FailureClass, FailureSignature}; + + let sig = FailureSignature::new( + "verify", + FailureClass::Deterministic, + None, + Some("assertion failed at line 42"), + ); + let sig_str = sig.to_string(); + + // Verify format: node_id|failure_class|normalized_reason + assert!(sig_str.starts_with("verify|deterministic|")); + // Line number should be normalized away + assert!( + sig_str.contains(""), + "line numbers should be normalized: {sig_str}" + ); + assert!( + !sig_str.contains("42"), + "raw digits should be replaced: {sig_str}" + ); +} + +// --- E2E Test: signature_hint takes priority over failure_reason --- + +#[test] +fn e2e_failure_signature_hint_priority() { + use arc_workflows::error::{FailureClass, FailureSignature}; + + let sig = FailureSignature::new( + "build", + FailureClass::Deterministic, + Some("custom-key-abc"), + Some("raw error with line 123 and hash deadbeef"), + ); + + // The hint should be used, not the raw reason + assert_eq!(sig.to_string(), "build|deterministic|custom-key-abc"); +} + +// --- E2E Test: is_signature_tracked only for deterministic + structural --- + +#[test] +fn e2e_only_deterministic_and_structural_tracked() { + use arc_workflows::error::FailureClass; + + // These should be tracked + assert!(FailureClass::Deterministic.is_signature_tracked()); + assert!(FailureClass::Structural.is_signature_tracked()); + + // These should NOT be tracked (transient failures retry naturally) + assert!(!FailureClass::TransientInfra.is_signature_tracked()); + assert!(!FailureClass::BudgetExhausted.is_signature_tracked()); + assert!(!FailureClass::Canceled.is_signature_tracked()); + assert!(!FailureClass::CompilationLoop.is_signature_tracked()); +} + +// --- E2E Test: loop_restart_signature_limit graph attribute --- + +#[test] +fn e2e_loop_restart_signature_limit_from_graph_attr() { + let graph = circuit_breaker_self_loop_graph(Some(5)); + assert_eq!(graph.loop_restart_signature_limit(), 5); + + let graph_default = circuit_breaker_self_loop_graph(None); + assert_eq!(graph_default.loop_restart_signature_limit(), 3); +} + +// --- E2E Test: deterministic failure in self-loop triggers circuit breaker --- + +#[tokio::test] +async fn e2e_circuit_breaker_deterministic_self_loop() { + let dir = tempfile::tempdir().unwrap(); + let graph = circuit_breaker_self_loop_graph(Some(3)); + + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register( + "test_handler", + Box::new(DeterministicFailHandler::new( + "assertion failed in foo_test", + )), + ); + + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "e2e-circuit-breaker".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!(result.is_err(), "pipeline should abort, not loop forever"); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("deterministic failure cycle detected"), + "error should mention cycle detection, got: {err}" + ); + assert!( + err.contains("repeated 3 times"), + "error should mention the count, got: {err}" + ); + assert!( + err.contains("work|deterministic|"), + "error should include the signature, got: {err}" + ); +} + +// --- E2E Test: custom signature limit (5) --- + +#[tokio::test] +async fn e2e_circuit_breaker_custom_limit() { + let dir = tempfile::tempdir().unwrap(); + let graph = circuit_breaker_self_loop_graph(Some(5)); + + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register( + "test_handler", + Box::new(DeterministicFailHandler::new("same error every time")), + ); + + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "e2e-custom-limit".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!(result.is_err()); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("repeated 5 times"), + "should fire at limit=5, got: {err}" + ); +} + +// --- E2E Test: transient_infra failures do NOT trigger circuit breaker --- + +#[tokio::test] +async fn e2e_circuit_breaker_ignores_transient_failures() { + let dir = tempfile::tempdir().unwrap(); + let mut graph = circuit_breaker_self_loop_graph(Some(3)); + // Lower visit limit so the test terminates quickly via visit limit + graph + .attrs + .insert("max_node_visits".to_string(), AttrValue::Integer(6)); + + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("test_handler", Box::new(TransientInfraFailHandler)); + + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "e2e-transient-no-breaker".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!(result.is_err()); + let err = result.unwrap_err().to_string(); + // Should hit visit limit, NOT circuit breaker + assert!( + err.contains("exceeded max visit limit"), + "transient failures should not trigger circuit breaker, got: {err}" + ); +} + +// --- E2E Test: different failure reasons produce different signatures --- + +#[tokio::test] +async fn e2e_circuit_breaker_different_reasons_separate_counters() { + let dir = tempfile::tempdir().unwrap(); + let mut graph = circuit_breaker_self_loop_graph(Some(3)); + // With 10 unique reasons and limit=3, we can do up to 30 iterations before + // any single reason hits 3. But max_node_visits=8 will fire first. + graph + .attrs + .insert("max_node_visits".to_string(), AttrValue::Integer(8)); + + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register( + "test_handler", + Box::new(VaryingReasonFailHandler { + counter: std::sync::atomic::AtomicU32::new(0), + }), + ); + + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "e2e-varying-reasons".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!(result.is_err()); + let err = result.unwrap_err().to_string(); + // Should hit visit limit because each failure has a unique signature + assert!( + err.contains("exceeded max visit limit"), + "varying reasons should not trigger circuit breaker, got: {err}" + ); +} + +// --- E2E Test: loop_restart edge triggers circuit breaker --- + +#[tokio::test] +async fn e2e_circuit_breaker_loop_restart() { + let dir = tempfile::tempdir().unwrap(); + let graph = circuit_breaker_restart_graph(Some(3)); + + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register( + "test_handler", + Box::new(DeterministicFailHandler::new("verify step failed")), + ); + + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "e2e-restart-breaker".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!( + result.is_err(), + "pipeline should abort, not restart forever" + ); + let err = result.unwrap_err().to_string(); + // Either the loop or restart circuit breaker fires + assert!( + err.contains("failure cycle detected") || err.contains("circuit breaker"), + "a circuit breaker should fire on repeated restart, got: {err}" + ); +} + +// --- E2E Test: failure_signature stored in context (checkpoint verification) --- + +#[tokio::test] +async fn e2e_failure_signature_persisted_in_context() { + let dir = tempfile::tempdir().unwrap(); + // Pipeline: start -> work (fails once) -> exit + // Work fails but the edge routes to exit unconditionally. + let mut graph = make_graph_with_start_exit("SignatureContextTest"); + graph + .attrs + .insert("default_max_retry".to_string(), AttrValue::Integer(0)); + + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("test_handler".to_string()), + ); + work.attrs + .insert("max_retries".to_string(), AttrValue::Integer(0)); + graph.nodes.insert("work".to_string(), work); + + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register( + "test_handler", + Box::new(DeterministicFailHandler::new("test assertion failed")), + ); + + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "e2e-sig-context".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let outcome = engine.run(&graph, &config).await.unwrap(); + // Pipeline reaches exit (terminal), last completed node is "work" (Fail). + // The engine doesn't execute exit handlers, just breaks on terminal nodes. + assert_eq!(outcome.status, StageStatus::Fail); + + // Verify checkpoint has failure_signature in context + let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + let sig_value = cp + .context_values + .get("failure_signature") + .expect("failure_signature should be in context"); + let sig_str = sig_value.as_str().unwrap(); + assert!( + sig_str.contains("work|deterministic|"), + "signature should contain node_id|class|, got: {sig_str}" + ); + assert!( + sig_str.contains("test assertion failed"), + "signature should contain normalized reason, got: {sig_str}" + ); +} + +// --- E2E Test: failure_signature hint from handler overrides raw reason --- + +#[tokio::test] +async fn e2e_failure_signature_hint_overrides_reason_in_context() { + let dir = tempfile::tempdir().unwrap(); + let mut graph = make_graph_with_start_exit("SignatureHintTest"); + graph + .attrs + .insert("default_max_retry".to_string(), AttrValue::Integer(0)); + + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("hint_handler".to_string()), + ); + work.attrs + .insert("max_retries".to_string(), AttrValue::Integer(0)); + graph.nodes.insert("work".to_string(), work); + + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("hint_handler", Box::new(SignatureHintHandler)); + + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "e2e-sig-hint".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let _outcome = engine.run(&graph, &config).await.unwrap(); + + let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + let sig_str = cp + .context_values + .get("failure_signature") + .and_then(|v| v.as_str()) + .expect("failure_signature should be set"); + // The hint "custom-grouping-key" should be used, not the raw reason + assert!( + sig_str.contains("custom-grouping-key"), + "hint should override raw reason, got: {sig_str}" + ); + // Raw reason contained line numbers and hex — verify they are NOT in the signature + assert!( + !sig_str.contains("42"), + "raw reason details should not leak through, got: {sig_str}" + ); +} + +// --- E2E Test: signature maps persisted in checkpoint and survive save/load --- + +#[tokio::test] +async fn e2e_signature_maps_persist_in_checkpoint() { + let dir = tempfile::tempdir().unwrap(); + // Pipeline where work fails twice then we check the checkpoint + let graph = circuit_breaker_self_loop_graph(Some(5)); + + // Use a handler that succeeds on the 3rd call (0-indexed), so we get + // exactly 3 failures at the work node before succeeding on the 4th visit. + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register( + "test_handler", + Box::new(SucceedOnNthHandler { + succeed_on: 3, + counter: std::sync::atomic::AtomicU32::new(0), + }), + ); + + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "e2e-sig-persist".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let outcome = engine.run(&graph, &config).await.unwrap(); + assert_eq!(outcome.status, StageStatus::Success); + + // Load checkpoint and verify signature maps + let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + // The pipeline had 3 deterministic failures at "work" before succeeding. + // loop_failure_signatures should have recorded them. + assert!( + !cp.loop_failure_signatures.is_empty(), + "loop_failure_signatures should have entries after deterministic failures" + ); + // Verify the signature key format + let (sig, count) = cp.loop_failure_signatures.iter().next().unwrap(); + assert!( + sig.to_string().starts_with("work|deterministic|"), + "signature key should have correct format, got: {sig}" + ); + assert_eq!( + *count, 3, + "should have recorded exactly 3 failures before success" + ); +} + +// --- E2E Test: checkpoint backward compat (old checkpoints without signature fields) --- + +#[test] +fn e2e_checkpoint_backward_compat_no_signatures() { + // Simulate loading a checkpoint saved before signature fields existed + let json = serde_json::json!({ + "timestamp": "2025-06-01T00:00:00Z", + "current_node": "work", + "completed_nodes": ["start", "work"], + "node_retries": {}, + "context_values": {"goal": "test"}, + "logs": ["some log entry"], + "node_outcomes": {} + }); + + let cp: Checkpoint = serde_json::from_value(json).expect("should deserialize old checkpoint"); + assert!(cp.loop_failure_signatures.is_empty()); + assert!(cp.restart_failure_signatures.is_empty()); + assert_eq!(cp.current_node, "work"); +} + +// --- E2E Test: checkpoint with signatures round-trips through save/load --- + +#[test] +fn e2e_checkpoint_signatures_roundtrip() { + use arc_workflows::error::{FailureClass, FailureSignature}; + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cp.json"); + + let ctx = Context::new(); + ctx.set("goal", serde_json::json!("test roundtrip")); + + let mut loop_sigs = std::collections::HashMap::new(); + let sig1 = FailureSignature::new( + "verify", + FailureClass::Deterministic, + None, + Some("assertion failed"), + ); + loop_sigs.insert(sig1.clone(), 2usize); + + let mut restart_sigs = std::collections::HashMap::new(); + let sig2 = FailureSignature::new( + "build", + FailureClass::Structural, + None, + Some("scope violation"), + ); + restart_sigs.insert(sig2.clone(), 1usize); + + let cp = Checkpoint::from_context( + &ctx, + "verify", + vec!["start".to_string(), "verify".to_string()], + std::collections::HashMap::new(), + std::collections::HashMap::new(), + None, + loop_sigs, + restart_sigs, + ); + cp.save(&path).unwrap(); + + let loaded = Checkpoint::load(&path).unwrap(); + assert_eq!(loaded.loop_failure_signatures.len(), 1); + assert_eq!(loaded.restart_failure_signatures.len(), 1); + assert_eq!(loaded.loop_failure_signatures.get(&sig1), Some(&2)); + assert_eq!(loaded.restart_failure_signatures.get(&sig2), Some(&1)); +} + +// --- E2E Test: pipeline events are emitted before circuit breaker aborts --- + +#[tokio::test] +async fn e2e_circuit_breaker_emits_events_before_abort() { + let dir = tempfile::tempdir().unwrap(); + let graph = circuit_breaker_self_loop_graph(Some(3)); + + let mut emitter = EventEmitter::new(); + let events = collect_events(&mut emitter); + + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register( + "test_handler", + Box::new(DeterministicFailHandler::new("assertion failed")), + ); + + let engine = PipelineEngine::new(registry, Arc::new(emitter), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "e2e-events".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!(result.is_err()); + + let events = events.lock().unwrap(); + // Should have at least PipelineStarted and some StageFailed/StageCompleted events + let has_pipeline_started = events + .iter() + .any(|e| matches!(e, PipelineEvent::PipelineStarted { .. })); + assert!( + has_pipeline_started, + "PipelineStarted event should be emitted" + ); + + // Verify we got stage events for the failing work node. + // The circuit breaker fires when count reaches the limit (3) *before* + // the stage event for that iteration is emitted, so we see limit-1 events. + let stage_failed_count = events + .iter() + .filter(|e| matches!(e, PipelineEvent::StageFailed { name, .. } if name == "work")) + .count(); + let stage_completed_count = events + .iter() + .filter(|e| matches!(e, PipelineEvent::StageCompleted { name, .. } if name == "work")) + .count(); + let total_work_events = stage_completed_count + stage_failed_count; + // With limit=3, the breaker fires on the 3rd failure before its event is emitted. + // So we get 2 events (for failures 1 and 2). + assert!( + total_work_events >= 2, + "should have at least 2 stage events before circuit breaker fires, got: {total_work_events}" + ); +} + +// --- E2E Test: success resets to success path, but signatures are preserved --- + +#[tokio::test] +async fn e2e_circuit_breaker_does_not_fire_below_limit() { + let dir = tempfile::tempdir().unwrap(); + let graph = circuit_breaker_self_loop_graph(Some(5)); + + // Handler that fails 4 times (below limit of 5) then succeeds + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register( + "test_handler", + Box::new(SucceedOnNthHandler { + succeed_on: 4, + counter: std::sync::atomic::AtomicU32::new(0), + }), + ); + + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "e2e-below-limit".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let outcome = engine.run(&graph, &config).await.unwrap(); + assert_eq!( + outcome.status, + StageStatus::Success, + "pipeline should succeed when failures stay below limit" + ); + + // Verify signatures were tracked but didn't trigger abort + let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap(); + let total_failures: usize = cp.loop_failure_signatures.values().sum(); + assert_eq!( + total_failures, 4, + "should have tracked 4 failures in signatures" + ); +} + +// --- E2E Test: multi-stage pipeline with impl/verify cycle detection --- + +#[tokio::test] +async fn e2e_circuit_breaker_multi_stage_impl_verify_cycle() { + // Pipeline: start -> impl (succeeds) -> verify (fails) -> impl -> verify -> ... + // The verify node always fails with the same deterministic reason. + // Circuit breaker should detect the verify failure cycling. + let dir = tempfile::tempdir().unwrap(); + let mut graph = make_graph_with_start_exit("ImplVerifyCycle"); + graph + .attrs + .insert("default_max_retry".to_string(), AttrValue::Integer(0)); + graph + .attrs + .insert("max_node_visits".to_string(), AttrValue::Integer(100)); + graph.attrs.insert( + "loop_restart_signature_limit".to_string(), + AttrValue::Integer(3), + ); + + let mut impl_node = Node::new("impl"); + impl_node.attrs.insert( + "type".to_string(), + AttrValue::String("success_handler".to_string()), + ); + graph.nodes.insert("impl".to_string(), impl_node); + + let mut verify_node = Node::new("verify"); + verify_node.attrs.insert( + "type".to_string(), + AttrValue::String("fail_handler".to_string()), + ); + verify_node + .attrs + .insert("max_retries".to_string(), AttrValue::Integer(0)); + graph.nodes.insert("verify".to_string(), verify_node); + + graph.edges.push(Edge::new("start", "impl")); + graph.edges.push(Edge::new("impl", "verify")); + // verify fail -> back to impl + let mut fail_edge = Edge::new("verify", "impl"); + fail_edge.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=fail".to_string()), + ); + graph.edges.push(fail_edge); + // verify success -> exit (never taken) + let mut ok_edge = Edge::new("verify", "exit"); + ok_edge.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=success".to_string()), + ); + graph.edges.push(ok_edge); + + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + registry.register("success_handler", Box::new(StartHandler)); // StartHandler returns success + registry.register( + "fail_handler", + Box::new(DeterministicFailHandler::new( + "test assertion: expected 42, got 0", + )), + ); + + let engine = PipelineEngine::new(registry, Arc::new(EventEmitter::new()), local_env()); + let config = RunConfig { + logs_root: dir.path().to_path_buf(), + cancel_token: None, + dry_run: false, + run_id: "e2e-impl-verify-cycle".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!( + result.is_err(), + "should detect impl/verify cycle, not loop forever" + ); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("deterministic failure cycle detected"), + "should identify deterministic failure cycle, got: {err}" + ); + assert!( + err.contains("verify|deterministic|"), + "signature should name the verify node, got: {err}" + ); +} + // Daytona parallel git branching test is in daytona_integration.rs