scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-07 08:27:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

refactor(workflow): simplify interview block state and stall watchdog

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Release Repro 2026-08-01 09:52:49 -04:00
parent 0de3817836
commit 3fa48c38aa
No known key found for this signature in database
5 changed files with 98 additions and 116 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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<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,
activity_revision: watch::Sender<u64>,
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 {
@ -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<u64> {
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,
);
}
}

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

@ -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<StageId, usize>,
blocked_stages: HashMap<StageId, usize>,
}
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<InterviewBlockState>,
/// 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));
}
}

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

@ -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<F>(
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,

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

@ -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<u64>,
emitter: Arc<Emitter>,
mut interview_blocks: watch::Receiver<InterviewBlockState>,
) {
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))

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

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