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;