diff --git a/docs/public/reference/dot-language.mdx b/docs/public/reference/dot-language.mdx index c31626c8c..d4194346d 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..d334b79a6 100644 --- a/lib/components/fabro-workflow/src/event/emitter.rs +++ b/lib/components/fabro-workflow/src/event/emitter.rs @@ -1,31 +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::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, + 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 { @@ -34,8 +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)) - .finish() + .field( + "last_activity_ms", + &self.last_activity_ms.load(Ordering::Relaxed), + ) + .finish_non_exhaustive() } } @@ -51,7 +51,8 @@ impl Emitter { Self { run_id, listeners: std::sync::Mutex::new(Vec::new()), - last_event_at: AtomicI64::new(0), + activity_origin: Instant::now(), + last_activity_ms: AtomicU64::new(0), } } @@ -118,7 +119,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 +131,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. @@ -145,16 +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)) } - /// 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.last_event_at.store(epoch_millis(), Ordering::Relaxed); + 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_activity_ms.store( + millis_u64(self.activity_origin.elapsed()), + Ordering::Relaxed, + ); } } 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..69f3e0899 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,93 @@ 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 +/// 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 { + blocked_stages: HashMap, +} + +impl InterviewBlockState { + pub(crate) fn is_run_blocked(&self) -> bool { + !self.blocked_stages.is_empty() + } + + 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.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) { + 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); + } + } +} + +/// 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. +/// +/// 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 { - unresolved_interviews: AtomicUsize, + 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<()>, } 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 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, }); @@ -39,30 +101,23 @@ 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 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); } } } @@ -70,6 +125,7 @@ impl RunInterviewBlocker { pub(crate) struct RunInterviewGuard { blocker: Arc, emitter: Arc, + stage_id: StageId, resolved: bool, } @@ -80,7 +136,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; } } @@ -96,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, } @@ -106,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, } } @@ -127,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, @@ -137,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, @@ -145,7 +204,7 @@ impl PendingAgentQuestionBatch { Self { emitter, stage_scope, - stage_id, + node_id, questions: prepared .iter() .map(|prepared_question| { @@ -181,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, }, @@ -218,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, @@ -234,9 +293,10 @@ 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.blocker + .block(Arc::clone(&self.emitter), self.stage_scope.stage_id()), interview_start, ); let ask_all = future::join_all( @@ -304,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), @@ -344,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, @@ -387,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, }, @@ -528,17 +588,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 +642,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 +688,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..70a35a34b 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,61 @@ 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, timeout}; 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; +/// 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, + mut interview_blocks: watch::Receiver, + future: F, +) -> Option +where + F: Future, +{ + tokio::pin!(future); + let mut remaining = duration; + + loop { + let blocked = interview_blocks + .borrow_and_update() + .is_stage_blocked(stage_id); + let active_started = Instant::now(); + tokio::select! { + biased; + output = &mut future => return Some(output), + changed = interview_blocks.changed() => { + 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()); + } + } + () = sleep(remaining), if !blocked => return None, + } + } +} + /// Production node handler that bridges fabro-core's NodeHandler to the /// existing fabro-workflow Handler trait via EngineServices. /// @@ -69,20 +114,26 @@ pub(crate) async fn execute_single_attempt( 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 stage_id = StageScope::for_handler(&wf_context, &node.id).stage_id(); + 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 +237,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 +288,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..3b0fb81f9 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute.rs @@ -1,18 +1,21 @@ 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::task::JoinHandle; +use tokio::time::{Instant as TokioInstant, sleep_until}; use tokio_util::sync::CancellationToken; 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; use crate::node_handler::WorkflowNodeHandler; use crate::outcome::Outcome; @@ -28,6 +31,91 @@ 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. +/// +/// 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, + emitter: Arc, + mut interview_blocks: watch::Receiver, +) { + let mut deadline = emitter.last_activity() + stall_timeout; + + loop { + let blocked = interview_blocks.borrow_and_update().is_run_blocked(); + tokio::select! { + biased; + () = shutdown.cancelled() => return, + changed = interview_blocks.changed() => { + if changed.is_err() { + return; + } + // Blocking parks the timer; unblocking restarts the full budget. + deadline = TokioInstant::now() + stall_timeout; + } + () = sleep_until(deadline), if !blocked => { + let extended = emitter.last_activity() + stall_timeout; + if extended > deadline { + deadline = extended; + continue; + } + stall_token.cancel(); + return; + } + } + } +} + /// EXECUTE phase: run the workflow graph. /// /// Infallible at the function level — engine errors are captured in `outcome`. @@ -185,54 +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_shutdown = - 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(); - 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) - } 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 { @@ -242,8 +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) = stall_shutdown { - shutdown.cancel(); + 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 6fd73a17c..af31798e3 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,94 @@ 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(), + Arc::clone(&emitter), + 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(); + // 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: 500, + 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(); + // 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: 500, + 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();