Function execute_steps_parallel_resumable

core/src/orchestration/combinators.rs:102–215

Like [`execute_steps_parallel`](super::execute_steps_parallel), but resumable**: progress is journaled to `store` under `workflow_id`, so an interrupted run picks up from the last completed step. On entry, any steps already recorded in a prior checkpoint are skipped and their cached outcomes reused; only the rest are dispatched. As each step completes, the checkpoint is rewritten (the step bounda

(
    executor: Arc<dyn AgentExecutor>,
    specs: Vec<AgentStepSpec>,
    workflow_id: &str,
    store: Arc<dyn SessionStore>,
    event_tx: Option<broadcast::Sender<AgentEvent>>,
)

/// checkpoint is deleted (the workflow is terminal); only a crash leaves one
/// behind for resume.
pub async fn execute_steps_parallel_resumable(
    executor: Arc<dyn AgentExecutor>,
    specs: Vec<AgentStepSpec>,
    workflow_id: &str,
    store: Arc<dyn SessionStore>,
    event_tx: Option<broadcast::Sender<AgentEvent>>,
) -> Vec<StepOutcome> {
    // Prior progress. An unreadable checkpoint — e.g. one written by a newer,
    // incompatible schema version, which the store rejects via
    // `ensure_loadable` — is treated as *no* prior progress: the workflow
    // re-runs from scratch rather than resuming from state it can't interpret.
    // That's a fail-safe (do the work), but surface it rather than swallow it.
    let done: HashMap<String, StepOutcome> = match store.load_workflow_checkpoint(workflow_id).await
    {
        Ok(Some(cp)) => cp.completed(),
        Ok(None) => HashMap::new(),
        Err(e) => {
            tracing::warn!(
                workflow_id = %workflow_id,
                error = %e,
                "workflow checkpoint unreadable; re-running the workflow from scratch"
            );
            HashMap::new()
        }
    };

    let pending: Vec<AgentStepSpec> = specs
        .iter()
        .filter(|s| !done.contains_key(&s.task_id))
        .cloned()
        .collect();
    let labels: Vec<(String, String)> = pending
        .iter()
        .map(|s| (s.task_id.clone(), s.agent.clone()))
        .collect();

    // Accumulator seeded with prior progress; persisted at every step boundary.
    let acc = Arc::new(tokio::sync::Mutex::new(done.clone()));
    let limit = executor.concurrency_hint();
    let workflow_id_owned = workflow_id.to_string();
    let store_steps = Arc::clone(&store);

    let results = run_ordered_parallel_with_limit(pending, limit, move |_idx, spec| {
        let executor = Arc::clone(&executor);
        let event_tx = event_tx.clone();
        let acc = Arc::clone(&acc);
        let store = Arc::clone(&store_steps);
        let workflow_id = workflow_id_owned.clone();
        async move {
            let outcome = executor.execute_step(spec, event_tx).await;
            // Step boundary: record only *successful* steps, so a failed step
            // is retried on resume (its effect didn't complete) while a
            // succeeded step's work is never redone.
            if outcome.success {
                let mut guard = acc.lock().await;
                guard.insert(outcome.task_id.clone(), outcome.clone());
                let checkpoint =
                    WorkflowCheckpoint::from_completed(&workflow_id, &guard, now_epoch_ms());