diff --git a/lib/crates/fabro-workflows/src/engine.rs b/lib/crates/fabro-workflows/src/engine.rs index 87df0908d..d1292c53b 100644 --- a/lib/crates/fabro-workflows/src/engine.rs +++ b/lib/crates/fabro-workflows/src/engine.rs @@ -1071,12 +1071,52 @@ impl WorkflowRunEngine { stage_index: usize, visit: usize, asset_globs: &[String], + run_id: &str, + hook_work_dir: Option<&Path>, ) -> Result<(Outcome, u32)> { let handler = self.services.registry.resolve(node); let node_timeout = node.timeout(); for attempt in 1..=policy.max_attempts { + // Run StageStart hook (blocking — can skip or block node) + { + let mut hook_ctx = HookContext::new( + HookEvent::StageStart, + run_id.to_string(), + graph.name.clone(), + ); + hook_ctx.cwd = hook_work_dir.map(|p| p.display().to_string()); + set_hook_node(&mut hook_ctx, node); + hook_ctx.attempt = Some(usize::try_from(attempt).unwrap_or(usize::MAX)); + hook_ctx.max_attempts = + Some(usize::try_from(policy.max_attempts).unwrap_or(usize::MAX)); + let decision = self.run_hooks(&hook_ctx, hook_work_dir).await; + match decision { + HookDecision::Skip { reason } => { + let mut outcome = Outcome::skipped(); + outcome.notes = + Some(reason.unwrap_or_else(|| "skipped by StageStart hook".into())); + return Ok((outcome, attempt)); + } + HookDecision::Block { reason } => { + let msg = reason.unwrap_or_else(|| "blocked by StageStart hook".into()); + return Err(FabroError::engine(msg)); + } + _ => {} + } + } + + // Emit StageStarted (fires once per attempt, only after hook passes) + self.services.emitter.emit(&WorkflowRunEvent::StageStarted { + node_id: node.id.clone(), + name: node.label().to_string(), + index: stage_index, + handler_type: node.handler_type().map(String::from), + script: node_script(node), + attempt: usize::try_from(attempt).unwrap_or(usize::MAX), + max_attempts: usize::try_from(policy.max_attempts).unwrap_or(usize::MAX), + }); // Floor to integer seconds: macOS stat reports mtime as integer seconds, // so a fractional epoch would reject files created in the same second. let command_start_epoch = std::time::SystemTime::now() @@ -1849,65 +1889,13 @@ impl WorkflowRunEngine { context.set(context::keys::CURRENT_NODE, serde_json::json!(&node.id)); let retry_policy = build_retry_policy(node, graph); - self.services.emitter.emit(&WorkflowRunEvent::StageStarted { - node_id: node.id.clone(), - name: node.label().to_string(), - index: stage_index, - handler_type: node.handler_type().map(String::from), - script: node_script(node), - attempt: 1, - max_attempts: usize::try_from(retry_policy.max_attempts).unwrap_or(usize::MAX), - }); - - // StageStart hook (blocking — can skip node) - { - let mut hook_ctx = - HookContext::new(HookEvent::StageStart, run_id.clone(), graph.name.clone()); - hook_ctx.cwd = hook_work_dir.as_ref().map(|p| p.display().to_string()); - set_hook_node(&mut hook_ctx, node); - hook_ctx.attempt = Some(1); - hook_ctx.max_attempts = - Some(usize::try_from(retry_policy.max_attempts).unwrap_or(usize::MAX)); - let decision = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await; - match decision { - HookDecision::Skip { reason } => { - let mut outcome = Outcome::skipped(); - outcome.notes = - Some(reason.unwrap_or_else(|| "skipped by StageStart hook".into())); - completed_nodes.push(node.id.clone()); - node_outcomes.insert(node.id.clone(), outcome); - previous_node_id = Some(node.id.clone()); - stage_index += 1; - // Select next edge and continue - let selection = select_edge( - &node.id, - &Outcome::skipped(), - &context, - graph, - node.selection(), - ); - if let Some(sel) = selection { - current_node_id = sel.edge.to.clone(); - incoming_edge = Some(sel.edge); - } else { - break; - } - continue; - } - HookDecision::Block { reason } => { - let msg = reason.unwrap_or_else(|| "blocked by StageStart hook".into()); - return Err(FabroError::engine(msg)); - } - _ => {} - } - } - let stage_start = Instant::now(); let (mut outcome, attempts_used) = if let Some((ref token, _)) = stall_token { tokio::select! { result = self.execute_with_retry( node, &context, graph, &config.run_dir, &retry_policy, stage_index, visit, &config.asset_globs, + &run_id, hook_work_dir.as_deref(), ) => result?, () = token.cancelled() => { let idle_secs = graph.stall_timeout().map_or(0, |d| d.as_secs()); @@ -1931,6 +1919,8 @@ impl WorkflowRunEngine { stage_index, visit, &config.asset_globs, + &run_id, + hook_work_dir.as_deref(), ) .await? }; @@ -1943,7 +1933,10 @@ impl WorkflowRunEngine { // Gap #1: Auto status -- when auto_status=true and outcome is non-success, // override to success with auto-status note - if node.auto_status() && outcome.status != StageStatus::Success { + if node.auto_status() + && outcome.status != StageStatus::Success + && outcome.status != StageStatus::Skipped + { outcome = Outcome { status: StageStatus::Success, notes: Some( @@ -1982,7 +1975,10 @@ impl WorkflowRunEngine { None }; - if outcome.status == StageStatus::Fail { + // Hook-skipped stages have no StageStarted, so skip completion events/hooks + if outcome.status == StageStatus::Skipped { + // No StageCompleted/StageFailed — proceed to write_node_status, edge selection, etc. + } else if outcome.status == StageStatus::Fail { self.services.emitter.emit(&WorkflowRunEvent::StageFailed { node_id: node.id.clone(), name: node.label().to_string(), @@ -5749,6 +5745,102 @@ mod tests { assert!(result.is_ok()); } + /// Handler that returns a retryable error on the first call and succeeds on subsequent calls. + struct FailOnceThenSucceedHandler { + call_count: std::sync::atomic::AtomicU32, + } + + #[async_trait] + impl HandlerTrait for FailOnceThenSucceedHandler { + async fn execute( + &self, + _node: &Node, + _context: &Context, + _graph: &Graph, + _run_dir: &Path, + _services: &crate::handler::EngineServices, + ) -> std::result::Result { + let n = self.call_count.fetch_add(1, Ordering::Relaxed); + if n == 0 { + Err(FabroError::handler("transient failure")) + } else { + Ok(Outcome::success()) + } + } + } + + #[tokio::test] + async fn retry_emits_stage_started_per_attempt() { + let dir = tempfile::tempdir().unwrap(); + let mut g = Graph::new("retry_events"); + g.attrs + .insert("goal".to_string(), AttrValue::String("test".to_string())); + + 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("fail_once".to_string()), + ); + // Allow 1 retry → 2 attempts total + work.attrs + .insert("max_retries".to_string(), AttrValue::Integer(1)); + 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 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(event.clone()); + }); + + let mut registry = make_registry(); + registry.register( + "fail_once", + Box::new(FailOnceThenSucceedHandler { + call_count: std::sync::atomic::AtomicU32::new(0), + }), + ); + + let engine = WorkflowRunEngine::new(registry, Arc::new(emitter), local_env()); + let config = test_run_config(dir.path(), "retry-events-test"); + let outcome = engine.run(&g, &config).await.unwrap(); + assert_eq!(outcome.status, StageStatus::Success); + + let collected = events.lock().unwrap(); + // Collect all StageStarted events for the "work" node + let work_started: Vec<_> = collected + .iter() + .filter_map(|e| match e { + WorkflowRunEvent::StageStarted { + node_id, attempt, .. + } if node_id == "work" => Some(*attempt), + _ => None, + }) + .collect(); + assert_eq!( + work_started, + vec![1, 2], + "expected StageStarted for attempt 1 and attempt 2, got: {work_started:?}" + ); + } + #[tokio::test] async fn run_with_lifecycle_emits_events_in_order() { let dir = tempfile::tempdir().unwrap(); diff --git a/lib/crates/fabro-workflows/tests/integration.rs b/lib/crates/fabro-workflows/tests/integration.rs index 23255c202..e1a23116b 100644 --- a/lib/crates/fabro-workflows/tests/integration.rs +++ b/lib/crates/fabro-workflows/tests/integration.rs @@ -7630,7 +7630,7 @@ async fn hook_stage_start_skip_bypasses_node() { "response.md should not exist when StageStart hook skips node" ); - // StageStarted should have been emitted + // StageStarted should NOT be emitted for hook-skipped stages (the stage never started) let captured = events.lock().unwrap(); let stage_starts: Vec<_> = captured .iter() @@ -7640,8 +7640,8 @@ async fn hook_stage_start_skip_bypasses_node() { }) .collect(); assert!( - !stage_starts.is_empty(), - "StageStarted should be emitted before hook skips" + stage_starts.is_empty(), + "StageStarted should not be emitted when StageStart hook skips" ); }