Pause and resume

You end up with a task that stops before a side effect, waits for a person or an outside event, and then continues from the node where it stopped. The workspace-architect showcase uses this to hold its generated ARCHITECTURE.md until someone approves it:

task.register_transition(review, move |decision| match decision {
    Review::Approved(draft) => write.transitions_with(draft).into(),
    Review::AwaitingApproval => Transition::pause(),
})?;

Source: examples/workspace-architect/src/graph.rs#L125-L128

The review node returns a Review. When the draft is approved, the transition sends it on to the write node. When it still waits for approval, the transition returns Transition::pause(), and run comes back with TaskRunOutcome::Paused instead of an output.

The .into() on the first arm is the same MarkedTransition conversion described in Nodes and transitions. The types live in swiftide::tasks, behind the swiftide-tasks feature flag:

cargo add swiftide@0.34 --features swiftide-tasks

A task pause is not the same as an agent pause. To stop an agent until someone approves a single tool call, see Human in the loop.

Approval gate

The review node is ApprovalGate, a task node that remembers the draft and whether someone approved it:

/// Lets a draft through once it is approved, and otherwise keeps it for review.
///
/// The graph pauses the task when the gate answers [`Review::AwaitingApproval`]. A paused branch
/// re-runs its node on `resume()`, so after [`ApprovalGate::approve`] the same draft passes
/// through. The gate is cheap to evaluate, which is why the pause sits here and not on the
/// expensive render node.
#[derive(Debug, Clone)]
pub struct ApprovalGate {
    state: Arc<Mutex<GateState>>,
    progress: Progress,
}

#[derive(Debug, Default)]
struct GateState {
    approved: bool,
    pending: Option<ArchitectureDraft>,
}

impl ApprovalGate {
10 collapsed lines
    pub fn new(pre_approved: bool, progress: Progress) -> Self {
        let state = GateState {
            approved: pre_approved,
            pending: None,
        };
        Self {
            state: Arc::new(Mutex::new(state)),
            progress,
        }
    }

    /// The draft waiting for a decision, if any.
    pub fn pending_draft(&self) -> Option<ArchitectureDraft> {
        self.state().pending.clone()
    }

    pub fn approve(&self) {
        self.state().approved = true;
    }

3 collapsed lines
    fn state(&self) -> MutexGuard<'_, GateState> {
        self.state.lock().unwrap_or_else(PoisonError::into_inner)
    }
}

#[async_trait]
impl TaskNode for ApprovalGate {
    type Input = ArchitectureDraft;
    type Output = Review;
    type Error = Infallible;

    async fn evaluate(
        &self,
        _node_id: &DynNodeId<Self>,
        draft: &ArchitectureDraft,
    ) -> Result<Review, Infallible> {
        let mut state = self.state();
        if state.approved {
            state.pending = None;
            self.progress.step("review", "draft approved");
            return Ok(Review::Approved(draft.clone()));
        }

        state.pending = Some(draft.clone());
        self.progress.step(
            "review",
            format_args!(
                "draft ready ({} bytes), waiting for approval",
                draft.as_str().len()
            ),
        );
        Ok(Review::AwaitingApproval)
    }
}

Source: examples/workspace-architect/src/approval.rs#L23-L95

A paused branch keeps its current node and that node’s input. On resume(), the runtime evaluates the same node again with the same input, and then runs its transition again. The gate relies on this. The first evaluation stores the draft and answers AwaitingApproval. After approve(), the second evaluation answers Approved with the same draft.

Put the pause on a cheap node. The showcase could pause right after render, but then resume() would call the model and draft the document a second time.

The gate keeps its state in Arc<Mutex<GateState>>. The Mutex is there because evaluate and approve take &self, so plain fields cannot change. The Arc is there because a task node must be Clone, and Mutex is not. With the Arc, clones of the gate share one state. The gate’s error type is Infallible, because waiting is an answer, not a failure.

Resume loop

The command-line interface (CLI) runs the task and loops until it completes or the user declines the draft:

let mut outcome = task.run(root).await?;
loop {
    match outcome {
        TaskRunOutcome::Completed(written) => return Ok(Some(written)),
        TaskRunOutcome::Paused => {
            if !approve_paused_draft(&task)? {
                return Ok(None);
            }
            outcome = task.resume().await?;
        }
    }
}

Source: examples/workspace-architect/src/main.rs#L125-L136

On TaskRunOutcome::Paused, approve_paused_draft (in the same file) finds the gate with task.current_nodes::<ApprovalGate>(), shows the draft and asks for confirmation. On yes it calls gate.approve() and returns true, and the loop calls task.resume(). On no it returns false, and draft_architecture returns Ok(None) and drops the paused task. Nothing is written.

This is the end of a captured run, with claude-sonnet-5-5 as the writer. The draft between the review line and the question is trimmed:

[  68.8s] review   draft ready (11210 bytes), waiting for approval
...
Write this ARCHITECTURE.md? [y/N]
[  68.8s] review   draft approved
[  68.8s] write    .../target/ARCHITECTURE.md (11210 bytes)
Wrote .../target/ARCHITECTURE.md (11210 bytes) in 68.8s.

The --yes flag creates the gate already approved, so the first evaluation answers Approved and the task never pauses.

Current nodes

Task::current_nodes::<T>() gives you the paused node, so you can read or change its state before you resume. This helper delivers the CI result from the next section through the task:

/// Delivers a CI result through the task, for callers that hold the task instead of the node.
pub fn deliver_through_task(task: &Task<String, String>, result: &str) -> anyhow::Result<()> {
    let CurrentNodes::One(ci) = task.current_nodes::<AwaitCi>() else {
        anyhow::bail!("the task is not waiting on exactly one CI result");
    };
    ci.deliver(result);
    Ok(())
}

Source: snippets/src/tasks/pause_and_resume.rs#L92-L99

The method returns the current nodes of type T as a CurrentNodes:

CurrentNodes::None when no current node has that type.
CurrentNodes::One(&T) for exactly one.
CurrentNodes::Multiple(Vec<&T>) for more than one.

Paused branches take precedence. The method returns runnable nodes only when no branch is paused. In a parallel fan-out several branches can pause on the same node type, so handle Multiple there. The references point at the registered node instances.

External callbacks

The same pattern waits for another system, such as a continuous integration (CI) webhook. The node reads a slot that the webhook handler fills, and the transition pauses while the slot is empty:

/// Waits for a CI result that a webhook delivers from outside the task.
#[derive(Debug, Clone, Default)]
pub struct AwaitCi {
    result: Arc<Mutex<Option<String>>>,
}

impl AwaitCi {
    /// Called by the webhook handler. Clones share the slot, so any clone can deliver.
    pub fn deliver(&self, result: impl Into<String>) {
        let mut slot = self.result.lock().unwrap_or_else(PoisonError::into_inner);
        *slot = Some(result.into());
    }
}

#[async_trait]
impl TaskNode for AwaitCi {
    type Input = String;
    type Output = CiStatus;
    type Error = Infallible;

    async fn evaluate(
        &self,
        _node_id: &DynNodeId<Self>,
        commit: &String,
    ) -> Result<CiStatus, Infallible> {
        let slot = self.result.lock().unwrap_or_else(PoisonError::into_inner);
        Ok(match slot.as_deref() {
            Some(result) => CiStatus::Finished(format!("{commit}: {result}")),
            None => CiStatus::Pending,
        })
    }
}

pub async fn wait_for_ci() -> anyhow::Result<String> {
    let ci = AwaitCi::default();

    let mut task = Task::<String, String>::new();
    let push = task
        .register_node_fn(|commit: &String| -> Result<String, NodeError> { Ok(commit.clone()) });
    let wait = task.register_node(ci.clone());

    task.starts_with(push);
    task.register_transition(push, move |commit| wait.transitions_with(commit))?;
    task.register_transition(wait, {
        let finish = task.transitions_to_finish();
        move |status| match status {
            CiStatus::Finished(report) => finish(report),
            CiStatus::Pending => Transition::pause(),
        }
    })?;

    assert_eq!(task.run("4f2a91c").await?, TaskRunOutcome::Paused);
    // One task value holds one run: a second `run` while paused is refused.
    assert!(matches!(
        task.run("77b0e3d").await,
        Err(TaskError::TaskActive)
    ));

    // Later, the webhook handler that holds `ci` receives the result,
    // and the same task value continues.
    ci.deliver("all checks passed");
    let TaskRunOutcome::Completed(report) = task.resume().await? else {
        anyhow::bail!("the CI result was delivered, so the task should finish");
    };

Source: snippets/src/tasks/pause_and_resume.rs#L23-L86

The excerpt registers ci.clone() and keeps ci itself. In a server, ci goes to the webhook handler, and the comment marks where that handler runs. The clone shares the Arc, so deliver fills the slot that the registered node reads. A handler that holds the task instead can use deliver_through_task from the previous section.

Keep the same Task value alive between the pause and the resume. The runtime state is not serializable. If the process restarts, the paused run is gone, so store what you need to start again, such as the commit. In a server, keep paused tasks in a map keyed by job.

One task value holds one run. While the task has paused or queued work, run returns TaskError::TaskActive. Build a new task for the next job, for example with a function like the showcase’s architect_task.

Reruns after reset

reset() discards the current run and queues a new one at the start node, with the last input passed to run. A resume() after it runs that queued run:

let mut task = Task::<i32, i32>::new();
let double = task.register_node_fn(|n: &i32| -> Result<i32, NodeError> { Ok(n * 2) });
task.starts_with(double);
task.register_transition(double, task.transitions_to_finish())?;

assert_eq!(task.run(21).await?, TaskRunOutcome::Completed(42));
// A completed run leaves nothing to continue.
assert!(matches!(task.resume().await, Err(TaskError::NotResumable)));

// `reset` queues a new run at the start node with the last input.
task.reset();
// That queued run counts as active work, so `run` with another input is refused.
assert!(matches!(task.run(5).await, Err(TaskError::TaskActive)));
assert_eq!(task.resume().await?, TaskRunOutcome::Completed(42));

// A clone has the same graph and a fresh runtime of its own.
let mut copy = task.clone();
assert_eq!(copy.run(5).await?, TaskRunOutcome::Completed(10));

Source: snippets/src/tasks/pause_and_resume.rs#L104-L121

After a completed run there is nothing to continue, so resume() returns TaskError::NotResumable.
reset() also discards a paused run. Use it to start over with the same input instead of continuing.
The queued run counts as active work, so run(other_input) after reset() returns TaskError::TaskActive. To start over with a different input, use task.clone(). Calling starts_with again also clears the runtime state.
reset() before the first run has no input to queue. The resume() that follows returns TaskError::NotResumable.
A failed run leaves no state to resume. reset() and resume() retry it with the same input.
task.clone() copies the graph with a fresh runtime, without the paused run. Node state behind an Arc is shared between the copies.

Pauses in parallel branches

A pause stops the whole task, not only its branch. When a branch asks to pause, the runtime starts no new branches. Parallel branches that are already running finish their current node and transition, and their next steps wait in the queue. Then run returns Paused, unless one of those running branches reaches the finish transition. In that case run returns Completed and the paused branch is discarded.

If a running branch fails after another branch asked to pause, run returns that error. The paused run is lost, and resume() has nothing to continue.

resume() puts the paused branches back in the queue in the order they were created and continues with all queued work. A join runs only when every branch has arrived, so a paused branch also holds back its join. Fan-out and join covers branches and joins.

Reference

Agents as task nodes