Pause workflow timeouts during human input

[brynary]

Summary

• Treat unresolved human input as a valid blocked state instead of a workflow stall.
• Suspend the run stall watchdog while any interview is pending, then start a fresh monotonic deadline after the final unblock.
• Exclude the current stage human-input wait from its executor-enforced node timeout without pausing timeout budgets for parallel sibling stages.
• Keep human-node timeout behavior as the response deadline and document the distinction.

Implementation notes

• RunInterviewBlocker now publishes ref-counted run and StageId block state.
• The event emitter publishes activity revisions so the watchdog can react to activity without wall-clock polling.
• Interview guards clear block state on normal completion and cancellation.

Validation

• cargo nextest run --workspace — 7,683 passed, 203 skipped
• cargo +nightly-2026-04-14 clippy --workspace --all-targets --all-features -- -D warnings
• cargo +nightly-2026-04-14 fmt --check --all
• git diff --check

[brynary]

Reviewer note: the subtle part of this change is the transition boundary, not the timer itself. Please focus on these invariants:

1. Block state becomes visible before run.blocked. On final resolution, run.unblocked is emitted before block state clears. This gives the watchdog a full fresh deadline after unblocking.
2. Node timeout suspension is keyed by StageId. A question in one parallel branch must not pause a sibling branch timeout.
3. Concurrent and cancelled interviews must leave one balanced run.blocked / run.unblocked pair and no stale blocked stage.

The focused coverage is in stall_watchdog_starts_a_fresh_deadline_after_human_input, node_timeout_does_not_pause_for_another_stage_interview, and cancelling_agent_question_unblocks_its_stage.

[Copilot]

Pull request overview

This PR updates the workflow runtime to treat pending human input as a first-class “blocked” state, so that workflow stall detection and node timeouts don’t incorrectly fire while an interview is awaiting answers. It introduces run- and stage-scoped interview block tracking, shifts the stall watchdog to react to activity revisions instead of wall-clock polling, and clarifies timeout semantics in docs.

Changes:

• Add ref-counted run + StageId interview block state and publish it via watch to pause stall detection and exclude interview waits from node timeouts.
• Replace stall watchdog wall-clock polling with an activity-revision subscription and restart the monotonic deadline after unblocking.
• Document that agent question waits do not consume node timeout (and that human-node timeout remains the response deadline).

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
lib/components/fabro-workflow/src/pipeline/execute/tests.rs	Adds regression tests for stall watchdog suspension/restart and node-timeout exclusion during interview waits.
lib/components/fabro-workflow/src/pipeline/execute.rs	Reworks stall watchdog into monitor_for_stall() driven by watch activity + interview block state.
lib/components/fabro-workflow/src/node_handler.rs	Introduces timeout_excluding_interview_wait() to pause a stage’s timeout budget during its own interview wait.
lib/components/fabro-workflow/src/interview_runtime.rs	Implements run/stage block refcounting and exposes block state via watch for time-budget control.
lib/components/fabro-workflow/src/handler/human.rs	Passes StageId into interview blocking so block state can be tracked per-stage.
lib/components/fabro-workflow/src/event/emitter.rs	Adds watch-based activity revisions so watchdogs can react to activity without wall-clock polling.
docs/public/workflows/stages-and-nodes.mdx	Documents that agent-question waiting time does not count toward node timeout.
docs/public/reference/dot-language.mdx	Clarifies node timeout semantics for agent vs human nodes.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated no new comments.

Suppressed comments (1)

lib/components/fabro-workflow/src/interview_runtime.rs:121

• RunInterviewBlocker::resolved emits RunUnblocked before updating the watch state. While the event is being dispatched, subscribers (stall watchdog / node timeout exclusion) will still see the run as blocked and may keep timeouts paused longer than intended. It also briefly makes the event stream inconsistent with the in-memory blocked state.

Update the watch state first, then emit RunUnblocked if the run is now unblocked.

    fn resolved(&self, emitter: &Emitter, stage_id: &StageId) {
        let _transition = self
            .transitions
            .lock()
            .expect("interview transition mutex should not be poisoned");
        let should_emit_unblocked = self.state.borrow().unresolved_interviews == 1;
        if should_emit_unblocked {
            emitter.emit(&Event::RunUnblocked);
        }
        self.state.send_modify(|state| state.resolve(stage_id));
    }

[brynary]

Ran a simplification pass over the branch. Net −18 lines, all 7,741 workspace tests pass, clippy and fmt clean.

Removed redundant state

• InterviewBlockState carried both unresolved_interviews and blocked_stages. The counter was just blocked_stages.values().sum(), and is_run_blocked() is !blocked_stages.is_empty(). Dropped the counter.
• Emitter tracked activity twice: last_event_at (epoch millis) and the new activity_revision watch channel. last_event_at() had no callers left anywhere in the workspace after this PR, and epoch_millis() existed only to feed it. Both removed.

Hot path

record_activity() runs on every event, and handler/llm/api.rs calls touch() on every agent stream delta. The watch channel made each delta take a write lock and wake the watchdog task, which then re-armed its timer — per token. Replaced it with a monotonic AtomicU64 timestamp. The watchdog now samples last_activity() when its own deadline fires and re-arms only if the run was active since. The hot path is one clock read and one relaxed store; the watchdog wakes on block-state transitions and its deadline, nothing else. Detection stays precise (no 2× overshoot) and start_paused tests still work because the timestamp uses tokio::time::Instant.

Event/state ordering (addresses the Copilot comment)

Both transitions now publish state before emitting, so a listener reading subscribe() from an event callback never sees state that disagrees with the event. This does not weaken invariant 1: the watchdog restarts its full deadline on the unblock transition, not on the event, so it still gets a fresh budget. stall_watchdog_starts_a_fresh_deadline_after_human_input still passes unchanged.

Panics

• InterviewBlockState::resolve had three .expect()s and runs from Drop — a panic there during unwind aborts the process. Made it non-panicking; RunInterviewGuard already guarantees resolve-once.
• timeout_excluding_interview_wait called .expect() on changed(). Swallowing the error would have busy-looped, so it now falls back to a plain deadline on the remaining budget.

Duplication

monitor_for_stall and timeout_excluding_interview_wait each had near-identical blocked and unblocked select! blocks joined by continue. Collapsed each into one loop using a branch precondition (if !blocked) to park the timer while blocked. biased ordering preserves the "state change wins over an expired deadline" check the explicit re-check used to provide.

Also: stage_id is now computed only when the node actually has a timeout.

All three invariants from the reviewer note still hold, and the three named tests pass unchanged.

[Copilot]

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated no new comments.

Suppressed comments (3)

lib/components/fabro-workflow/src/interview_runtime.rs:113

• RunInterviewBlocker::resolved also uses expect(...) on the transitions mutex, so the same listener-panic poisoning risk applies during unblocking (and would prevent recovery on subsequent resolves).

        let _transition = self
            .transitions
            .lock()
            .expect("interview transition mutex should not be poisoned");

lib/components/fabro-workflow/src/interview_runtime.rs:90

• RunInterviewBlocker::block holds a poisoned std::sync::Mutex guard via expect(...). Because the critical section includes emitter.emit(...) (which runs arbitrary listeners), a panic in any listener would poison this mutex and make all future block/unblock transitions panic, potentially breaking run execution.

This issue also appears on line 110 of the same file.

        let _transition = self
            .transitions
            .lock()
            .expect("interview transition mutex should not be poisoned");

lib/components/fabro-workflow/src/pipeline/execute.rs:60

• monitor_for_stall exits permanently when the interview_blocks watch channel closes (changed.is_err()), which disables stall detection instead of falling back to activity-only monitoring. This is inconsistent with timeout_excluding_interview_wait (which falls back to a plain timeout when the blocker disappears) and could allow a hung run to never trip the watchdog if the sender is dropped unexpectedly.

            changed = interview_blocks.changed() => {
                if changed.is_err() {
                    return;
                }

[brynary]

Second pass, from the remaining review findings.

• StallWatchdog type. The call site kept two parallel Options derived from the same condition and threaded out an Option<(CancellationToken, JoinHandle<()>)>. monitor_for_stall also took two same-typed CancellationToken params pointing opposite directions — stall_token is an output, shutdown is an input — where swapping them compiles and yields a run that silently never stalls. Now one Option<StallWatchdog> with spawn / stall_token / stop.
• stage_id → node_id on WorkflowAgentQuestionRuntime and PendingAgentQuestionBatch. Both hold node.id, and the first commit put them two lines from stage_scope.stage_id(), which returns a real StageId.
• Widened the two real-time interview tests. node_timeout_excludes_human_input_wait allowed 20ms of active work against a 50ms budget — tight enough to flake under parallel nextest load. Now 20ms against 300ms, with the blocked wait at 500ms so it still outruns the timeout and both tests still fail if the pause regresses.

Considered and skipped: wrapping the block-state receiver in a BlockWatch handle to privatize InterviewBlockState. It would add two types to remove one parameter across two call sites — not a net simplification. Happy to add it if you'd rather have the encapsulation.

cargo nextest run --workspace 7,741 passed; clippy and fmt clean.

[brynary]

Correction on my first comment's sizing of the emitter change, so the cost/benefit is accurate.

I said the watch channel woke the watchdog "per token." Sender-side that's right, but receiver wakes coalesce — watch collapses N revisions into one changed() — so the absolute CPU was smaller than that framing implies: roughly 0.2ms/s per streaming stage, about 0.02% of a core, ~0.2% across 10 parallel branches. The real case for the change is the non-coalescing sender-side cost plus avoidable timer-driver churn, not throughput.

Sender-side per event, even with zero subscribers: an RwLock write plus BigNotify::notify_waiters() looping 8 Notify objects, each taking its waiter-list mutex before the empty fast path — ~9 lock pairs, ~150-300ns, against a ~5ns atomic store. Event rate is ~50-200/s per agent stage, since TextDelta/ReasoningDelta fire once per LLM SSE chunk and api.rs:583 touches on every one.

Exec output is not a factor either way — command streaming goes to CommandLogRecorder, not the emitter, and ToolCallOutputDelta fires once per tool call rather than per line.

Two things worth stating explicitly since they were the load-bearing details:

• The watchdog re-arms to last_activity() + stall_timeout, not now + stall_timeout. Re-arming to now would have reintroduced the original bug shape, where a stall beginning just after a wake is detected up to 2x late — with the 1800s default that is roughly an hour.
• Monotonic time is also the right basis for a second reason: the old SystemTime basis meant an NTP step or a suspend/resume could false-trigger or mask a stall.

Also considered and skipped, both pre-existing and outside this diff:

• api.rs:583 records activity twice for non-streaming events (unconditional touch(), then emit_scoped -> record_activity). Now that record_activity is one clock read and a relaxed store, the saving is ~25ns on low-frequency events, and gating the touch() on is_streaming_noise() would silently stop SessionStarted/ProcessingEnd/SessionEnded from resetting the watchdog. Not worth the behavior risk.
• Hoisting a pinned Sleep and using reset() in timeout_excluding_interview_wait. That loop only iterates on interview transitions, which are human-paced, so it is a couple of timer ops per human interaction.

[Copilot]

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated 1 comment.