Add stall watchdog to detect hung pipeline handlers

A background tokio task polls EventEmitter.last_event_at(). When idle
time exceeds the graph-level stall_timeout (default 600s), it cancels
a CancellationToken that races against execute_with_retry via
tokio::select!, dropping the hung handler future.

- EventEmitter: AtomicI64 last_event_at field, touch() to seed, emit()
  auto-updates
- Graph::stall_timeout(): reads Duration attr, defaults 600s, None for 0
- StallWatchdogTimeout event variant with warn-level trace()
- Engine: watchdog spawn before main loop, select! at handler call,
  shutdown after loop
- CLI: format arms for summary and detail views
- Unit tests for emitter, graph accessor, event serialization, and
  engine watchdog behavior (hung, keepalive, disabled)
- E2e integration tests: DOT-parsed pipelines with 100-200ms stall
  timeouts verifying trigger, keepalive, disable, and timing

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

crates/arc-workflows/src/cli/mod.rs

@@ -559,6 +559,9 @@ pub fn format_event_summary(event: &PipelineEvent, styles: &Styles) -> String {
             };
             format!("[SETUP_FAILED] index={index} command=\"{command}\" exit_code={exit_code} stderr=\"{truncated}\"")
         }
+        PipelineEvent::StallWatchdogTimeout { node, idle_seconds } => {
+            format!("[STALL_WATCHDOG_TIMEOUT] node={node} idle_seconds={idle_seconds}")
+        }
     };
     format!("{dim}{body}{reset}", dim = styles.dim, reset = styles.reset)
 }
@@ -904,6 +907,9 @@ pub fn format_event_detail(event: &PipelineEvent, styles: &Styles) -> String {
         PipelineEvent::SetupFailed { command, index, exit_code, stderr } => {
             format!("{d}── SETUP_FAILED ─────────────────────────────{r}\n  {d}index:{r}     {index}\n  {d}command:{r}   {command}\n  {d}exit_code:{r} {exit_code}\n  {d}stderr:{r}    {stderr}\n")
         }
+        PipelineEvent::StallWatchdogTimeout { node, idle_seconds } => {
+            format!("{d}── STALL_WATCHDOG_TIMEOUT ────────────────────{r}\n  {d}node:{r}         {node}\n  {d}idle_seconds:{r} {idle_seconds}\n")
+        }
     }
 }
 

crates/arc-workflows/src/engine.rs

@@ -9,6 +9,7 @@ use arc_agent::ExecutionEnvironment;
 use chrono::Utc;
 use futures::FutureExt;
 use rand::Rng;
+use tokio_util::sync::CancellationToken;
 
 use arc_git_storage::trailerlink::{self, Trailer};
 
@@ -1096,6 +1097,44 @@ impl PipelineEngine {
             );
         }
 
+        // Stall watchdog: background task that cancels `stall_token` when no events
+        // have been emitted for longer than `stall_timeout`.
+        let stall_token = graph.stall_timeout().map(|timeout| {
+            let token = CancellationToken::new();
+            let shutdown = CancellationToken::new();
+            let check_interval = (timeout / 10)
+                .max(std::time::Duration::from_millis(50))
+                .min(std::time::Duration::from_secs(5));
+            self.services.emitter.touch();
+            let emitter = Arc::clone(&self.services.emitter);
+            let cancel = token.clone();
+            let stop = shutdown.clone();
+            tracing::debug!(
+                stall_timeout_ms = timeout.as_millis() as u64,
+                check_interval_ms = check_interval.as_millis() as u64,
+                "Stall watchdog started"
+            );
+            tokio::spawn(async move {
+                loop {
+                    tokio::select! {
+                        () = stop.cancelled() => break,
+                        () = tokio::time::sleep(check_interval) => {
+                            let last = emitter.last_event_at();
+                            let now = std::time::SystemTime::now()
+                                .duration_since(std::time::UNIX_EPOCH)
+                                .unwrap_or_default()
+                                .as_millis() as i64;
+                            if now - last >= timeout.as_millis() as i64 {
+                                cancel.cancel();
+                                break;
+                            }
+                        }
+                    }
+                }
+            });
+            (token, shutdown)
+        });
+
         loop {
             // Check for cancellation before processing each node
             if let Some(ref token) = config.cancel_token {
@@ -1195,17 +1234,28 @@ impl PipelineEngine {
             }
             let stage_start = Instant::now();
 
-            let (mut outcome, attempts_used) = self
-                .execute_with_retry(
-                    node,
-                    &context,
-                    graph,
-                    &config.logs_root,
-                    &retry_policy,
-                    stage_index,
-                    visit,
-                )
-                .await?;
+            let (mut outcome, attempts_used) = if let Some((ref token, _)) = stall_token {
+                tokio::select! {
+                    result = self.execute_with_retry(
+                        node, &context, graph, &config.logs_root, &retry_policy, stage_index, visit,
+                    ) => result?,
+                    () = token.cancelled() => {
+                        let idle_secs = graph.stall_timeout().map_or(0, |d| d.as_secs());
+                        self.services.emitter.emit(&PipelineEvent::StallWatchdogTimeout {
+                            node: node.id.clone(),
+                            idle_seconds: idle_secs,
+                        });
+                        return Err(ArcError::Engine(format!(
+                            "stall watchdog: node \"{}\" had no activity for {}s",
+                            node.id, idle_secs,
+                        )));
+                    }
+                }
+            } else {
+                self.execute_with_retry(
+                    node, &context, graph, &config.logs_root, &retry_policy, stage_index, visit,
+                ).await?
+            };
             // Gap #5: Track retry count per node
             node_retries.insert(node.id.clone(), attempts_used);
             context.set(
@@ -1542,6 +1592,11 @@ impl PipelineEngine {
             }
         }
 
+        // Shut down stall watchdog
+        if let Some((_, ref shutdown)) = stall_token {
+            shutdown.cancel();
+        }
+
         let duration_ms = millis_u64(run_start.elapsed());
         let total_cost: Option<f64> = {
             let sum: f64 = node_outcomes
@@ -3829,6 +3884,207 @@ mod tests {
         );
     }
 
+    /// Handler that emits events every `interval_ms` for `total_ms`, then succeeds.
+    struct EmittingHandler {
+        interval_ms: u64,
+        total_ms: u64,
+    }
+
+    #[async_trait]
+    impl HandlerTrait for EmittingHandler {
+        async fn execute(
+            &self,
+            node: &Node,
+            _context: &Context,
+            _graph: &Graph,
+            _logs_root: &Path,
+            services: &crate::handler::EngineServices,
+        ) -> std::result::Result<Outcome, ArcError> {
+            let start = Instant::now();
+            while start.elapsed() < Duration::from_millis(self.total_ms) {
+                tokio::time::sleep(Duration::from_millis(self.interval_ms)).await;
+                services.emitter.emit(&PipelineEvent::Prompt {
+                    stage: node.id.clone(),
+                    text: "keepalive".to_string(),
+                });
+            }
+            Ok(Outcome::success())
+        }
+    }
+
+    #[tokio::test]
+    async fn stall_watchdog_triggers_on_hung_handler() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut g = Graph::new("stall_test");
+        g.attrs
+            .insert("goal".to_string(), AttrValue::String("test".to_string()));
+        g.attrs.insert(
+            "stall_timeout".to_string(),
+            AttrValue::Duration(Duration::from_millis(200)),
+        );
+        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("slow".to_string()),
+        );
+        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("slow", Box::new(SlowHandler { sleep_ms: 60_000 }));
+        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("stall watchdog"),
+            "expected stall watchdog error, got: {err}"
+        );
+    }
+
+    #[tokio::test]
+    async fn stall_watchdog_active_handler_resets_timer() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut g = Graph::new("stall_active_test");
+        g.attrs
+            .insert("goal".to_string(), AttrValue::String("test".to_string()));
+        g.attrs.insert(
+            "stall_timeout".to_string(),
+            AttrValue::Duration(Duration::from_millis(200)),
+        );
+        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("emitting".to_string()),
+        );
+        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(
+            "emitting",
+            Box::new(EmittingHandler {
+                interval_ms: 100,
+                total_ms: 500,
+            }),
+        );
+        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();
+        assert_eq!(outcome.status, StageStatus::Success);
+    }
+
+    #[tokio::test]
+    async fn stall_watchdog_disabled_when_zero() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut g = Graph::new("stall_disabled_test");
+        g.attrs
+            .insert("goal".to_string(), AttrValue::String("test".to_string()));
+        g.attrs.insert(
+            "stall_timeout".to_string(),
+            AttrValue::Duration(Duration::ZERO),
+        );
+        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("slow".to_string()),
+        );
+        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("slow", Box::new(SlowHandler { sleep_ms: 200 }));
+        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();
+        assert_eq!(outcome.status, StageStatus::Success);
+    }
+
     #[tokio::test]
     async fn failure_signature_stored_in_context() {
         let dir = tempfile::tempdir().unwrap();

crates/arc-workflows/src/event.rs

@@ -1,3 +1,5 @@
+use std::sync::atomic::{AtomicI64, Ordering};
+
 use serde::{Deserialize, Serialize};
 
 use crate::outcome::StageUsage;
@@ -171,6 +173,10 @@ pub enum PipelineEvent {
         exit_code: i32,
         stderr: String,
     },
+    StallWatchdogTimeout {
+        node: String,
+        idle_seconds: u64,
+    },
 }
 
 impl PipelineEvent {
@@ -422,22 +428,39 @@ impl PipelineEvent {
             } => {
                 error!(command, index, exit_code, "Setup command failed");
             }
+            Self::StallWatchdogTimeout {
+                node,
+                idle_seconds,
+            } => {
+                warn!(node, idle_seconds, "Stall watchdog timeout");
+            }
         }
     }
 }
 
+/// Current time as epoch milliseconds.
+fn epoch_millis() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as i64
+}
+
 /// Listener callback type for pipeline events.
 type EventListener = Box<dyn Fn(&PipelineEvent) + Send + Sync>;
 
 /// Callback-based event emitter for pipeline events.
 pub struct EventEmitter {
     listeners: Vec<EventListener>,
+    /// Epoch milliseconds of the last `emit()` or `touch()` call. 0 until first event.
+    last_event_at: AtomicI64,
 }
 
 impl std::fmt::Debug for EventEmitter {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("EventEmitter")
             .field("listener_count", &self.listeners.len())
+            .field("last_event_at", &self.last_event_at.load(Ordering::Relaxed))
             .finish()
     }
 }
@@ -453,6 +476,7 @@ impl EventEmitter {
     pub fn new() -> Self {
         Self {
             listeners: Vec::new(),
+            last_event_at: AtomicI64::new(0),
         }
     }
 
@@ -461,11 +485,23 @@ impl EventEmitter {
     }
 
     pub fn emit(&self, event: &PipelineEvent) {
+        self.last_event_at.store(epoch_millis(), Ordering::Relaxed);
         event.trace();
         for listener in &self.listeners {
             listener(event);
         }
     }
+
+    /// Returns the epoch milliseconds of the last `emit()` or `touch()` call.
+    /// Returns 0 if neither has been called.
+    pub fn last_event_at(&self) -> i64 {
+        self.last_event_at.load(Ordering::Relaxed)
+    }
+
+    /// Manually update the last-event timestamp (e.g. to seed the watchdog at pipeline start).
+    pub fn touch(&self) {
+        self.last_event_at.store(epoch_millis(), Ordering::Relaxed);
+    }
 }
 
 #[cfg(test)]
@@ -951,6 +987,51 @@ mod tests {
         assert!(matches!(deserialized, PipelineEvent::ExecutionEnv { .. }));
     }
 
+    #[test]
+    fn emitter_last_event_at_initially_zero() {
+        let emitter = EventEmitter::new();
+        assert_eq!(emitter.last_event_at(), 0);
+    }
+
+    #[test]
+    fn emitter_last_event_at_updates_after_emit() {
+        let emitter = EventEmitter::new();
+        assert_eq!(emitter.last_event_at(), 0);
+        emitter.emit(&PipelineEvent::PipelineStarted {
+            name: "test".to_string(),
+            run_id: "1".to_string(),
+            base_sha: None,
+            run_branch: None,
+            worktree_dir: None,
+        });
+        assert!(emitter.last_event_at() > 0);
+    }
+
+    #[test]
+    fn emitter_touch_updates_last_event_at() {
+        let emitter = EventEmitter::new();
+        assert_eq!(emitter.last_event_at(), 0);
+        emitter.touch();
+        assert!(emitter.last_event_at() > 0);
+    }
+
+    #[test]
+    fn stall_watchdog_timeout_serialization() {
+        let event = PipelineEvent::StallWatchdogTimeout {
+            node: "work".to_string(),
+            idle_seconds: 600,
+        };
+        let json = serde_json::to_string(&event).unwrap();
+        assert!(json.contains("StallWatchdogTimeout"));
+        assert!(json.contains("\"node\":\"work\""));
+        assert!(json.contains("\"idle_seconds\":600"));
+
+        let deserialized: PipelineEvent = serde_json::from_str(&json).unwrap();
+        assert!(
+            matches!(deserialized, PipelineEvent::StallWatchdogTimeout { node, idle_seconds } if node == "work" && idle_seconds == 600)
+        );
+    }
+
     #[test]
     fn setup_events_serialization() {
         let events = vec![

crates/arc-workflows/src/graph/types.rs

@@ -429,6 +429,15 @@ impl Graph {
             .map_or(3, |v| v as usize)
     }
 
+    /// Graph-level `stall_timeout`. Defaults to 600s. Returns `None` when set to zero (disabled).
+    pub fn stall_timeout(&self) -> Option<Duration> {
+        match self.attrs.get("stall_timeout").and_then(AttrValue::as_duration) {
+            Some(d) if d.is_zero() => None,
+            Some(d) => Some(d),
+            None => Some(Duration::from_secs(600)),
+        }
+    }
+
     /// Graph-level `max_node_visits` (default 0 = disabled).
     pub fn max_node_visits(&self) -> u64 {
         self.attrs
@@ -695,6 +704,32 @@ mod tests {
         assert!(g.find_start_node().is_none());
     }
 
+    #[test]
+    fn graph_stall_timeout_default() {
+        let g = Graph::new("empty");
+        assert_eq!(g.stall_timeout(), Some(Duration::from_secs(600)));
+    }
+
+    #[test]
+    fn graph_stall_timeout_set() {
+        let mut g = Graph::new("test");
+        g.attrs.insert(
+            "stall_timeout".to_string(),
+            AttrValue::Duration(Duration::from_millis(200)),
+        );
+        assert_eq!(g.stall_timeout(), Some(Duration::from_millis(200)));
+    }
+
+    #[test]
+    fn graph_stall_timeout_zero_disables() {
+        let mut g = Graph::new("test");
+        g.attrs.insert(
+            "stall_timeout".to_string(),
+            AttrValue::Duration(Duration::ZERO),
+        );
+        assert_eq!(g.stall_timeout(), None);
+    }
+
     #[test]
     fn graph_max_node_visits_default() {
         let g = Graph::new("empty");

crates/arc-workflows/tests/integration.rs

@@ -10671,4 +10671,262 @@ async fn e2e_loop_restart_allowed_for_transient_infra() {
     );
 }
 
+// ---------------------------------------------------------------------------
+// Stall watchdog e2e tests
+// ---------------------------------------------------------------------------
+
+/// Handler that sleeps forever (for stall watchdog testing).
+struct HangingHandler;
+
+#[async_trait::async_trait]
+impl Handler for HangingHandler {
+    async fn execute(
+        &self,
+        _node: &Node,
+        _context: &Context,
+        _graph: &Graph,
+        _logs_root: &Path,
+        _services: &arc_workflows::handler::EngineServices,
+    ) -> Result<Outcome, ArcError> {
+        tokio::time::sleep(std::time::Duration::from_secs(60)).await;
+        Ok(Outcome::success())
+    }
+}
+
+/// Handler that emits keepalive events periodically, then succeeds.
+struct KeepaliveHandler {
+    interval_ms: u64,
+    total_ms: u64,
+}
+
+#[async_trait::async_trait]
+impl Handler for KeepaliveHandler {
+    async fn execute(
+        &self,
+        node: &Node,
+        _context: &Context,
+        _graph: &Graph,
+        _logs_root: &Path,
+        services: &arc_workflows::handler::EngineServices,
+    ) -> Result<Outcome, ArcError> {
+        let start = std::time::Instant::now();
+        while start.elapsed() < std::time::Duration::from_millis(self.total_ms) {
+            tokio::time::sleep(std::time::Duration::from_millis(self.interval_ms)).await;
+            services.emitter.emit(&PipelineEvent::Prompt {
+                stage: node.id.clone(),
+                text: "keepalive".to_string(),
+            });
+        }
+        Ok(Outcome::success())
+    }
+}
+
+#[tokio::test]
+async fn e2e_stall_watchdog_triggers_from_dot_parsed_pipeline() {
+    // Parse a DOT graph with stall_timeout set to 200ms
+    let dot = r#"digraph StallTest {
+        graph [goal="Test stall watchdog", stall_timeout="200ms", default_max_retry=0]
+        start [shape=Mdiamond]
+        work  [type="hanging", label="Work"]
+        exit  [shape=Msquare]
+        start -> work -> exit
+    }"#;
+    let graph = parse(dot).expect("parse should succeed");
+
+    // Verify the stall_timeout was parsed correctly
+    assert_eq!(
+        graph.stall_timeout(),
+        Some(std::time::Duration::from_millis(200)),
+    );
+
+    let dir = tempfile::tempdir().unwrap();
+    let mut registry = HandlerRegistry::new(Box::new(StartHandler));
+    registry.register("start", Box::new(StartHandler));
+    registry.register("exit", Box::new(ExitHandler));
+    registry.register("hanging", Box::new(HangingHandler));
+
+    let events = Arc::new(std::sync::Mutex::new(Vec::new()));
+    let events_clone = events.clone();
+    let mut emitter = EventEmitter::new();
+    emitter.on_event(move |event| {
+        events_clone
+            .lock()
+            .unwrap()
+            .push(format!("{event:?}"));
+    });
+
+    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: "stall-e2e".into(),
+        git_checkpoint: None,
+        base_sha: None,
+        run_branch: None,
+        meta_branch: None,
+    };
+
+    let result = engine.run(&graph, &config).await;
+    assert!(result.is_err(), "expected stall watchdog error");
+    let err = result.unwrap_err().to_string();
+    assert!(
+        err.contains("stall watchdog"),
+        "expected error to contain 'stall watchdog', got: {err}"
+    );
+
+    // Verify StallWatchdogTimeout event was emitted
+    let collected = events.lock().unwrap();
+    assert!(
+        collected.iter().any(|e| e.contains("StallWatchdogTimeout")),
+        "expected StallWatchdogTimeout event in: {collected:?}"
+    );
+}
+
+#[tokio::test]
+async fn e2e_stall_watchdog_kept_alive_by_handler_events() {
+    // Parse a DOT graph with stall_timeout 200ms, but the handler emits events
+    // every 100ms for 500ms total — the watchdog should NOT trigger.
+    let dot = r#"digraph StallAliveTest {
+        graph [goal="Test stall keepalive", stall_timeout="200ms", default_max_retry=0]
+        start [shape=Mdiamond]
+        work  [type="keepalive", label="Work"]
+        exit  [shape=Msquare]
+        start -> work -> exit
+    }"#;
+    let graph = parse(dot).expect("parse should succeed");
+
+    let dir = tempfile::tempdir().unwrap();
+    let mut registry = HandlerRegistry::new(Box::new(StartHandler));
+    registry.register("start", Box::new(StartHandler));
+    registry.register("exit", Box::new(ExitHandler));
+    registry.register(
+        "keepalive",
+        Box::new(KeepaliveHandler {
+            interval_ms: 100,
+            total_ms: 500,
+        }),
+    );
+
+    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: "stall-alive-e2e".into(),
+        git_checkpoint: None,
+        base_sha: None,
+        run_branch: None,
+        meta_branch: None,
+    };
+
+    let outcome = engine.run(&graph, &config).await.expect("pipeline should succeed");
+    assert_eq!(outcome.status, StageStatus::Success);
+}
+
+#[tokio::test]
+async fn e2e_stall_watchdog_disabled_with_zero_timeout() {
+    // Parse a DOT graph with stall_timeout="0s" — watchdog should be disabled,
+    // and a short sleep handler should complete successfully.
+    let dot = r#"digraph StallDisabledTest {
+        graph [goal="Test stall disabled", stall_timeout="0s", default_max_retry=0]
+        start [shape=Mdiamond]
+        work  [type="slow", label="Work"]
+        exit  [shape=Msquare]
+        start -> work -> exit
+    }"#;
+    let graph = parse(dot).expect("parse should succeed");
+    assert_eq!(graph.stall_timeout(), None, "zero timeout should disable watchdog");
+
+    let dir = tempfile::tempdir().unwrap();
+    let mut registry = HandlerRegistry::new(Box::new(StartHandler));
+    registry.register("start", Box::new(StartHandler));
+    registry.register("exit", Box::new(ExitHandler));
+    registry.register("slow", Box::new(SlowTestHandler { sleep_ms: 200 }));
+
+    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: "stall-disabled-e2e".into(),
+        git_checkpoint: None,
+        base_sha: None,
+        run_branch: None,
+        meta_branch: None,
+    };
+
+    let outcome = engine.run(&graph, &config).await.expect("pipeline should succeed");
+    assert_eq!(outcome.status, StageStatus::Success);
+}
+
+/// Handler that sleeps for a configurable duration, then succeeds (for e2e tests).
+struct SlowTestHandler {
+    sleep_ms: u64,
+}
+
+#[async_trait::async_trait]
+impl Handler for SlowTestHandler {
+    async fn execute(
+        &self,
+        _node: &Node,
+        _context: &Context,
+        _graph: &Graph,
+        _logs_root: &Path,
+        _services: &arc_workflows::handler::EngineServices,
+    ) -> Result<Outcome, ArcError> {
+        tokio::time::sleep(std::time::Duration::from_millis(self.sleep_ms)).await;
+        Ok(Outcome::success())
+    }
+}
+
+#[tokio::test]
+async fn e2e_stall_watchdog_with_explicit_timeout_override() {
+    // A short stall_timeout of 100ms should trigger faster than the default 600s.
+    // This tests that the graph attribute is actually respected.
+    let dot = r#"digraph StallOverrideTest {
+        graph [goal="Test stall override", stall_timeout="100ms", default_max_retry=0]
+        start [shape=Mdiamond]
+        work  [type="hanging", label="Work"]
+        exit  [shape=Msquare]
+        start -> work -> exit
+    }"#;
+    let graph = parse(dot).expect("parse should succeed");
+    assert_eq!(
+        graph.stall_timeout(),
+        Some(std::time::Duration::from_millis(100)),
+    );
+
+    let dir = tempfile::tempdir().unwrap();
+    let mut registry = HandlerRegistry::new(Box::new(StartHandler));
+    registry.register("start", Box::new(StartHandler));
+    registry.register("exit", Box::new(ExitHandler));
+    registry.register("hanging", Box::new(HangingHandler));
+
+    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: "stall-override-e2e".into(),
+        git_checkpoint: None,
+        base_sha: None,
+        run_branch: None,
+        meta_branch: None,
+    };
+
+    let start = std::time::Instant::now();
+    let result = engine.run(&graph, &config).await;
+    let elapsed = start.elapsed();
+
+    assert!(result.is_err(), "expected stall watchdog error");
+    let err = result.unwrap_err().to_string();
+    assert!(err.contains("stall watchdog"), "got: {err}");
+    // Should trigger well under 2 seconds (100ms timeout + check interval overhead)
+    assert!(
+        elapsed < std::time::Duration::from_secs(2),
+        "stall watchdog took too long: {elapsed:?}"
+    );
+}
+
 // Daytona parallel git branching test is in daytona_integration.rs