From 066c44190f4aa52b7a44daf170f8cb3bd4a4de95 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 19 Aug 2026 11:01:00 +0530 Subject: [PATCH 1/2] feat(engine): continue a failed run from the node that failed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The engine could answer a pause. It could not answer a break. `resume_with_checkpointer` speaks approvals — its resume value is a list of gate decisions — so the only boundary a caller could continue from was a human interrupt. A run whose node errored returned `Err` and that was the whole story: the caller's only move was to run the workflow again from the trigger. The graph runtime has not needed that to be true for a while. When a handler fails past its retry budget, the executor folds the branches that already completed into committed state and writes a **failure-boundary checkpoint** whose pending nodes are the failed one and the not-yet-run tail of its step, with the failed node and error stamped into its metadata. `CompiledGraph::retry` re-runs exactly that. Nothing at the workflow level reached it. This is the two calls that do. `failure_boundary(checkpointer, thread_id)` answers the question the error cannot: is there something to continue, and where did it stop? A separate read rather than a wider error type — every caller already handles `Err`, and widening it would make all of them carry a concept most do not use. It also reads the way the decision is actually made: the run failed; is it worth continuing? `retry_with_checkpointer` continues it, carrying no resume value. There is nothing to decide, only work to redo — and a value would be delivered to `NodeContext::resume` and read as an approval by any gate that happened to be in the pending set. Both paths now share one implementation. The difference between answering a pause and answering a break is a `Continuation`, so the failure path reuses the whole of the approval path — the graph rebuild, the checkpointer re-attach, the fold — rather than growing a parallel copy of it that drifts. **Why this matters more for side effects than for cost.** A prefix that posted a comment, opened a pull request or charged something does not do it twice. Re-running from the trigger was never a neutral choice for a graph with effects in it; it was a second set of them. The saved compute is real too — a prefix step can be a whole coding session — but the correctness argument is the one that justifies the API. **The graph must be the one that failed.** Node handlers are rebuilt from `workflow` and committed state is keyed by node id, so a prefix that differs from the one that ran will re-enter the tail on state it would never have produced — green, and quietly wrong. Editing a *later* node is the supported case and the useful one. `failure_boundary` names the failed node so a caller can check before it continues. Tests count **invocations**, not state. A state diff cannot tell a prefix that was skipped from a prefix that ran again and produced the same thing, and for an effectful prefix those are opposite outcomes — `fuzz_resume.rs` says exactly this in its own module doc and leaves the counter "with the work that fixes it". This is that work. The central test was checked by falsification: swapped for a plain re-run it reports `left: 2, right: 1` on the effectful node, so it is measuring what it claims to. --- src/engine/resumable.rs | 270 ++++++++++++++++++++++++---- tests/resume_after_failure_e2e.rs | 283 ++++++++++++++++++++++++++++++ 2 files changed, 519 insertions(+), 34 deletions(-) create mode 100644 tests/resume_after_failure_e2e.rs diff --git a/src/engine/resumable.rs b/src/engine/resumable.rs index 1c06386..1f9177f 100644 --- a/src/engine/resumable.rs +++ b/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, + /// Gates the operator refused. + rejected: Vec, + }, + /// 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 { + 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>, thread_id: &str, - newly_approved: Vec, - rejected: Vec, + continuation: Continuation, journal: Option>, observer: &Arc, ) -> 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,171 @@ 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, +} + +/// 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>, + thread_id: &str, +) -> Result> { + 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>, + thread_id: &str, +) -> Result { + let observer = Arc::new(crate::observability::NoopObserver) as Arc; + 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 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>, + thread_id: &str, + observer: &Arc, +) -> Result { + let (outcome, _run_ids) = resume_with_checkpointer_inner( + workflow, + capabilities, + checkpointer, + thread_id, + Continuation::Retry, + None, + observer, + ) + .await?; + Ok(outcome) +} diff --git a/tests/resume_after_failure_e2e.rs b/tests/resume_after_failure_e2e.rs new file mode 100644 index 0000000..7f3f07a --- /dev/null +++ b/tests/resume_after_failure_e2e.rs @@ -0,0 +1,283 @@ +#![cfg(feature = "mock")] +//! Continuing a run that *failed*, rather than one that paused. +//! +//! The engine has always been able to answer a pause with a decision. This is +//! the other boundary: a node broke, the runtime committed everything before +//! it, and the question is whether the caller can pick the run back up instead +//! of starting it over. +//! +//! Every test here counts **invocations**, not state. A state diff cannot tell +//! a prefix that was skipped from a prefix that ran again and produced the same +//! thing — and for a prefix that posts a comment or opens a pull request, those +//! are opposite outcomes. `fuzz_resume.rs` says exactly this in its own module +//! doc and leaves the counter "with the work that fixes it"; this is that work. + +mod support; + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use serde_json::{Value, json}; +use tinyflows::caps::mock::mock_capabilities; +use tinyflows::caps::{Capabilities, ToolInvoker}; +use tinyflows::compiler::compile; +use tinyflows::engine::{ + InMemoryCheckpointer, RunInput, failure_boundary, retry_with_checkpointer, + run_with_checkpointer, +}; +use tinyflows::error::{EngineError, Result as EngineResult}; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; + +/// A tool host that counts every call and can be told to break one slug. +/// +/// The counter is the whole point: it is the only thing that distinguishes +/// "the prefix was reused" from "the prefix ran again and looked the same". +#[derive(Default)] +struct CountingTools { + calls: Mutex>, + broken: Mutex>, +} + +impl CountingTools { + fn with_broken(slug: &str) -> Arc { + let tools = Arc::new(Self::default()); + *tools.broken.lock().expect("lock") = Some(slug.to_string()); + tools + } + + /// How many times `slug` was invoked across every run so far. + fn calls(&self, slug: &str) -> usize { + self.calls + .lock() + .expect("lock") + .get(slug) + .copied() + .unwrap_or(0) + } + + /// Stop breaking whatever was broken — the "someone fixed it" step. + fn repair(&self) { + *self.broken.lock().expect("lock") = None; + } +} + +#[async_trait] +impl ToolInvoker for CountingTools { + async fn invoke(&self, slug: &str, _args: Value, _conn: Option<&str>) -> EngineResult { + *self + .calls + .lock() + .expect("lock") + .entry(slug.to_string()) + .or_insert(0) += 1; + if self.broken.lock().expect("lock").as_deref() == Some(slug) { + return Err(EngineError::Capability(format!("{slug} is down"))); + } + Ok(json!({ "slug": slug, "ok": true })) + } +} + +fn caps(tools: Arc) -> Capabilities { + Capabilities { + tools, + ..mock_capabilities() + } +} + +/// `start → post_comment → tally → summarise`, all tool calls. +/// +/// Named for effects on purpose. The middle step is the one broken in these +/// tests, so `post_comment` is the step a re-run from the trigger would +/// perform twice — which is the cost this feature exists to avoid. +fn graph() -> WorkflowGraph { + let tool = |id: &str| Node { + id: id.to_string(), + kind: NodeKind::ToolCall, + type_version: 1, + name: id.to_string(), + config: json!({ "slug": id, "args": {} }), + ports: Vec::new(), + position: None, + }; + let edge = |from: &str, to: &str| Edge { + from_node: from.to_string(), + from_port: "main".to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + }; + WorkflowGraph { + schema_version: 1, + id: None, + name: "review and report".to_string(), + inputs: Vec::new(), + agents: Vec::new(), + nodes: vec![ + Node { + id: "start".to_string(), + kind: NodeKind::Trigger, + type_version: 1, + name: "start".to_string(), + config: json!({ "trigger_kind": "manual" }), + ports: Vec::new(), + position: None, + }, + tool("post_comment"), + tool("tally"), + tool("summarise"), + ], + edges: vec![ + edge("start", "post_comment"), + edge("post_comment", "tally"), + edge("tally", "summarise"), + ], + } +} + +#[tokio::test] +async fn a_continued_run_finishes_the_tail_without_repeating_the_prefix() { + // The property the whole feature rests on, and the one a state diff cannot + // see: the effectful prefix happens exactly once across a failure and the + // continue that follows it. + let tools = CountingTools::with_broken("tally"); + let compiled = compile(&graph()).expect("compiles"); + let checkpointer = Arc::new(InMemoryCheckpointer::new()); + let thread = "run-1"; + + let failed = run_with_checkpointer( + &compiled, + RunInput::new(json!({})), + &caps(tools.clone()), + checkpointer.clone(), + thread, + ) + .await; + assert!(failed.is_err(), "the broken tool must fail the run"); + assert_eq!(tools.calls("post_comment"), 1, "the prefix ran once"); + assert_eq!(tools.calls("summarise"), 0, "the tail did not run at all"); + + let boundary = failure_boundary(&(checkpointer.clone() as _), thread) + .await + .expect("readable") + .expect("a failed run leaves a boundary to continue from"); + assert_eq!(boundary.failed_node, "tally"); + assert!(boundary.error.contains("tally is down"), "{boundary:?}"); + assert_eq!( + boundary.pending, + vec!["tally".to_string()], + "the failed node is what a continue would run" + ); + + // Someone fixes the thing that was broken, and the run picks up where it + // stopped. + tools.repair(); + let outcome = retry_with_checkpointer(&compiled, &caps(tools.clone()), checkpointer, thread) + .await + .expect("the continued run completes"); + + assert_eq!( + tools.calls("post_comment"), + 1, + "THE POINT: the effectful prefix was not performed a second time" + ); + assert_eq!( + tools.calls("tally"), + 2, + "the failed node ran again — once broken, once fixed" + ); + assert_eq!(tools.calls("summarise"), 1, "and the tail finally ran"); + assert!( + outcome.pending_approvals.is_empty(), + "nothing was waiting on a human" + ); + // The prefix's output survived in committed state rather than being + // recomputed — the run's final state carries all three steps. + for node in ["post_comment", "tally", "summarise"] { + assert!( + outcome.output["nodes"].get(node).is_some(), + "{node} missing from the continued run's state: {}", + outcome.output["nodes"] + ); + } +} + +#[tokio::test] +async fn a_run_that_is_still_broken_fails_again_and_leaves_the_boundary_standing() { + // A continue is not a fix. Retrying without repairing must not consume the + // boundary — an operator who tries twice before finding the cause has to + // still be able to continue afterwards. + let tools = CountingTools::with_broken("tally"); + let compiled = compile(&graph()).expect("compiles"); + let checkpointer = Arc::new(InMemoryCheckpointer::new()); + let thread = "run-2"; + + let _ = run_with_checkpointer( + &compiled, + RunInput::new(json!({})), + &caps(tools.clone()), + checkpointer.clone(), + thread, + ) + .await; + + let again = retry_with_checkpointer( + &compiled, + &caps(tools.clone()), + checkpointer.clone(), + thread, + ) + .await; + assert!(again.is_err(), "still broken, so still failing"); + assert_eq!( + tools.calls("post_comment"), + 1, + "and still without repeating the prefix" + ); + + let boundary = failure_boundary(&(checkpointer as _), thread) + .await + .expect("readable") + .expect("the boundary survives a failed continue"); + assert_eq!(boundary.failed_node, "tally"); +} + +#[tokio::test] +async fn a_completed_run_reports_no_failure_boundary() { + // The negative half. `failure_boundary` is read after an `Err`, but a + // caller that reads it anywhere else must not be told a healthy thread has + // something to continue. + let tools = Arc::new(CountingTools::default()); + let compiled = compile(&graph()).expect("compiles"); + let checkpointer = Arc::new(InMemoryCheckpointer::new()); + let thread = "run-3"; + + run_with_checkpointer( + &compiled, + RunInput::new(json!({})), + &caps(tools.clone()), + checkpointer.clone(), + thread, + ) + .await + .expect("nothing is broken"); + + assert!( + failure_boundary(&(checkpointer as _), thread) + .await + .expect("readable") + .is_none(), + "a run that finished has no failure to continue from" + ); +} + +#[tokio::test] +async fn a_thread_nothing_ever_ran_on_has_no_boundary() { + let checkpointer: Arc> = + Arc::new(InMemoryCheckpointer::new()); + assert!( + failure_boundary(&checkpointer, "never-ran") + .await + .expect("readable") + .is_none() + ); +} From 4551998a8ae9641b006438db092a3ba5cdc8379a Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 19 Aug 2026 11:24:02 +0530 Subject: [PATCH 2/2] feat(engine): journaled + observed variant of the failure continue The shape a host that records runs actually needs, and the one the first consumer reached for immediately. `resume` has had it since approvals existed; the failure path shipped without it, so a host whose run records are built from observed steps could continue a run and then write a record claiming the tail never ran. --- src/engine/resumable.rs | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/src/engine/resumable.rs b/src/engine/resumable.rs index 1f9177f..8417cf7 100644 --- a/src/engine/resumable.rs +++ b/src/engine/resumable.rs @@ -690,6 +690,41 @@ pub async fn retry_with_checkpointer( 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>, + thread_id: &str, + journal: Arc, + observer: &Arc, +) -> Result { + 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