diff --git a/crates/arc-workflows/src/engine.rs b/crates/arc-workflows/src/engine.rs index 504fbec3f..0b4375449 100644 --- a/crates/arc-workflows/src/engine.rs +++ b/crates/arc-workflows/src/engine.rs @@ -1505,6 +1505,16 @@ impl PipelineEngine { incoming_edge = Some(edge); // Gap #6: Handle loop_restart by recursively running from the target if edge.loop_restart() { + // Guard: only transient_infra failures may loop_restart (matches Kilroy) + if let Some(fc) = outcome_failure_class { + if fc != FailureClass::TransientInfra { + return Err(ArcError::Engine(format!( + "loop_restart blocked: failure_class={fc} (requires transient_infra), node={}, failure_reason={}", + node.id, + outcome.failure_reason.as_deref().unwrap_or("none"), + ))); + } + } // Circuit breaker: check restart failure signatures if let Some(ref sig) = failure_sig { let count = loop_state @@ -3815,10 +3825,12 @@ mod tests { 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 + // The loop_restart guard blocks non-transient_infra failures immediately assert!( - err.contains("failure cycle detected") || err.contains("circuit breaker"), - "expected circuit breaker error, got: {err}" + err.contains("loop_restart blocked") + || err.contains("failure cycle detected") + || err.contains("circuit breaker"), + "expected loop_restart guard or circuit breaker error, got: {err}" ); } diff --git a/crates/arc-workflows/src/error.rs b/crates/arc-workflows/src/error.rs index 866ca917a..b80e0a6a7 100644 --- a/crates/arc-workflows/src/error.rs +++ b/crates/arc-workflows/src/error.rs @@ -238,7 +238,7 @@ pub fn normalize_failure_reason(reason: &str) -> String { let s = WHITESPACE_RE.replace_all(&s, " "); let s = s.trim(); if s.len() > 240 { - s[..240].to_string() + s[..s.floor_char_boundary(240)].to_string() } else { s.to_string() } @@ -1313,6 +1313,25 @@ mod tests { assert_eq!(result.len(), 240); } + #[test] + fn normalize_truncation_respects_utf8_boundaries() { + // Build a string of 2-byte chars ("é" is 2 bytes in UTF-8) that crosses + // the 240 byte boundary mid-character. + let input = "é".repeat(200); // 400 bytes, each char is 2 bytes + let result = normalize_failure_reason(&input); + assert!(result.len() <= 240); + // Must be valid UTF-8 (String guarantees this, but verify length is even + // since every char is 2 bytes) + assert_eq!(result.len() % 2, 0); + + // Also test with a mix: 239 ASCII bytes + a 2-byte char + let input2 = format!("{}{}", "a".repeat(239), "é"); + let result2 = normalize_failure_reason(&input2); + assert!(result2.len() <= 240); + // Should truncate to 239 (dropping the 2-byte char that would push to 241) + assert_eq!(result2.len(), 239); + } + #[test] fn normalize_combined_example() { assert_eq!( diff --git a/crates/arc-workflows/tests/integration.rs b/crates/arc-workflows/tests/integration.rs index bf3f4ca5c..040ee5563 100644 --- a/crates/arc-workflows/tests/integration.rs +++ b/crates/arc-workflows/tests/integration.rs @@ -1172,7 +1172,12 @@ impl Handler for CounterHandler { .call_count .fetch_add(1, std::sync::atomic::Ordering::SeqCst); if count == 0 { - Ok(Outcome::fail("first call fails")) + let mut outcome = Outcome::fail("first call fails"); + outcome.context_updates.insert( + "failure_class".to_string(), + serde_json::json!("transient_infra"), + ); + Ok(outcome) } else { Ok(Outcome::success()) } @@ -9947,10 +9952,12 @@ async fn e2e_circuit_breaker_loop_restart() { "pipeline should abort, not restart forever" ); let err = result.unwrap_err().to_string(); - // Either the loop or restart circuit breaker fires + // The loop_restart guard blocks non-transient_infra failures immediately assert!( - err.contains("failure cycle detected") || err.contains("circuit breaker"), - "a circuit breaker should fire on repeated restart, got: {err}" + err.contains("loop_restart blocked") + || err.contains("failure cycle detected") + || err.contains("circuit breaker"), + "expected loop_restart guard or circuit breaker error, got: {err}" ); } @@ -10408,4 +10415,260 @@ async fn e2e_circuit_breaker_multi_stage_impl_verify_cycle() { ); } +// --- E2E Tests: loop_restart guard (only transient_infra may restart) --- + +/// Handler that fails with an explicit failure_class hint and succeeds on the Nth call. +struct ClassifiedFailHandler { + failure_class: &'static str, + succeed_on: u32, + counter: std::sync::atomic::AtomicU32, +} + +impl ClassifiedFailHandler { + fn always(failure_class: &'static str) -> Self { + Self { + failure_class, + succeed_on: u32::MAX, + counter: std::sync::atomic::AtomicU32::new(0), + } + } + + fn succeed_on(failure_class: &'static str, n: u32) -> Self { + Self { + failure_class, + succeed_on: n, + counter: std::sync::atomic::AtomicU32::new(0), + } + } +} + +#[async_trait::async_trait] +impl Handler for ClassifiedFailHandler { + 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 { + return Ok(Outcome::success()); + } + let mut outcome = Outcome::fail("classified failure"); + outcome.context_updates.insert( + "failure_class".to_string(), + serde_json::json!(self.failure_class), + ); + Ok(outcome) + } +} + +#[tokio::test] +async fn e2e_loop_restart_blocked_for_deterministic_failure() { + let dir = tempfile::tempdir().unwrap(); + let graph = circuit_breaker_restart_graph(Some(10)); + + 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(ClassifiedFailHandler::always("deterministic")), + ); + + 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-blocked-det".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!(result.is_err(), "deterministic failure should not loop_restart"); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("loop_restart blocked"), + "expected loop_restart blocked error, got: {err}" + ); +} + +#[tokio::test] +async fn e2e_loop_restart_blocked_for_structural_failure() { + let dir = tempfile::tempdir().unwrap(); + let graph = circuit_breaker_restart_graph(Some(10)); + + 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(ClassifiedFailHandler::always("structural")), + ); + + 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-blocked-struct".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!(result.is_err(), "structural failure should not loop_restart"); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("loop_restart blocked"), + "expected loop_restart blocked error, got: {err}" + ); +} + +#[tokio::test] +async fn e2e_loop_restart_blocked_for_budget_exhausted_failure() { + let dir = tempfile::tempdir().unwrap(); + let graph = circuit_breaker_restart_graph(Some(10)); + + 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(ClassifiedFailHandler::always("budget_exhausted")), + ); + + 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-blocked-budget".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!(result.is_err(), "budget_exhausted failure should not loop_restart"); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("loop_restart blocked"), + "expected loop_restart blocked error, got: {err}" + ); +} + +#[tokio::test] +async fn e2e_loop_restart_blocked_for_canceled_failure() { + let dir = tempfile::tempdir().unwrap(); + let graph = circuit_breaker_restart_graph(Some(10)); + + 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(ClassifiedFailHandler::always("canceled")), + ); + + 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-blocked-canceled".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!(result.is_err(), "canceled failure should not loop_restart"); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("loop_restart blocked"), + "expected loop_restart blocked error, got: {err}" + ); +} + +#[tokio::test] +async fn e2e_loop_restart_blocked_for_compilation_loop_failure() { + let dir = tempfile::tempdir().unwrap(); + let graph = circuit_breaker_restart_graph(Some(10)); + + 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(ClassifiedFailHandler::always("compilation_loop")), + ); + + 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-blocked-comploop".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!(result.is_err(), "compilation_loop failure should not loop_restart"); + let err = result.unwrap_err().to_string(); + assert!( + err.contains("loop_restart blocked"), + "expected loop_restart blocked error, got: {err}" + ); +} + +#[tokio::test] +async fn e2e_loop_restart_allowed_for_transient_infra() { + let dir = tempfile::tempdir().unwrap(); + let graph = circuit_breaker_restart_graph(Some(10)); + + let mut registry = HandlerRegistry::new(Box::new(StartHandler)); + registry.register("start", Box::new(StartHandler)); + registry.register("exit", Box::new(ExitHandler)); + // Fails with transient_infra on first call, succeeds on second + registry.register( + "test_handler", + Box::new(ClassifiedFailHandler::succeed_on("transient_infra", 1)), + ); + + 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-allowed-transient".into(), + git_checkpoint: None, + base_sha: None, + run_branch: None, + meta_branch: None, + }; + + let result = engine.run(&graph, &config).await; + assert!( + result.is_ok(), + "transient_infra failure should be allowed to loop_restart, got: {:?}", + result.unwrap_err() + ); +} + // Daytona parallel git branching test is in daytona_integration.rs