feat(engine): continue a failed run from the node that failed

src/engine/resumable.rs

@@ -257,8 +257,10 @@ pub async fn resume_with_checkpointer(
         capabilities,
         checkpointer,
         thread_id,
-        newly_approved,
-        Vec::new(),
+        Continuation::Approvals {
+            approved: newly_approved,
+            rejected: Vec::new(),
+        },
         None,
         &observer,
     )
@@ -405,8 +407,10 @@ pub async fn resume_with_checkpointer_journaled_observed(
         capabilities,
         checkpointer,
         thread_id,
-        newly_approved,
-        rejected,
+        Continuation::Approvals {
+            approved: newly_approved,
+            rejected,
+        },
         Some(journal),
         observer,
     )
@@ -422,18 +426,73 @@ pub async fn resume_with_checkpointer_journaled_observed(
     })
 }
 
-/// Shared implementation of the checkpointed resume path: rebuilds the graph
+/// Why a checkpointed thread is being continued.
+///
+/// The two boundaries a run can stop at need different things delivered back
+/// into it, and nothing else about continuing differs — same graph rebuild,
+/// same checkpointer, same fold. Keeping the difference in one value is what
+/// lets the failure path reuse the whole of the approval path rather than
+/// growing a parallel copy of it.
+enum Continuation {
+    /// The run paused at approval gates. Carries the operator's decisions.
+    Approvals {
+        /// Gates the operator allowed.
+        approved: Vec<String>,
+        /// Gates the operator refused.
+        rejected: Vec<String>,
+    },
+    /// The run *failed*. Re-run the node that failed and the not-yet-run tail
+    /// of its step, carrying no value — there is no decision to deliver, only
+    /// work to redo.
+    Retry,
+}
+
+impl Continuation {
+    /// The command that carries this continuation into the runtime.
+    fn command(self) -> Command<Value> {
+        match self {
+            Self::Approvals { approved, rejected } => {
+                // Approvals recorded for downstream visibility. On resume the
+                // interrupted gate is approved because the resume value reaches
+                // it via `NodeContext::resume`; the `with_update` mirrors
+                // `ResumableRun::resume` (the runtime ignores it on resume, so
+                // the resume value is the real approval channel).
+                let update = json!({
+                    "run": { "trigger": { "approvals": approved.clone() } }
+                });
+                if !rejected.is_empty() {
+                    tracing::info!(?rejected, "resuming with denied approval gate(s)");
+                }
+                // Always a structured resume value carrying the explicit
+                // `approved` and `rejected` gate id lists. Each interrupted gate
+                // decides for itself: gates in `approved` proceed, gates in
+                // `rejected` route to their `error` port (or fail), and gates in
+                // neither stay pending. This is essential when several parallel
+                // gates are interrupted and the host resolves only some of them
+                // — a bare `true` would blanket-approve every interrupt
+                // regardless of the host's decision.
+                let value = json!({ "approved": approved, "rejected": rejected });
+                Command::resume(value).with_update(update)
+            }
+            // Deliberately empty. A failed node is re-entered from its start
+            // with the state the boundary committed; a resume *value* would be
+            // delivered to `NodeContext::resume` and read as an approval
+            // decision by any gate that happened to be in the pending set.
+            Self::Retry => Command::new(),
+        }
+    }
+}
+
+/// Shared implementation of the checkpointed continue path: rebuilds the graph
 /// (optionally journaled), re-attaches the same `checkpointer`, and resumes
 /// `thread_id`. Returns the outcome plus the resumed execution's
 /// runtime-minted run ids.
-#[allow(clippy::too_many_arguments)]
 async fn resume_with_checkpointer_inner(
     workflow: &CompiledWorkflow,
     capabilities: &Capabilities,
     checkpointer: Arc<dyn Checkpointer<Value>>,
     thread_id: &str,
-    newly_approved: Vec<String>,
-    rejected: Vec<String>,
+    continuation: Continuation,
     journal: Option<Arc<dyn GraphEventJournal>>,
     observer: &Arc<dyn RunObserver>,
 ) -> Result<(RunOutcome, GraphRunIds)> {
@@ -457,32 +516,7 @@ async fn resume_with_checkpointer_inner(
         &config,
     )?;
 
-    // Approvals recorded for downstream visibility. On resume the interrupted
-    // gate is approved because the resume value reaches it via
-    // `NodeContext::resume`; the `with_update` mirrors `ResumableRun::resume`
-    // (the runtime ignores it on resume, so the resume value is the real
-    // approval channel).
-    let approvals_update = json!({
-        "run": { "trigger": { "approvals": newly_approved.clone() } }
-    });
-    if !rejected.is_empty() {
-        tracing::info!(?rejected, "resuming with denied approval gate(s)");
-    }
-    // Always deliver a structured resume value carrying the explicit `approved`
-    // and `rejected` gate id lists. the runtime ignores the `with_update` state
-    // write on resume, so this value is the sole approval channel and each
-    // interrupted gate decides for itself: gates in `approved` proceed, gates in
-    // `rejected` route to their `error` port (or fail), and gates in neither stay
-    // pending. This is essential when several parallel gates are interrupted and
-    // the host resolves only some of them — a bare `true` would blanket-approve
-    // every interrupt regardless of the host's decision.
-    let resume_value = json!({ "approved": newly_approved, "rejected": rejected });
-    let execution = compiled
-        .resume(
-            thread_id,
-            Command::resume(resume_value).with_update(approvals_update),
-        )
-        .await;
+    let execution = compiled.resume(thread_id, continuation.command()).await;
     let execution = match execution {
         Ok(execution) => execution,
         Err(error) => {
@@ -516,3 +550,206 @@ async fn resume_with_checkpointer_inner(
         graph_run_ids,
     ))
 }
+
+/// What a failed run left behind, and what it would take to continue it.
+///
+/// A run that fails does not necessarily lose its work. On a checkpointed
+/// thread the runtime folds the branches that already completed into committed
+/// state and writes a **failure boundary** — a checkpoint whose pending nodes
+/// are the node that failed and the not-yet-run tail of its step. Everything
+/// before it is durable and does not have to happen twice.
+///
+/// The engine reports the failure as an `Err`, which is the right shape for a
+/// caller that just wants to know the run did not finish. This is the question
+/// that error cannot answer: *is there something to continue, and where did it
+/// stop?* Read it after a failed run to decide between fixing and retrying,
+/// and re-running from the trigger.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct FailureBoundary {
+    /// The node whose handler failed.
+    pub failed_node: String,
+    /// The error as the runtime rendered it, for diagnosis.
+    pub error: String,
+    /// The checkpoint holding the committed prefix — the id
+    /// [`ResumeTarget::Checkpoint`](crate::graph::ResumeTarget) addresses.
+    pub checkpoint_id: String,
+    /// Which superstep the run reached.
+    pub step: usize,
+    /// The nodes a continue would run: the failed one, and whatever else in
+    /// its step had not run when it aborted.
+    pub pending: Vec<String>,
+}
+
+/// Read the failure boundary a thread's latest checkpoint records, if it is one.
+///
+/// `Ok(None)` for a thread that has no checkpoint, or whose latest is an
+/// ordinary boundary — a completed run, or one paused at an approval gate.
+/// Those are not failures and have nothing to continue *from a failure*.
+///
+/// Deliberately a separate read rather than a field on the error. A failed run
+/// already returns [`EngineError`], every caller handles that, and widening it
+/// would make every one of them carry a concept most do not use. Asking
+/// afterwards also reads the way the decision is actually made: the run
+/// failed — is it worth continuing?
+///
+/// # Errors
+/// When the checkpointer cannot be read.
+pub async fn failure_boundary(
+    checkpointer: &Arc<dyn Checkpointer<Value>>,
+    thread_id: &str,
+) -> Result<Option<FailureBoundary>> {
+    let checkpoint = checkpointer
+        .get(thread_id, None)
+        .await
+        .map_err(|error| EngineError::Capability(error.to_string()))?;
+    let Some(checkpoint) = checkpoint else {
+        return Ok(None);
+    };
+    // `failed_node` is what makes a boundary a *failure* boundary — an
+    // interrupt boundary and a terminal one both lack it. Reading the key
+    // rather than a status field keeps this to one checkpoint load.
+    let Some(failed_node) = checkpoint
+        .metadata
+        .get("failed_node")
+        .and_then(Value::as_str)
+    else {
+        return Ok(None);
+    };
+    Ok(Some(FailureBoundary {
+        failed_node: failed_node.to_string(),
+        error: checkpoint
+            .metadata
+            .get("error")
+            .and_then(Value::as_str)
+            .unwrap_or_default()
+            .to_string(),
+        checkpoint_id: checkpoint.checkpoint_id.clone(),
+        step: checkpoint
+            .metadata
+            .get("step")
+            .and_then(Value::as_u64)
+            .and_then(|step| usize::try_from(step).ok())
+            .unwrap_or(0),
+        pending: checkpoint
+            .next_nodes
+            .iter()
+            .map(ToString::to_string)
+            .collect(),
+    }))
+}
+
+/// Continue a **failed** run from where it stopped: re-run the node that
+/// failed and the not-yet-run tail of its step, on the state the failure
+/// boundary committed.
+///
+/// The counterpart of [`resume_with_checkpointer`] for the failure path. That
+/// one answers a pause with a decision; this one answers a break with another
+/// go, and carries no resume value — there is nothing to decide, only work to
+/// redo.
+///
+/// Two reasons to reach for this over re-running the workflow:
+///
+/// * **Side effects.** A prefix that posted a comment, opened a pull request
+///   or charged something does not do it twice. Re-running from the trigger is
+///   not a neutral choice for a graph with effects in it; it is a second set
+///   of them.
+/// * **Cost.** A prefix step can be a whole coding session. Paying for it
+///   again to reach the same failed node buys nothing.
+///
+/// **The graph must be the one that failed.** Node handlers are rebuilt from
+/// `workflow`, and the committed state is keyed by node id, so a `workflow`
+/// whose prefix differs from the one that ran will re-enter the tail on state
+/// it would never have produced — a run that goes green and is quietly wrong.
+/// Editing a *later* node is the supported case and the useful one: fix the
+/// step that failed, continue, keep the prefix. A caller that changed anything
+/// at or upstream of `failed_node` must re-run from the trigger instead, and
+/// [`failure_boundary`] names that node so the check is possible.
+///
+/// # Errors
+/// [`EngineError::Capability`] when the thread has no checkpoint, or the
+/// checkpoint schedules nothing to run — a completed run has no tail, and
+/// asking it to continue is a caller mistake worth naming rather than a
+/// silently empty outcome. Otherwise as [`run`].
+pub async fn retry_with_checkpointer(
+    workflow: &CompiledWorkflow,
+    capabilities: &Capabilities,
+    checkpointer: Arc<dyn Checkpointer<Value>>,
+    thread_id: &str,
+) -> Result<RunOutcome> {
+    let observer = Arc::new(crate::observability::NoopObserver) as Arc<dyn RunObserver>;
+    let (outcome, _run_ids) = resume_with_checkpointer_inner(
+        workflow,
+        capabilities,
+        checkpointer,
+        thread_id,
+        Continuation::Retry,
+        None,
+        &observer,
+    )
+    .await?;
+    Ok(outcome)
+}
+
+/// Like [`retry_with_checkpointer`], but journaled and observed — the shape a
+/// host that records runs actually needs.
+///
+/// The journaled counterpart of
+/// [`resume_with_checkpointer_journaled_observed`], and for the same reason: a
+/// host whose run records are built from observed steps must see the continued
+/// leg the same way it saw the first one, or the record it writes claims the
+/// tail never ran.
+///
+/// # Errors
+/// Same as [`retry_with_checkpointer`].
+pub async fn retry_with_checkpointer_journaled_observed(
+    workflow: &CompiledWorkflow,
+    capabilities: &Capabilities,
+    checkpointer: Arc<dyn Checkpointer<Value>>,
+    thread_id: &str,
+    journal: Arc<dyn GraphEventJournal>,
+    observer: &Arc<dyn RunObserver>,
+) -> Result<JournaledRunOutcome> {
+    let (outcome, graph_run_ids) = resume_with_checkpointer_inner(
+        workflow,
+        capabilities,
+        checkpointer,
+        thread_id,
+        Continuation::Retry,
+        Some(journal),
+        observer,
+    )
+    .await?;
+    Ok(JournaledRunOutcome {
+        outcome,
+        graph_run_ids,
+    })
+}
+
+/// Like [`retry_with_checkpointer`], but reports live progress to `observer`.
+///
+/// The observer sees `on_step_finish` for every node that runs *after* the
+/// failure boundary — which is the point: a host watching a continued run
+/// should see the work that is actually happening, not a replay of the prefix
+/// that is not.
+///
+/// # Errors
+/// Same as [`retry_with_checkpointer`].
+pub async fn retry_with_checkpointer_observed(
+    workflow: &CompiledWorkflow,
+    capabilities: &Capabilities,
+    checkpointer: Arc<dyn Checkpointer<Value>>,
+    thread_id: &str,
+    observer: &Arc<dyn RunObserver>,
+) -> Result<RunOutcome> {
+    let (outcome, _run_ids) = resume_with_checkpointer_inner(
+        workflow,
+        capabilities,
+        checkpointer,
+        thread_id,
+        Continuation::Retry,
+        None,
+        observer,
+    )
+    .await?;
+    Ok(outcome)
+}