scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-07 03:00:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Merge pull request #693 from fabro-sh/fix/human-input-timeout-accounting

Browse source 
Pause workflow timeouts during human input
This commit is contained in:
Bryan Helmkamp 2026-08-01 10:13:49 -04:00 • committed by GitHub
commit 41cd1aac49
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 602 additions and 145 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
docs/public/reference/dot-language.mdx
View file

@ -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` |

2
docs/public/workflows/stages-and-nodes.mdx
View file

@ -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:**

62
lib/components/fabro-workflow/src/event/emitter.rs
View file

@ -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<dyn Fn(&RunEvent) + Send + Sync>;
/// Callback-based event emitter for workflow run events.
pub struct Emitter {
run_id: RunId,
listeners: std::sync::Mutex<Vec<EventListener>>,
/// Epoch milliseconds of the last `emit()` or `touch()` call. 0 until first
/// event.
last_event_at: AtomicI64,
run_id: RunId,
listeners: std::sync::Mutex<Vec<EventListener>>,
/// 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,
);
}
}

14
lib/components/fabro-workflow/src/handler/human.rs
View file

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

221
lib/components/fabro-workflow/src/interview_runtime.rs
View file

@ -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<StageId, usize>,
}
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<InterviewBlockState>,
/// 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<Self>, emitter: Arc<Emitter>) -> RunInterviewGuard {
if self.unresolved_interviews.fetch_add(1, Ordering::AcqRel) == 0 {
pub(crate) fn subscribe(&self) -> watch::Receiver<InterviewBlockState> {
self.state.subscribe()
}
pub(crate) fn block(
self: &Arc<Self>,
emitter: Arc<Emitter>,
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<RunInterviewBlocker>,
emitter: Arc<Emitter>,
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<dyn Interviewer>,
emitter: Arc<Emitter>,
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<RunInterviewBlocker>,
}
@ -106,14 +165,14 @@ impl WorkflowAgentQuestionRuntime {
interviewer: Arc<dyn Interviewer>,
emitter: Arc<Emitter>,
stage_scope: StageScope,
stage_id: impl Into<String>,
node_id: impl Into<String>,
blocker: Arc<RunInterviewBlocker>,
) -> 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<Emitter>,
stage_scope: StageScope,
stage_id: String,
node_id: String,
questions: Vec<(String, String)>,
started_at: Instant,
guard: Option<RunInterviewGuard>,
@ -137,7 +196,7 @@ impl PendingAgentQuestionBatch {
fn new(
emitter: Arc<Emitter>,
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));
}
}

153
lib/components/fabro-workflow/src/node_handler.rs
View file

@ -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<F>(
duration: Duration,
stage_id: &StageId,
mut interview_blocks: watch::Receiver<InterviewBlockState>,
future: F,
) -> Option<F::Output>
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);
}
}

149
lib/components/fabro-workflow/src/pipeline/execute.rs
View file

@ -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<Emitter>,
interview_blocks: watch::Receiver<InterviewBlockState>,
) -> 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<Emitter>,
mut interview_blocks: watch::Receiver<InterviewBlockState>,
) {
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<dyn NodeHandler<WorkflowGraph>>)
.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 {

144
lib/components/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -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<Outcome, Error> {
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<Duration>) -> 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<MockSandbox>,
}
@ -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();