From 5208399e822cbeda9eba9c8834a5712b3dda4feb Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 31 Jul 2026 08:19:10 -0400 Subject: [PATCH 1/3] fix(workflow): pause timeouts for human input --- docs/public/reference/dot-language.mdx | 2 +- docs/public/workflows/stages-and-nodes.mdx | 2 +- .../fabro-workflow/src/event/emitter.rs | 24 ++- .../fabro-workflow/src/handler/human.rs | 14 +- .../fabro-workflow/src/interview_runtime.rs | 193 ++++++++++++++---- .../fabro-workflow/src/node_handler.rs | 155 ++++++++++++-- .../fabro-workflow/src/pipeline/execute.rs | 108 ++++++---- .../src/pipeline/execute/tests.rs | 140 ++++++++++++- 8 files changed, 538 insertions(+), 100 deletions(-) diff --git a/docs/public/reference/dot-language.mdx b/docs/public/reference/dot-language.mdx index 86ee23736..4ab0e2b27 100644 --- a/docs/public/reference/dot-language.mdx +++ b/docs/public/reference/dot-language.mdx @@ -196,7 +196,7 @@ Other node types still need their shape, because their attributes don't identify | `shape` | Identifier | Graphviz shape — determines handler type (see table above) | | `type` | String | Explicit handler type (overrides shape) | | `class` | String | Classes for [stylesheet](/workflows/stylesheets) targeting. Separate multiple classes with spaces. Commas are also accepted for compatibility. | -| `timeout` | Duration | Execution timeout (e.g. `900s`) | +| `timeout` | Duration | Execution timeout (e.g. `900s`). An agent's wait for human input does not consume this budget. On a human node, this is the response deadline. | | `max_visits` | Integer | Max times this node can execute in a run. Overrides the graph-level `max_node_visits` for this node. | | `max_retries` | Integer | Override default retry count | | `retry_policy` | String | Named preset: `none`, `standard`, `aggressive`, `linear`, `patient` | diff --git a/docs/public/workflows/stages-and-nodes.mdx b/docs/public/workflows/stages-and-nodes.mdx index 02a552990..d17db0d36 100644 --- a/docs/public/workflows/stages-and-nodes.mdx +++ b/docs/public/workflows/stages-and-nodes.mdx @@ -52,7 +52,7 @@ Key attributes: | `max_tokens` | Maximum tokens for LLM responses | | `fidelity` | How much prior context is passed to this node (see [Context](/execution/context#fidelity-controlling-agent-context)) | | `thread_id` | Groups nodes into a shared conversation thread (advanced — see below) | -| `timeout` | Execution timeout (e.g. `"900s"`) | +| `timeout` | Execution timeout (e.g. `"900s"`). Time spent waiting for an answer to an agent question does not count. | **Fidelity levels:** diff --git a/lib/components/fabro-workflow/src/event/emitter.rs b/lib/components/fabro-workflow/src/event/emitter.rs index 89f4d8600..75c5a6e6b 100644 --- a/lib/components/fabro-workflow/src/event/emitter.rs +++ b/lib/components/fabro-workflow/src/event/emitter.rs @@ -3,6 +3,7 @@ use std::sync::atomic::{AtomicI64, Ordering}; use ::fabro_types::{ExecOutputTail, RunEvent, RunId, RunNoticeCode, RunNoticeLevel}; use chrono::Utc; +use tokio::sync::watch; use super::Event; use super::convert::to_run_event_at; @@ -21,11 +22,12 @@ type EventListener = Arc; /// Callback-based event emitter for workflow run events. pub struct Emitter { - run_id: RunId, - listeners: std::sync::Mutex>, + run_id: RunId, + listeners: std::sync::Mutex>, /// Epoch milliseconds of the last `emit()` or `touch()` call. 0 until first /// event. - last_event_at: AtomicI64, + last_event_at: AtomicI64, + activity_revision: watch::Sender, } impl std::fmt::Debug for Emitter { @@ -35,6 +37,7 @@ impl std::fmt::Debug for Emitter { .field("run_id", &self.run_id) .field("listener_count", &count) .field("last_event_at", &self.last_event_at.load(Ordering::Relaxed)) + .field("activity_revision", &*self.activity_revision.borrow()) .finish() } } @@ -48,10 +51,12 @@ impl Default for Emitter { impl Emitter { #[must_use] pub fn new(run_id: RunId) -> Self { + let (activity_revision, _) = watch::channel(0); Self { run_id, listeners: std::sync::Mutex::new(Vec::new()), last_event_at: AtomicI64::new(0), + activity_revision, } } @@ -118,7 +123,6 @@ impl Emitter { } fn emit_with_scope(&self, event: &Event, scope: Option<&StageScope>) { - self.last_event_at.store(epoch_millis(), Ordering::Relaxed); event.trace(); if let Event::WorkflowRunStarted { run_id, .. } = event { debug_assert_eq!( @@ -131,7 +135,7 @@ impl Emitter { } pub(crate) fn dispatch_run_event(&self, event: &RunEvent) { - self.last_event_at.store(epoch_millis(), Ordering::Relaxed); + self.record_activity(); // Clone the listener list so we don't hold the lock during dispatch. // This prevents deadlocks if a listener calls emit() reentrantly. // Note: listeners added during this emit() won't receive the current event. @@ -151,10 +155,20 @@ impl Emitter { self.last_event_at.load(Ordering::Relaxed) } + pub(crate) fn subscribe_activity(&self) -> watch::Receiver { + self.activity_revision.subscribe() + } + /// Manually update the last-event timestamp (e.g. to seed the watchdog at /// workflow run start). pub fn touch(&self) { + self.record_activity(); + } + + fn record_activity(&self) { self.last_event_at.store(epoch_millis(), Ordering::Relaxed); + self.activity_revision + .send_modify(|revision| *revision = revision.wrapping_add(1)); } } diff --git a/lib/components/fabro-workflow/src/handler/human.rs b/lib/components/fabro-workflow/src/handler/human.rs index a961ecb02..999f4c984 100644 --- a/lib/components/fabro-workflow/src/handler/human.rs +++ b/lib/components/fabro-workflow/src/handler/human.rs @@ -277,7 +277,7 @@ impl Handler for HumanHandler { let interview_guard = services .run .interview_blocker - .block(Arc::clone(&services.run.emitter)); + .block(Arc::clone(&services.run.emitter), stage_scope.stage_id()); let interview_start = Instant::now(); let answer_submission = ask_with_timeout(self.interviewer.as_ref(), question).await; let answer_actor = answer_submission.actor.clone(); @@ -1264,6 +1264,8 @@ mod tests { let emitter = Arc::new(Emitter::new(fabro_types::fixtures::RUN_1)); let event_names = Arc::new(Mutex::new(Vec::new())); let guards = Arc::new(Mutex::new(Vec::new())); + let block_state = blocker.subscribe(); + let stage_id = fabro_types::StageId::new("gate", 1); emitter.on_event({ let event_names = Arc::clone(&event_names); @@ -1284,11 +1286,17 @@ mod tests { let blocker = Arc::clone(&blocker); let emitter = Arc::clone(&emitter); let guards = Arc::clone(&guards); + let stage_id = stage_id.clone(); scope.spawn(move || { - guards.lock().unwrap().push(blocker.block(emitter)); + guards + .lock() + .unwrap() + .push(blocker.block(emitter, stage_id)); }); } }); + assert!(block_state.borrow().is_run_blocked()); + assert!(block_state.borrow().is_stage_blocked(&stage_id)); std::thread::scope(|scope| { for _ in 0..8 { @@ -1304,5 +1312,7 @@ mod tests { "run.blocked", "run.unblocked" ],); + assert!(!block_state.borrow().is_run_blocked()); + assert!(!block_state.borrow().is_stage_blocked(&stage_id)); } } diff --git a/lib/components/fabro-workflow/src/interview_runtime.rs b/lib/components/fabro-workflow/src/interview_runtime.rs index 8eb58d889..8f9388c9b 100644 --- a/lib/components/fabro-workflow/src/interview_runtime.rs +++ b/lib/components/fabro-workflow/src/interview_runtime.rs @@ -1,5 +1,5 @@ -use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; use std::time::Instant; use async_trait::async_trait; @@ -7,31 +7,95 @@ use fabro_agent::{ AgentQuestion, AgentQuestionAnswer, AgentQuestionAnswerStatus, AgentQuestionRuntime, }; use fabro_interview::{Answer, AnswerSubmission, AnswerValue, Interviewer, Question}; -use fabro_types::{BlockedReason, InterviewOption, Principal, SystemActorKind}; +use fabro_types::{BlockedReason, InterviewOption, Principal, StageId, SystemActorKind}; use futures::future; +use tokio::sync::watch; use tokio_util::sync::CancellationToken; use ulid::Ulid; use crate::event::{Emitter, Event, StageScope}; use crate::millis_u64; -/// Run-scoped refcount for unresolved human input. Emits `run.blocked` on the +#[derive(Debug, Default)] +pub(crate) struct InterviewBlockState { + unresolved_interviews: usize, + blocked_stages: HashMap, +} + +impl InterviewBlockState { + pub(crate) fn is_run_blocked(&self) -> bool { + self.unresolved_interviews > 0 + } + + pub(crate) fn is_stage_blocked(&self, stage_id: &StageId) -> bool { + self.blocked_stages.contains_key(stage_id) + } + + fn block(&mut self, stage_id: StageId) { + self.unresolved_interviews = self + .unresolved_interviews + .checked_add(1) + .expect("unresolved interview count should not overflow"); + let stage_count = self.blocked_stages.entry(stage_id).or_default(); + *stage_count = stage_count + .checked_add(1) + .expect("stage interview count should not overflow"); + } + + fn resolve(&mut self, stage_id: &StageId) { + self.unresolved_interviews = self + .unresolved_interviews + .checked_sub(1) + .expect("an interview guard should resolve only once"); + let stage_count = self + .blocked_stages + .get_mut(stage_id) + .expect("a guarded stage should remain registered until resolution"); + *stage_count = stage_count + .checked_sub(1) + .expect("a stage interview guard should resolve only once"); + if *stage_count == 0 { + self.blocked_stages.remove(stage_id); + } + } +} + +/// Run-scoped state for unresolved human input. Emits `run.blocked` on the /// first unresolved human/agent interview and `run.unblocked` after the last -/// one resolves. +/// one resolves. Subscribers use the same state to suspend run and stage +/// timeout budgets without deriving runtime control from persisted events. pub(crate) struct RunInterviewBlocker { - unresolved_interviews: AtomicUsize, + state: watch::Sender, + transitions: Mutex<()>, } impl RunInterviewBlocker { #[must_use] pub(crate) fn new() -> Self { + let (state, _) = watch::channel(InterviewBlockState::default()); Self { - unresolved_interviews: AtomicUsize::new(0), + state, + transitions: Mutex::new(()), } } - pub(crate) fn block(self: &Arc, emitter: Arc) -> RunInterviewGuard { - if self.unresolved_interviews.fetch_add(1, Ordering::AcqRel) == 0 { + pub(crate) fn subscribe(&self) -> watch::Receiver { + self.state.subscribe() + } + + pub(crate) fn block( + self: &Arc, + emitter: Arc, + stage_id: StageId, + ) -> RunInterviewGuard { + let _transition = self + .transitions + .lock() + .expect("interview transition mutex should not be poisoned"); + let should_emit_blocked = !self.state.borrow().is_run_blocked(); + self.state + .send_modify(|state| state.block(stage_id.clone())); + if should_emit_blocked { emitter.emit(&Event::RunBlocked { blocked_reason: BlockedReason::HumanInputRequired, }); @@ -39,37 +103,28 @@ impl RunInterviewBlocker { RunInterviewGuard { blocker: Arc::clone(self), emitter, + stage_id, resolved: false, } } - fn resolved(&self, emitter: &Emitter) { - let mut current = self.unresolved_interviews.load(Ordering::Acquire); - loop { - if current == 0 { - return; - } - match self.unresolved_interviews.compare_exchange_weak( - current, - current - 1, - Ordering::AcqRel, - Ordering::Acquire, - ) { - Ok(_) => { - if current == 1 { - emitter.emit(&Event::RunUnblocked); - } - return; - } - Err(observed) => current = observed, - } + fn resolved(&self, emitter: &Emitter, stage_id: &StageId) { + let _transition = self + .transitions + .lock() + .expect("interview transition mutex should not be poisoned"); + let should_emit_unblocked = self.state.borrow().unresolved_interviews == 1; + if should_emit_unblocked { + emitter.emit(&Event::RunUnblocked); } + self.state.send_modify(|state| state.resolve(stage_id)); } } pub(crate) struct RunInterviewGuard { blocker: Arc, emitter: Arc, + stage_id: StageId, resolved: bool, } @@ -80,7 +135,7 @@ impl RunInterviewGuard { fn resolve_in_place(&mut self) { if !self.resolved { - self.blocker.resolved(self.emitter.as_ref()); + self.blocker.resolved(self.emitter.as_ref(), &self.stage_id); self.resolved = true; } } @@ -236,7 +291,8 @@ impl AgentQuestionRuntime for WorkflowAgentQuestionRuntime { self.stage_scope.clone(), self.stage_id.clone(), &prepared, - self.blocker.block(Arc::clone(&self.emitter)), + self.blocker + .block(Arc::clone(&self.emitter), self.stage_scope.stage_id()), interview_start, ); let ask_all = future::join_all( @@ -528,17 +584,21 @@ mod tests { let events = Arc::clone(&events); move |event| events.lock().unwrap().push(event.clone()) }); + let stage_scope = StageScope { + node_id: "ask".to_string(), + visit: 1, + parallel_group_id: None, + parallel_branch_id: None, + }; + let stage_id = stage_scope.stage_id(); + let blocker = Arc::new(RunInterviewBlocker::new()); + let block_state = blocker.subscribe(); let runtime = WorkflowAgentQuestionRuntime::new( interviewer.clone(), Arc::clone(&emitter), - StageScope { - node_id: "ask".to_string(), - visit: 1, - parallel_group_id: None, - parallel_branch_id: None, - }, + stage_scope, "ask", - Arc::new(RunInterviewBlocker::new()), + blocker, ); let option = InterviewOption { key: "ship".to_string(), @@ -578,6 +638,8 @@ mod tests { }); tokio::task::yield_now().await; + assert!(block_state.borrow().is_run_blocked()); + assert!(block_state.borrow().is_stage_blocked(&stage_id)); let question_ids = { let events = events.lock().unwrap(); assert!(matches!(events[0].body, EventBody::InterviewStarted(_))); @@ -622,5 +684,60 @@ mod tests { .iter() .any(|event| matches!(event.body, EventBody::RunUnblocked(_))) ); + assert!(!block_state.borrow().is_run_blocked()); + assert!(!block_state.borrow().is_stage_blocked(&stage_id)); + } + + #[tokio::test] + async fn cancelling_agent_question_unblocks_its_stage() { + let interviewer = Arc::new(ControlInterviewer::new()); + let emitter = Arc::new(Emitter::new(RunId::new())); + let stage_scope = StageScope { + node_id: "ask".to_string(), + visit: 1, + parallel_group_id: None, + parallel_branch_id: None, + }; + let stage_id = stage_scope.stage_id(); + let blocker = Arc::new(RunInterviewBlocker::new()); + let block_state = blocker.subscribe(); + let runtime = WorkflowAgentQuestionRuntime::new( + interviewer, + emitter, + stage_scope, + "ask", + Arc::clone(&blocker), + ); + let cancel_token = CancellationToken::new(); + let ask_cancel_token = cancel_token.clone(); + let ask = tokio::spawn(async move { + runtime + .ask_questions( + "call_1", + vec![AgentQuestion { + original_id: Some("q1".to_string()), + original_question: "Continue?".to_string(), + header: None, + text: "Continue?".to_string(), + question_type: fabro_types::QuestionType::Freeform, + options: Vec::new(), + allow_freeform: true, + }], + ask_cancel_token, + ) + .await + .unwrap() + }); + + tokio::task::yield_now().await; + assert!(block_state.borrow().is_run_blocked()); + assert!(block_state.borrow().is_stage_blocked(&stage_id)); + + cancel_token.cancel(); + let answers = ask.await.unwrap(); + + assert_eq!(answers[0].status, AgentQuestionAnswerStatus::Interrupted); + assert!(!block_state.borrow().is_run_blocked()); + assert!(!block_state.borrow().is_stage_blocked(&stage_id)); } } diff --git a/lib/components/fabro-workflow/src/node_handler.rs b/lib/components/fabro-workflow/src/node_handler.rs index b54ff0f22..9e0a3eefe 100644 --- a/lib/components/fabro-workflow/src/node_handler.rs +++ b/lib/components/fabro-workflow/src/node_handler.rs @@ -1,6 +1,8 @@ +use std::future::Future; use std::panic::AssertUnwindSafe; use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::time::Duration; use async_trait::async_trait; use fabro_core::error::{Error as CoreError, HandlerErrorDetail, Result as CoreResult}; @@ -8,18 +10,63 @@ use fabro_core::handler::NodeHandler; use fabro_core::outcome::FailureCategory; use fabro_core::retry::RetryPolicy as CoreRetryPolicy; use fabro_graphviz::graph::types::{Graph as GvGraph, Node as GvNode}; -use fabro_types::SystemActorKind; +use fabro_types::{StageId, SystemActorKind}; use futures::FutureExt; -use tokio::time::timeout; +use tokio::sync::watch; +use tokio::time::{Instant, sleep}; use crate::artifact; use crate::context::Context; use crate::error::Error; +use crate::event::StageScope; use crate::graph::{WorkflowGraph, WorkflowNode}; use crate::handler::{EngineServices, NodeTimeoutPolicy, dispatch_handler, format_panic_message}; +use crate::interview_runtime::InterviewBlockState; use crate::outcome::{FailureDetail, Outcome, StageOutcome}; use crate::retry::build_retry_policy; +async fn timeout_excluding_interview_wait( + duration: Duration, + stage_id: &StageId, + mut interview_blocks: watch::Receiver, + future: F, +) -> Option +where + F: Future, +{ + tokio::pin!(future); + let mut remaining = duration; + + loop { + if interview_blocks + .borrow_and_update() + .is_stage_blocked(stage_id) + { + tokio::select! { + biased; + output = &mut future => return Some(output), + changed = interview_blocks.changed() => { + changed.expect("run services should own interview state for the handler lifetime"); + } + } + continue; + } + + let active_started = Instant::now(); + let deadline = sleep(remaining); + tokio::pin!(deadline); + tokio::select! { + biased; + output = &mut future => return Some(output), + changed = interview_blocks.changed() => { + changed.expect("run services should own interview state for the handler lifetime"); + remaining = remaining.saturating_sub(active_started.elapsed()); + } + () = &mut deadline => return None, + } + } +} + /// Production node handler that bridges fabro-core's NodeHandler to the /// existing fabro-workflow Handler trait via EngineServices. /// @@ -66,23 +113,29 @@ pub(crate) async fn execute_single_attempt( NodeTimeoutPolicy::HandlerManaged => None, }; + let stage_id = StageScope::for_handler(&wf_context, &node.id).stage_id(); let future = dispatch_handler(handler, node, &wf_context, graph, run_dir, services); let panic_safe = AssertUnwindSafe(future).catch_unwind(); let timed_result = if let Some(duration) = node_timeout { - match timeout(duration, panic_safe).await { - Ok(inner) => inner, - Err(_elapsed) => { - let mut failure = FailureDetail::new( - format!("handler timed out after {}ms", duration.as_millis()), - FailureCategory::TransientInfra, - ); - failure.system_actor = Some(SystemActorKind::Timeout); - return Err(CoreError::handler(HandlerErrorDetail { - retryable: true, - failure, - })); - } - } + let Some(inner) = timeout_excluding_interview_wait( + duration, + &stage_id, + services.run.interview_blocker.subscribe(), + panic_safe, + ) + .await + else { + let mut failure = FailureDetail::new( + format!("handler timed out after {}ms", duration.as_millis()), + FailureCategory::TransientInfra, + ); + failure.system_actor = Some(SystemActorKind::Timeout); + return Err(CoreError::handler(HandlerErrorDetail { + retryable: true, + failure, + })); + }; + inner } else { panic_safe.await }; @@ -186,7 +239,9 @@ mod tests { use fabro_graphviz::graph::types::{Edge, Graph, Node}; use super::*; + use crate::event::Emitter; use crate::graph::WorkflowGraph; + use crate::interview_runtime::RunInterviewBlocker; /// Minimal spike handler that always succeeds — proves the trait plumbing. pub(crate) struct SpikeHandler; @@ -235,4 +290,72 @@ mod tests { let (result, _) = executor.run(&wf_graph, state).await.unwrap(); assert_eq!(result.status, StageOutcome::Succeeded); } + + #[tokio::test(start_paused = true)] + async fn node_timeout_does_not_count_own_interview_wait() { + let blocker = Arc::new(RunInterviewBlocker::new()); + let emitter = Arc::new(Emitter::default()); + let stage_id = StageId::new("agent", 1); + let block_state = blocker.subscribe(); + let guard = blocker.block(emitter, stage_id.clone()); + + let result = timeout_excluding_interview_wait( + Duration::from_millis(50), + &stage_id, + block_state, + async move { + sleep(Duration::from_millis(100)).await; + guard.resolve(); + sleep(Duration::from_millis(40)).await; + "completed" + }, + ) + .await; + + assert_eq!(result, Some("completed")); + } + + #[tokio::test(start_paused = true)] + async fn node_timeout_still_limits_active_work_after_interview() { + let blocker = Arc::new(RunInterviewBlocker::new()); + let emitter = Arc::new(Emitter::default()); + let stage_id = StageId::new("agent", 1); + let block_state = blocker.subscribe(); + let guard = blocker.block(emitter, stage_id.clone()); + + let result = timeout_excluding_interview_wait( + Duration::from_millis(50), + &stage_id, + block_state, + async move { + sleep(Duration::from_millis(100)).await; + guard.resolve(); + sleep(Duration::from_millis(60)).await; + }, + ) + .await; + + assert_eq!(result, None); + } + + #[tokio::test(start_paused = true)] + async fn node_timeout_does_not_pause_for_another_stage_interview() { + let blocker = Arc::new(RunInterviewBlocker::new()); + let emitter = Arc::new(Emitter::default()); + let blocked_stage = StageId::new("agent_a", 1); + let active_stage = StageId::new("agent_b", 1); + let block_state = blocker.subscribe(); + let guard = blocker.block(emitter, blocked_stage); + + let result = timeout_excluding_interview_wait( + Duration::from_millis(50), + &active_stage, + block_state, + sleep(Duration::from_millis(100)), + ) + .await; + guard.resolve(); + + assert_eq!(result, None); + } } diff --git a/lib/components/fabro-workflow/src/pipeline/execute.rs b/lib/components/fabro-workflow/src/pipeline/execute.rs index a82fd7c8a..5edb7a600 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute.rs @@ -1,10 +1,11 @@ use std::sync::Arc; -use std::time::Instant; +use std::time::{Duration, Instant}; use fabro_core::executor::ExecutorBuilder; use fabro_core::handler::NodeHandler; use fabro_core::state::ExecutionState; -use tokio::time::sleep; +use tokio::sync::watch; +use tokio::time::{Instant as TokioInstant, sleep_until}; use tokio_util::sync::CancellationToken; use super::types::{Executed, Initialized}; @@ -13,6 +14,7 @@ use crate::context::{self, Context}; use crate::error::Error; use crate::event::Event; use crate::graph::WorkflowGraph; +use crate::interview_runtime::InterviewBlockState; use crate::lifecycle::WorkflowLifecycle; use crate::node_handler::WorkflowNodeHandler; use crate::outcome::Outcome; @@ -28,6 +30,60 @@ fn seed_context_from_checkpoint(checkpoint: Option<&Checkpoint>) -> Context { context } +async fn monitor_for_stall( + stall_timeout: Duration, + stall_token: CancellationToken, + shutdown: CancellationToken, + mut activity: watch::Receiver, + mut interview_blocks: watch::Receiver, +) { + let mut deadline = TokioInstant::now() + stall_timeout; + + loop { + if interview_blocks.borrow_and_update().is_run_blocked() { + tokio::select! { + biased; + () = shutdown.cancelled() => return, + changed = interview_blocks.changed() => { + if changed.is_err() { + return; + } + if !interview_blocks.borrow().is_run_blocked() { + deadline = TokioInstant::now() + stall_timeout; + } + } + } + continue; + } + + let deadline_timer = sleep_until(deadline); + tokio::pin!(deadline_timer); + tokio::select! { + biased; + () = shutdown.cancelled() => return, + changed = interview_blocks.changed() => { + if changed.is_err() { + return; + } + deadline = TokioInstant::now() + stall_timeout; + } + changed = activity.changed() => { + if changed.is_err() { + return; + } + deadline = TokioInstant::now() + stall_timeout; + } + () = &mut deadline_timer => { + if interview_blocks.borrow().is_run_blocked() { + continue; + } + stall_token.cancel(); + return; + } + } + } +} + /// EXECUTE phase: run the workflow graph. /// /// Infallible at the function level — engine errors are captured in `outcome`. @@ -187,43 +243,20 @@ pub async fn execute(init: Initialized) -> Executed { let stall_timeout_opt = graph.stall_timeout(); let stall_token = stall_timeout_opt.map(|_| CancellationToken::new()); - let stall_shutdown = + let stall_watchdog = if let (Some(stall_timeout), Some(ref token)) = (stall_timeout_opt, &stall_token) { let shutdown = CancellationToken::new(); let emitter = Arc::clone(&engine.run.emitter); - let token_clone = token.clone(); - let shutdown_clone = shutdown.clone(); + let interview_blocks = engine.run.interview_blocker.subscribe(); emitter.touch(); - tokio::spawn(async move { - loop { - tokio::select! { - () = sleep(stall_timeout) => { - if shutdown_clone.is_cancelled() { - return; - } - let last = emitter.last_event_at(); - let now = i64::try_from( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis(), - ) - .unwrap_or(i64::MAX); - let idle_ms = now.saturating_sub(last); - let stall_timeout_ms = - i64::try_from(stall_timeout.as_millis()).unwrap_or(i64::MAX); - if idle_ms >= stall_timeout_ms { - token_clone.cancel(); - return; - } - } - () = shutdown_clone.cancelled() => { - return; - } - } - } - }); - Some(shutdown) + let task = tokio::spawn(monitor_for_stall( + stall_timeout, + token.clone(), + shutdown.clone(), + emitter.subscribe_activity(), + interview_blocks, + )); + Some((shutdown, task)) } else { None }; @@ -242,8 +275,11 @@ pub async fn execute(init: Initialized) -> Executed { let executor = builder.build(); let result = executor.run(&wf_graph, state).await; - if let Some(shutdown) = stall_shutdown { + if let Some((shutdown, task)) = stall_watchdog { shutdown.cancel(); + if let Err(error) = task.await { + tracing::error!(error = ?error, "stall watchdog task failed"); + } } let (outcome, final_context) = match result { diff --git a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs index 6fd73a17c..a4a15f467 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs @@ -28,9 +28,10 @@ use object_store::memory::InMemory; use super::*; use crate::context::{self, Context}; use crate::error::Error; -use crate::event::{Emitter, Event, StoreProgressLogger, append_event}; +use crate::event::{Emitter, Event, StageScope, StoreProgressLogger, append_event}; use crate::handler::start::StartHandler; use crate::handler::{Handler as HandlerTrait, HandlerRegistry}; +use crate::interview_runtime::RunInterviewBlocker; use crate::model_fallback::ModelFallbackPolicy; use crate::outcome::{Outcome, OutcomeExt, StageOutcome}; use crate::pipeline::initialize; @@ -650,6 +651,59 @@ impl HandlerTrait for SlowHandler { } } +struct InterviewWaitHandler { + wait_ms: u64, + active_ms: u64, +} + +#[async_trait] +impl HandlerTrait for InterviewWaitHandler { + async fn execute( + &self, + node: &Node, + context: &Context, + _graph: &Graph, + _run_dir: &Path, + services: &crate::handler::EngineServices, + ) -> std::result::Result { + let stage_id = StageScope::for_handler(context, &node.id).stage_id(); + let guard = services + .run + .interview_blocker + .block(Arc::clone(&services.run.emitter), stage_id); + tokio::time::sleep(Duration::from_millis(self.wait_ms)).await; + guard.resolve(); + tokio::time::sleep(Duration::from_millis(self.active_ms)).await; + Ok(Outcome::success()) + } +} + +fn interview_wait_graph(stall_timeout: Duration, node_timeout: Option) -> Graph { + let mut graph = simple_graph(); + graph.attrs.insert( + "stall_timeout".to_string(), + AttrValue::Duration(stall_timeout), + ); + graph + .attrs + .insert("default_max_retries".to_string(), AttrValue::Integer(0)); + + let mut work = Node::new("work"); + work.attrs.insert( + "type".to_string(), + AttrValue::String("interview_wait".to_string()), + ); + if let Some(timeout) = node_timeout { + work.attrs + .insert("timeout".to_string(), AttrValue::Duration(timeout)); + } + graph.nodes.insert("work".to_string(), work); + graph.edges.clear(); + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + graph +} + struct StopsSandboxHandler { sandbox: Arc, } @@ -1371,6 +1425,90 @@ async fn stall_watchdog_triggers_on_hung_handler() { assert!(err.contains("stall watchdog")); } +#[tokio::test(start_paused = true)] +async fn stall_watchdog_starts_a_fresh_deadline_after_human_input() { + let emitter = Arc::new(Emitter::default()); + let blocker = Arc::new(RunInterviewBlocker::new()); + let stall_token = CancellationToken::new(); + let shutdown = CancellationToken::new(); + let watchdog = tokio::spawn(monitor_for_stall( + Duration::from_millis(50), + stall_token.clone(), + shutdown.clone(), + emitter.subscribe_activity(), + blocker.subscribe(), + )); + tokio::task::yield_now().await; + + let guard = blocker.block(Arc::clone(&emitter), fabro_types::StageId::new("work", 1)); + tokio::time::advance(Duration::from_millis(200)).await; + assert!(!stall_token.is_cancelled()); + + guard.resolve(); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_millis(49)).await; + assert!(!stall_token.is_cancelled()); + + tokio::time::advance(Duration::from_millis(2)).await; + tokio::task::yield_now().await; + assert!(stall_token.is_cancelled()); + + shutdown.cancel(); + watchdog.await.unwrap(); +} + +#[tokio::test] +async fn stall_watchdog_suspends_while_run_waits_for_human_input() { + let dir = tempfile::tempdir().unwrap(); + let graph = interview_wait_graph(Duration::from_millis(50), None); + let mut registry = make_registry(); + registry.register( + "interview_wait", + Box::new(InterviewWaitHandler { + wait_ms: 150, + active_ms: 10, + }), + ); + + let outcome = run_graph( + registry, + test_emitter_arc("test-run"), + local_env(), + &graph, + &test_run_options(dir.path(), "test-run"), + ) + .await + .expect("human input wait should not trigger the stall watchdog"); + + assert_eq!(outcome.status, StageOutcome::Succeeded); +} + +#[tokio::test] +async fn node_timeout_excludes_human_input_wait() { + let dir = tempfile::tempdir().unwrap(); + let graph = interview_wait_graph(Duration::ZERO, Some(Duration::from_millis(50))); + let mut registry = make_registry(); + registry.register( + "interview_wait", + Box::new(InterviewWaitHandler { + wait_ms: 150, + active_ms: 20, + }), + ); + + let outcome = run_graph( + registry, + test_emitter_arc("test-run"), + local_env(), + &graph, + &test_run_options(dir.path(), "test-run"), + ) + .await + .expect("human input wait should not consume the node timeout"); + + assert_eq!(outcome.status, StageOutcome::Succeeded); +} + #[tokio::test] async fn retry_emits_stage_started_per_attempt() { let dir = tempfile::tempdir().unwrap(); From 3fa48c38aa197e39b2edd98bbe3ebeb664cb04a2 Mon Sep 17 00:00:00 2001 From: Release Repro Date: Sat, 1 Aug 2026 09:52:49 -0400 Subject: [PATCH 2/3] refactor(workflow): simplify interview block state and stall watchdog Follow-up cleanup on the human-input timeout work. - Drop the `unresolved_interviews` counter from `InterviewBlockState`. It duplicated `blocked_stages`, which is non-empty exactly when the run is blocked. - Publish block state before emitting `run.blocked` / `run.unblocked` in both directions, so a listener reading `subscribe()` from an event callback never sees state that disagrees with the event. The watchdog still gets a full fresh deadline because it restarts on the unblock transition. - Stop panicking in `InterviewBlockState::resolve`. It runs from `Drop`, where a panic during unwind aborts the process. - Replace the emitter's `activity_revision` watch channel with a monotonic timestamp. `record_activity` runs on every agent stream delta, and the channel woke the watchdog task and re-armed its timer per event. The watchdog now samples `last_activity()` when its deadline fires and re-arms only if the run was active, so the hot path is one clock read and one relaxed store. - Remove the now-unused `last_event_at()` and `epoch_millis()`. - Collapse the duplicated blocked/unblocked `select!` arms in `monitor_for_stall` and `timeout_excluding_interview_wait` into one loop each, using a branch precondition to park the timer while blocked. - Handle a dropped block-state sender in `timeout_excluding_interview_wait` by falling back to a plain deadline instead of panicking, which also removes a potential busy loop. - Only compute `stage_id` when the node actually has a timeout. Co-Authored-By: Claude Opus 5 (1M context) --- .../fabro-workflow/src/event/emitter.rs | 66 +++++++++---------- .../fabro-workflow/src/interview_runtime.rs | 61 ++++++++--------- .../fabro-workflow/src/node_handler.rs | 38 +++++------ .../fabro-workflow/src/pipeline/execute.rs | 47 +++++-------- .../src/pipeline/execute/tests.rs | 2 +- 5 files changed, 98 insertions(+), 116 deletions(-) diff --git a/lib/components/fabro-workflow/src/event/emitter.rs b/lib/components/fabro-workflow/src/event/emitter.rs index 75c5a6e6b..d334b79a6 100644 --- a/lib/components/fabro-workflow/src/event/emitter.rs +++ b/lib/components/fabro-workflow/src/event/emitter.rs @@ -1,33 +1,28 @@ use std::sync::Arc; -use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; use ::fabro_types::{ExecOutputTail, RunEvent, RunId, RunNoticeCode, RunNoticeLevel}; use chrono::Utc; -use tokio::sync::watch; +use tokio::time::Instant; use super::Event; use super::convert::to_run_event_at; +use crate::millis_u64; use crate::stage_scope::StageScope; -fn epoch_millis() -> i64 { - let millis = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis(); - i64::try_from(millis).unwrap_or(i64::MAX) -} - /// Listener callback type for workflow run events. type EventListener = Arc; /// Callback-based event emitter for workflow run events. pub struct Emitter { - run_id: RunId, - listeners: std::sync::Mutex>, - /// Epoch milliseconds of the last `emit()` or `touch()` call. 0 until first - /// event. - last_event_at: AtomicI64, - activity_revision: watch::Sender, + run_id: RunId, + listeners: std::sync::Mutex>, + /// Monotonic origin that `last_activity_ms` is measured from. + activity_origin: Instant, + /// Milliseconds after `activity_origin` of the last `emit()` or `touch()`. + /// 0 until the first event. + last_activity_ms: AtomicU64, } impl std::fmt::Debug for Emitter { @@ -36,9 +31,11 @@ impl std::fmt::Debug for Emitter { f.debug_struct("Emitter") .field("run_id", &self.run_id) .field("listener_count", &count) - .field("last_event_at", &self.last_event_at.load(Ordering::Relaxed)) - .field("activity_revision", &*self.activity_revision.borrow()) - .finish() + .field( + "last_activity_ms", + &self.last_activity_ms.load(Ordering::Relaxed), + ) + .finish_non_exhaustive() } } @@ -51,12 +48,11 @@ impl Default for Emitter { impl Emitter { #[must_use] pub fn new(run_id: RunId) -> Self { - let (activity_revision, _) = watch::channel(0); Self { run_id, listeners: std::sync::Mutex::new(Vec::new()), - last_event_at: AtomicI64::new(0), - activity_revision, + activity_origin: Instant::now(), + last_activity_ms: AtomicU64::new(0), } } @@ -149,26 +145,26 @@ impl Emitter { } } - /// 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) + /// Returns the monotonic instant of the last `emit()` or `touch()` call, + /// or the emitter's creation instant if neither has been called. + pub(crate) fn last_activity(&self) -> Instant { + self.activity_origin + Duration::from_millis(self.last_activity_ms.load(Ordering::Relaxed)) } - pub(crate) fn subscribe_activity(&self) -> watch::Receiver { - self.activity_revision.subscribe() - } - - /// Manually update the last-event timestamp (e.g. to seed the watchdog at - /// workflow run start). + /// Manually record activity (e.g. to seed the watchdog at workflow run + /// start, or for agent stream deltas that are not emitted as run events). pub fn touch(&self) { self.record_activity(); } + /// Called for every event, including agent streaming deltas. Keep this to a + /// single clock read and a relaxed store — the stall watchdog samples it at + /// its own deadline rather than being woken here. fn record_activity(&self) { - self.last_event_at.store(epoch_millis(), Ordering::Relaxed); - self.activity_revision - .send_modify(|revision| *revision = revision.wrapping_add(1)); + self.last_activity_ms.store( + millis_u64(self.activity_origin.elapsed()), + Ordering::Relaxed, + ); } } diff --git a/lib/components/fabro-workflow/src/interview_runtime.rs b/lib/components/fabro-workflow/src/interview_runtime.rs index 8f9388c9b..dc2855864 100644 --- a/lib/components/fabro-workflow/src/interview_runtime.rs +++ b/lib/components/fabro-workflow/src/interview_runtime.rs @@ -16,15 +16,16 @@ use ulid::Ulid; use crate::event::{Emitter, Event, StageScope}; use crate::millis_u64; +/// Unresolved interviews per stage. A stage is present only while it has at +/// least one, so the run is blocked exactly when the map is non-empty. #[derive(Debug, Default)] pub(crate) struct InterviewBlockState { - unresolved_interviews: usize, - blocked_stages: HashMap, + blocked_stages: HashMap, } impl InterviewBlockState { pub(crate) fn is_run_blocked(&self) -> bool { - self.unresolved_interviews > 0 + !self.blocked_stages.is_empty() } pub(crate) fn is_stage_blocked(&self, stage_id: &StageId) -> bool { @@ -32,29 +33,18 @@ impl InterviewBlockState { } fn block(&mut self, stage_id: StageId) { - self.unresolved_interviews = self - .unresolved_interviews - .checked_add(1) - .expect("unresolved interview count should not overflow"); - let stage_count = self.blocked_stages.entry(stage_id).or_default(); - *stage_count = stage_count - .checked_add(1) - .expect("stage interview count should not overflow"); + *self.blocked_stages.entry(stage_id).or_default() += 1; } + /// `RunInterviewGuard` resolves at most once, so an unknown stage here + /// means the state is already clear. Runs from `Drop`, so it must not + /// panic. fn resolve(&mut self, stage_id: &StageId) { - self.unresolved_interviews = self - .unresolved_interviews - .checked_sub(1) - .expect("an interview guard should resolve only once"); - let stage_count = self - .blocked_stages - .get_mut(stage_id) - .expect("a guarded stage should remain registered until resolution"); - *stage_count = stage_count - .checked_sub(1) - .expect("a stage interview guard should resolve only once"); - if *stage_count == 0 { + let Some(count) = self.blocked_stages.get_mut(stage_id) else { + return; + }; + *count = count.saturating_sub(1); + if *count == 0 { self.blocked_stages.remove(stage_id); } } @@ -64,8 +54,14 @@ impl InterviewBlockState { /// first unresolved human/agent interview and `run.unblocked` after the last /// one resolves. Subscribers use the same state to suspend run and stage /// timeout budgets without deriving runtime control from persisted events. +/// +/// Both transitions publish the new state before emitting the event, so a +/// listener that reads `subscribe()` from an event callback always sees state +/// that agrees with the event it just received. pub(crate) struct RunInterviewBlocker { state: watch::Sender, + /// Serializes state change plus event emission so concurrent guards cannot + /// interleave into an out-of-order `run.blocked` / `run.unblocked` pair. transitions: Mutex<()>, } @@ -92,10 +88,12 @@ impl RunInterviewBlocker { .transitions .lock() .expect("interview transition mutex should not be poisoned"); - let should_emit_blocked = !self.state.borrow().is_run_blocked(); - self.state - .send_modify(|state| state.block(stage_id.clone())); - if should_emit_blocked { + let mut newly_blocked = false; + self.state.send_modify(|state| { + newly_blocked = !state.is_run_blocked(); + state.block(stage_id.clone()); + }); + if newly_blocked { emitter.emit(&Event::RunBlocked { blocked_reason: BlockedReason::HumanInputRequired, }); @@ -113,11 +111,14 @@ impl RunInterviewBlocker { .transitions .lock() .expect("interview transition mutex should not be poisoned"); - let should_emit_unblocked = self.state.borrow().unresolved_interviews == 1; - if should_emit_unblocked { + let mut fully_unblocked = false; + self.state.send_modify(|state| { + state.resolve(stage_id); + fully_unblocked = !state.is_run_blocked(); + }); + if fully_unblocked { emitter.emit(&Event::RunUnblocked); } - self.state.send_modify(|state| state.resolve(stage_id)); } } diff --git a/lib/components/fabro-workflow/src/node_handler.rs b/lib/components/fabro-workflow/src/node_handler.rs index 9e0a3eefe..70a35a34b 100644 --- a/lib/components/fabro-workflow/src/node_handler.rs +++ b/lib/components/fabro-workflow/src/node_handler.rs @@ -13,7 +13,7 @@ use fabro_graphviz::graph::types::{Graph as GvGraph, Node as GvNode}; use fabro_types::{StageId, SystemActorKind}; use futures::FutureExt; use tokio::sync::watch; -use tokio::time::{Instant, sleep}; +use tokio::time::{Instant, sleep, timeout}; use crate::artifact; use crate::context::Context; @@ -25,6 +25,11 @@ use crate::interview_runtime::InterviewBlockState; use crate::outcome::{FailureDetail, Outcome, StageOutcome}; use crate::retry::build_retry_policy; +/// Runs `future` under a `duration` budget that only counts time when this +/// stage is not waiting on human input. A sibling stage's interview does not +/// pause this budget — the wait is keyed by `stage_id`. +/// +/// Returns `None` if the budget runs out first. async fn timeout_excluding_interview_wait( duration: Duration, stage_id: &StageId, @@ -38,31 +43,24 @@ where let mut remaining = duration; loop { - if interview_blocks + let blocked = interview_blocks .borrow_and_update() - .is_stage_blocked(stage_id) - { - tokio::select! { - biased; - output = &mut future => return Some(output), - changed = interview_blocks.changed() => { - changed.expect("run services should own interview state for the handler lifetime"); - } - } - continue; - } - + .is_stage_blocked(stage_id); let active_started = Instant::now(); - let deadline = sleep(remaining); - tokio::pin!(deadline); tokio::select! { biased; output = &mut future => return Some(output), changed = interview_blocks.changed() => { - changed.expect("run services should own interview state for the handler lifetime"); - remaining = remaining.saturating_sub(active_started.elapsed()); + if changed.is_err() { + // The blocker outlives every handler. If it ever goes away, + // fall back to a plain deadline rather than spinning. + return timeout(remaining, future).await.ok(); + } + if !blocked { + remaining = remaining.saturating_sub(active_started.elapsed()); + } } - () = &mut deadline => return None, + () = sleep(remaining), if !blocked => return None, } } } @@ -113,10 +111,10 @@ pub(crate) async fn execute_single_attempt( NodeTimeoutPolicy::HandlerManaged => None, }; - let stage_id = StageScope::for_handler(&wf_context, &node.id).stage_id(); let future = dispatch_handler(handler, node, &wf_context, graph, run_dir, services); let panic_safe = AssertUnwindSafe(future).catch_unwind(); let timed_result = if let Some(duration) = node_timeout { + let stage_id = StageScope::for_handler(&wf_context, &node.id).stage_id(); let Some(inner) = timeout_excluding_interview_wait( duration, &stage_id, diff --git a/lib/components/fabro-workflow/src/pipeline/execute.rs b/lib/components/fabro-workflow/src/pipeline/execute.rs index 5edb7a600..60d07b070 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute.rs @@ -12,7 +12,7 @@ use super::types::{Executed, Initialized}; use crate::artifact; use crate::context::{self, Context}; use crate::error::Error; -use crate::event::Event; +use crate::event::{Emitter, Event}; use crate::graph::WorkflowGraph; use crate::interview_runtime::InterviewBlockState; use crate::lifecycle::WorkflowLifecycle; @@ -30,34 +30,24 @@ fn seed_context_from_checkpoint(checkpoint: Option<&Checkpoint>) -> Context { context } +/// Cancels `stall_token` once the run goes `stall_timeout` without emitting an +/// event. Waiting on human input suspends the timer, and the first unblock +/// starts a fresh full deadline. +/// +/// Ordinary activity does not wake this task — a busy run emits an event per +/// agent stream delta. The deadline instead re-reads `Emitter::last_activity()` +/// when it fires and re-arms if the run was active in the meantime. async fn monitor_for_stall( stall_timeout: Duration, stall_token: CancellationToken, shutdown: CancellationToken, - mut activity: watch::Receiver, + emitter: Arc, mut interview_blocks: watch::Receiver, ) { - let mut deadline = TokioInstant::now() + stall_timeout; + let mut deadline = emitter.last_activity() + stall_timeout; loop { - if interview_blocks.borrow_and_update().is_run_blocked() { - tokio::select! { - biased; - () = shutdown.cancelled() => return, - changed = interview_blocks.changed() => { - if changed.is_err() { - return; - } - if !interview_blocks.borrow().is_run_blocked() { - deadline = TokioInstant::now() + stall_timeout; - } - } - } - continue; - } - - let deadline_timer = sleep_until(deadline); - tokio::pin!(deadline_timer); + let blocked = interview_blocks.borrow_and_update().is_run_blocked(); tokio::select! { biased; () = shutdown.cancelled() => return, @@ -65,16 +55,13 @@ async fn monitor_for_stall( if changed.is_err() { return; } + // Blocking parks the timer; unblocking restarts the full budget. deadline = TokioInstant::now() + stall_timeout; } - changed = activity.changed() => { - if changed.is_err() { - return; - } - deadline = TokioInstant::now() + stall_timeout; - } - () = &mut deadline_timer => { - if interview_blocks.borrow().is_run_blocked() { + () = sleep_until(deadline), if !blocked => { + let extended = emitter.last_activity() + stall_timeout; + if extended > deadline { + deadline = extended; continue; } stall_token.cancel(); @@ -253,7 +240,7 @@ pub async fn execute(init: Initialized) -> Executed { stall_timeout, token.clone(), shutdown.clone(), - emitter.subscribe_activity(), + emitter, interview_blocks, )); Some((shutdown, task)) diff --git a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs index a4a15f467..8b77767ce 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs @@ -1435,7 +1435,7 @@ async fn stall_watchdog_starts_a_fresh_deadline_after_human_input() { Duration::from_millis(50), stall_token.clone(), shutdown.clone(), - emitter.subscribe_activity(), + Arc::clone(&emitter), blocker.subscribe(), )); tokio::task::yield_now().await; From fae39d9fd67af5a35302fd2be4988d68bc76d94b Mon Sep 17 00:00:00 2001 From: Release Repro Date: Sat, 1 Aug 2026 10:08:03 -0400 Subject: [PATCH 3/3] refactor(workflow): tidy stall watchdog wiring and interview naming Second pass, from the remaining review findings. - Wrap the stall watchdog in a `StallWatchdog` type. The call site kept two parallel `Option`s derived from the same condition and threaded out an `Option<(CancellationToken, JoinHandle<()>)>`. `monitor_for_stall` also took two same-typed `CancellationToken` params pointing opposite directions, where swapping them compiles and yields a run that silently never stalls. - Rename `WorkflowAgentQuestionRuntime::stage_id` and `PendingAgentQuestionBatch::stage_id` to `node_id`. They hold `node.id`, and the previous commit put them two lines from `stage_scope.stage_id()`, which returns a real `StageId`. - Widen the two real-time interview tests. `node_timeout_excludes_ human_input_wait` allowed 20ms of active work against a 50ms budget, which is tight enough to flake under parallel nextest load. The blocked wait still outruns the timeout, so both still fail if the pause regresses. Co-Authored-By: Claude Opus 5 (1M context) --- .../fabro-workflow/src/interview_runtime.rs | 27 ++++--- .../fabro-workflow/src/pipeline/execute.rs | 80 +++++++++++++------ .../src/pipeline/execute/tests.rs | 12 ++- 3 files changed, 78 insertions(+), 41 deletions(-) diff --git a/lib/components/fabro-workflow/src/interview_runtime.rs b/lib/components/fabro-workflow/src/interview_runtime.rs index dc2855864..69f3e0899 100644 --- a/lib/components/fabro-workflow/src/interview_runtime.rs +++ b/lib/components/fabro-workflow/src/interview_runtime.rs @@ -152,7 +152,10 @@ pub(crate) struct WorkflowAgentQuestionRuntime { interviewer: Arc, emitter: Arc, stage_scope: StageScope, - stage_id: String, + /// Graph node id, reported as the `stage` on interview events. Distinct + /// from `stage_scope.stage_id()`, which is the visit-qualified `StageId` + /// used to key block state. + node_id: String, blocker: Arc, } @@ -162,14 +165,14 @@ impl WorkflowAgentQuestionRuntime { interviewer: Arc, emitter: Arc, stage_scope: StageScope, - stage_id: impl Into, + node_id: impl Into, blocker: Arc, ) -> Self { Self { interviewer, emitter, stage_scope, - stage_id: stage_id.into(), + node_id: node_id.into(), blocker, } } @@ -183,7 +186,7 @@ struct PreparedQuestion { struct PendingAgentQuestionBatch { emitter: Arc, stage_scope: StageScope, - stage_id: String, + node_id: String, questions: Vec<(String, String)>, started_at: Instant, guard: Option, @@ -193,7 +196,7 @@ impl PendingAgentQuestionBatch { fn new( emitter: Arc, stage_scope: StageScope, - stage_id: String, + node_id: String, prepared: &[PreparedQuestion], guard: RunInterviewGuard, started_at: Instant, @@ -201,7 +204,7 @@ impl PendingAgentQuestionBatch { Self { emitter, stage_scope, - stage_id, + node_id, questions: prepared .iter() .map(|prepared_question| { @@ -237,7 +240,7 @@ impl Drop for PendingAgentQuestionBatch { }), question_id: question_id.clone(), question: question.clone(), - stage: self.stage_id.clone(), + stage: self.node_id.clone(), reason: "interrupted".to_string(), duration_ms, }, @@ -274,7 +277,7 @@ impl AgentQuestionRuntime for WorkflowAgentQuestionRuntime { &Event::InterviewStarted { question_id: question.id.clone(), question: question.text.clone(), - stage: self.stage_id.clone(), + stage: self.node_id.clone(), question_type: question.question_type.to_string(), options: question.options.clone(), allow_freeform: question.allow_freeform, @@ -290,7 +293,7 @@ impl AgentQuestionRuntime for WorkflowAgentQuestionRuntime { let cleanup = PendingAgentQuestionBatch::new( Arc::clone(&self.emitter), self.stage_scope.clone(), - self.stage_id.clone(), + self.node_id.clone(), &prepared, self.blocker .block(Arc::clone(&self.emitter), self.stage_scope.stage_id()), @@ -361,7 +364,7 @@ impl WorkflowAgentQuestionRuntime { question.id = internal_question_id(&self.stage_scope, tool_call_id, index); question.options.clone_from(&agent_question.options); question.allow_freeform = agent_question.allow_freeform; - question.stage.clone_from(&self.stage_id); + question.stage.clone_from(&self.node_id); question.metadata.insert( "agent.tool_call_id".to_string(), serde_json::json!(tool_call_id), @@ -401,7 +404,7 @@ impl WorkflowAgentQuestionRuntime { }), question_id: prepared.question.id.clone(), question: prepared.question.text.clone(), - stage: self.stage_id.clone(), + stage: self.node_id.clone(), duration_ms, }, &self.stage_scope, @@ -444,7 +447,7 @@ impl WorkflowAgentQuestionRuntime { actor, question_id: prepared.question.id.clone(), question: prepared.question.text.clone(), - stage: self.stage_id.clone(), + stage: self.node_id.clone(), reason: reason.to_string(), duration_ms, }, diff --git a/lib/components/fabro-workflow/src/pipeline/execute.rs b/lib/components/fabro-workflow/src/pipeline/execute.rs index 60d07b070..3b0fb81f9 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute.rs @@ -5,6 +5,7 @@ use fabro_core::executor::ExecutorBuilder; use fabro_core::handler::NodeHandler; use fabro_core::state::ExecutionState; use tokio::sync::watch; +use tokio::task::JoinHandle; use tokio::time::{Instant as TokioInstant, sleep_until}; use tokio_util::sync::CancellationToken; @@ -30,6 +31,50 @@ fn seed_context_from_checkpoint(checkpoint: Option<&Checkpoint>) -> Context { context } +/// Background watchdog that cancels a run which stops emitting events. +struct StallWatchdog { + /// Cancelled by the monitor once the run stalls. Handed to the executor. + stall_token: CancellationToken, + /// Cancelled by us to stop the monitor once the run finishes. + shutdown: CancellationToken, + task: JoinHandle<()>, +} + +impl StallWatchdog { + fn spawn( + stall_timeout: Duration, + emitter: Arc, + interview_blocks: watch::Receiver, + ) -> Self { + let stall_token = CancellationToken::new(); + let shutdown = CancellationToken::new(); + emitter.touch(); + let task = tokio::spawn(monitor_for_stall( + stall_timeout, + stall_token.clone(), + shutdown.clone(), + emitter, + interview_blocks, + )); + Self { + stall_token, + shutdown, + task, + } + } + + fn stall_token(&self) -> CancellationToken { + self.stall_token.clone() + } + + async fn stop(self) { + self.shutdown.cancel(); + if let Err(error) = self.task.await { + tracing::error!(error = ?error, "stall watchdog task failed"); + } + } +} + /// Cancels `stall_token` once the run goes `stall_timeout` without emitting an /// event. Waiting on human input suspends the timer, and the first unblock /// starts a fresh full deadline. @@ -228,31 +273,19 @@ pub async fn execute(init: Initialized) -> Executed { None }; - let stall_timeout_opt = graph.stall_timeout(); - let stall_token = stall_timeout_opt.map(|_| CancellationToken::new()); - let stall_watchdog = - if let (Some(stall_timeout), Some(ref token)) = (stall_timeout_opt, &stall_token) { - let shutdown = CancellationToken::new(); - let emitter = Arc::clone(&engine.run.emitter); - let interview_blocks = engine.run.interview_blocker.subscribe(); - emitter.touch(); - let task = tokio::spawn(monitor_for_stall( - stall_timeout, - token.clone(), - shutdown.clone(), - emitter, - interview_blocks, - )); - Some((shutdown, task)) - } else { - None - }; + let stall_watchdog = graph.stall_timeout().map(|stall_timeout| { + StallWatchdog::spawn( + stall_timeout, + Arc::clone(&engine.run.emitter), + engine.run.interview_blocker.subscribe(), + ) + }); let mut builder = ExecutorBuilder::new(handler as Arc>) .lifecycle(Box::new(lifecycle)); builder = builder.cancel_token(run_options.cancel_token.clone()); - if let Some(token) = stall_token.clone() { + if let Some(token) = stall_watchdog.as_ref().map(StallWatchdog::stall_token) { builder = builder.stall_token(token); } if let Some(limit) = max_node_visits { @@ -262,11 +295,8 @@ pub async fn execute(init: Initialized) -> Executed { let executor = builder.build(); let result = executor.run(&wf_graph, state).await; - if let Some((shutdown, task)) = stall_watchdog { - shutdown.cancel(); - if let Err(error) = task.await { - tracing::error!(error = ?error, "stall watchdog task failed"); - } + if let Some(watchdog) = stall_watchdog { + watchdog.stop().await; } let (outcome, final_context) = match result { diff --git a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs index 8b77767ce..af31798e3 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs @@ -1460,12 +1460,14 @@ async fn stall_watchdog_starts_a_fresh_deadline_after_human_input() { #[tokio::test] async fn stall_watchdog_suspends_while_run_waits_for_human_input() { let dir = tempfile::tempdir().unwrap(); - let graph = interview_wait_graph(Duration::from_millis(50), None); + // The blocked wait outruns the stall timeout, so this only passes if the + // watchdog stays suspended and then restarts on a fresh deadline. + let graph = interview_wait_graph(Duration::from_millis(300), None); let mut registry = make_registry(); registry.register( "interview_wait", Box::new(InterviewWaitHandler { - wait_ms: 150, + wait_ms: 500, active_ms: 10, }), ); @@ -1486,12 +1488,14 @@ async fn stall_watchdog_suspends_while_run_waits_for_human_input() { #[tokio::test] async fn node_timeout_excludes_human_input_wait() { let dir = tempfile::tempdir().unwrap(); - let graph = interview_wait_graph(Duration::ZERO, Some(Duration::from_millis(50))); + // The blocked wait outruns the node timeout, but the active work is well + // inside it, so this only fails if the interview wait is being charged. + let graph = interview_wait_graph(Duration::ZERO, Some(Duration::from_millis(300))); let mut registry = make_registry(); registry.register( "interview_wait", Box::new(InterviewWaitHandler { - wait_ms: 150, + wait_ms: 500, active_ms: 20, }), );