Block non-transient_infra failures from loop_restart and fix UTF-8 truncation

Guard loop_restart edges to only allow transient_infra failures, matching
Kilroy's behavior. Non-transient failures (deterministic, structural,
budget_exhausted, canceled, compilation_loop) are now blocked immediately
instead of getting restart attempts before the circuit breaker fires.

Fix normalize_failure_reason truncation to use floor_char_boundary(240)
instead of a raw byte slice, preventing panics on multi-byte UTF-8 chars.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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}"
         );
     }
 

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!(

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<Outcome, ArcError> {
+        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