feat(orchestration): node-level checkpoint & resume for delegations (#3884) (#5178)

Co-authored-by: M3gA-Mind <elvin@tinyhumans.ai>

src/openhuman/agent_orchestration/delegation.rs

@@ -23,7 +23,8 @@ use crate::openhuman::agent::harness::subagent_runner::{run_subagent, SubagentRu
 use crate::openhuman::agent_orchestration::parent_context::build_root_parent;
 use crate::openhuman::config::Config;
 use crate::openhuman::tinyagents::delegation::{
-    run_delegation, DelegationConfig, DelegationStage, DelegationStageOutput, DelegationState,
+    run_or_resume_delegation, DelegationConfig, DelegationStage, DelegationStageOutput,
+    DelegationState,
 };
 use tinyagents::graph::checkpoint::Checkpointer;
 use tinyagents::graph::SqliteCheckpointer;
@@ -107,6 +108,10 @@ pub(crate) async fn run_subagent_delegation(
                         Ok(DelegationStageOutput {
                             text: outcome.output,
                             approved,
+                            // Persist the exact prompt this stage sent so the
+                            // execute step's `StepRecord` carries its provenance
+                            // (read only for the execute stage; #3884).
+                            prompt: Some(prompt),
                         })
                     }
                     Err(e) => Err(format!("delegation stage {stage:?} failed: {e}")),
@@ -125,7 +130,12 @@ pub(crate) async fn run_subagent_delegation(
             // stays off here until a human-review delegation surface wires it.
             ..DelegationConfig::default()
         };
-        run_delegation(delegation_config, run_stage).await
+        // Resume-aware entry (#3884): with today's fresh-per-run `thread_id` this
+        // is always a fresh run; reusing a stable `thread_id` resumes from the
+        // last checkpoint boundary instead of restarting.
+        run_or_resume_delegation(delegation_config, run_stage)
+            .await
+            .map(|outcome| outcome.state)
     };
 
     if current_parent().is_some() {
@@ -160,11 +170,7 @@ fn build_stage_prompt(stage: DelegationStage, task: &str, state: &DelegationStat
             )
         }
         DelegationStage::Review => {
-            let result = state
-                .executions
-                .last()
-                .map(String::as_str)
-                .unwrap_or("(no execution produced)");
+            let result = state.last_result().unwrap_or("(no execution produced)");
             format!(
                 "Review the result below against the task. If it fully and correctly \
                  accomplishes the task, reply with `APPROVE` on the first line. Otherwise \

src/openhuman/tinyagents/delegation.rs

@@ -32,7 +32,7 @@ use std::sync::Arc;
 
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
-use tinyagents::graph::checkpoint::Checkpointer;
+use tinyagents::graph::checkpoint::{Checkpoint, Checkpointer};
 use tinyagents::graph::export::GraphTopology;
 use tinyagents::graph::recursion::RecursionPolicy;
 use tinyagents::graph::ClosureStateReducer;
@@ -61,26 +61,58 @@ pub(crate) struct DelegationStageOutput {
     /// Only meaningful for [`DelegationStage::Review`]: `true` approves the
     /// execution and ends the loop; `false` requests another revision.
     pub(crate) approved: bool,
+    /// The exact prompt handed to this stage's worker, when it surfaces one.
+    /// Persisted into [`StepRecord::prompt`] for per-step provenance (read only
+    /// for the execute stage; ignored elsewhere). `None` when the worker does not
+    /// surface a prompt — e.g. the deterministic test mock.
+    pub(crate) prompt: Option<String>,
 }
 
 impl DelegationStageOutput {
-    /// A plain non-review stage output (the `approved` flag is unused).
+    /// A plain non-review stage output (the `approved` flag is unused and no
+    /// prompt is surfaced).
     pub(crate) fn done(text: impl Into<String>) -> Self {
         Self {
             text: text.into(),
             approved: true,
+            prompt: None,
         }
     }
 }
 
+/// Current on-disk schema version for a checkpointed [`DelegationState`]. Bumped
+/// only on a breaking state-shape change — introduced with the
+/// `executions: Vec<String>` → `Vec<StepRecord>` migration (issue #3884).
+/// Pre-versioned records deserialize to `0` via `#[serde(default)]`, so a resume
+/// can tell a stale checkpoint from a current one.
+pub(crate) const CURRENT_SCHEMA_VERSION: u32 = 1;
+
+/// One completed execute-stage pass, recorded durably so a resumed run knows
+/// exactly how far it got and can render/finalize per step rather than from a
+/// flat text log. Replaces the former `executions: Vec<String>` (issue #3884).
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub(crate) struct StepRecord {
+    /// 0-based execute pass: `0` is the first execution, `n` the n-th revision.
+    pub(crate) index: usize,
+    /// The exact prompt handed to the execute sub-agent for this pass — per-step
+    /// provenance and the seam a later plan-edit slice (#3881) diffs against.
+    /// Empty when the worker did not surface one.
+    #[serde(default)]
+    pub(crate) prompt: String,
+    /// The sub-agent's result text — the value the former `Vec<String>` entry held.
+    pub(crate) result: String,
+}
+
 /// Typed working state threaded through (and checkpointed across) the delegation
 /// graph. Serde-serializable so a [`Checkpointer`] can persist and restore it.
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub(crate) struct DelegationState {
     /// The plan produced by the `plan` stage.
     pub(crate) plan: Option<String>,
-    /// One entry per execution pass (the first plus each revision).
-    pub(crate) executions: Vec<String>,
+    /// One record per execution pass (the first plus each revision), typed so a
+    /// resumed run can render/finalize per step (widened from `Vec<String>`,
+    /// issue #3884).
+    pub(crate) executions: Vec<StepRecord>,
     /// One entry per review pass.
     pub(crate) reviews: Vec<String>,
     /// Number of revisions the reviewer requested (loops back to `execute`).
@@ -100,12 +132,47 @@ pub(crate) struct DelegationState {
     /// (deny semantics: block the action, finalize as denied).
     #[serde(default)]
     pub(crate) denied: bool,
+    /// On-disk schema version, stamped [`CURRENT_SCHEMA_VERSION`] on a fresh run
+    /// and defaulting to `0` for pre-versioned checkpoints.
+    /// [`run_or_resume_delegation`] expires any checkpoint whose version is below
+    /// `CURRENT_SCHEMA_VERSION` (and any that fails to deserialize) instead of
+    /// resuming or returning it — so a shape change that stays structurally
+    /// decodable is still not misread.
+    #[serde(default)]
+    pub(crate) schema_version: u32,
+}
+
+impl DelegationState {
+    /// A fresh run's initial state, stamped with the current schema version so
+    /// its checkpoints are self-identifying.
+    fn new_run() -> Self {
+        Self {
+            schema_version: CURRENT_SCHEMA_VERSION,
+            ..Self::default()
+        }
+    }
+
+    /// The latest execution result text, if any — the projection the review
+    /// prompt and the finalize summary read.
+    pub(crate) fn last_result(&self) -> Option<&str> {
+        self.executions.last().map(|r| r.result.as_str())
+    }
+
+    /// The execution result texts in order — the flat projection used for the
+    /// durable approval-interrupt payload, kept `Vec<&str>` so that wire shape
+    /// is unchanged from the pre-#3884 `Vec<String>`.
+    pub(crate) fn executions_texts(&self) -> Vec<&str> {
+        self.executions.iter().map(|r| r.result.as_str()).collect()
+    }
 }
 
 /// Reducer updates emitted by the delegation nodes.
 enum DelegationUpdate {
     Plan(String),
-    Execution(String),
+    Execution {
+        prompt: String,
+        result: String,
+    },
     Review {
         note: String,
         approved: bool,
@@ -243,9 +310,10 @@ where
         "[delegation] running sub-agent delegation graph"
     );
 
+    let initial = DelegationState::new_run();
     let execution = match thread_id.clone() {
-        Some(tid) => graph.run_with_thread(tid, DelegationState::default()).await,
-        None => graph.run(DelegationState::default()).await,
+        Some(tid) => graph.run_with_thread(tid, initial).await,
+        None => graph.run(initial).await,
     }
     .map_err(|e| format!("delegation graph run failed: {e}"))?;
 
@@ -271,6 +339,183 @@ pub(crate) async fn resume_delegation<F, Fut>(
     decision: Value,
     run_stage: F,
 ) -> Result<DelegationOutcome, String>
+where
+    F: Fn(DelegationStage, DelegationState) -> Fut + Clone + Send + Sync + 'static,
+    Fut: Future<Output = Result<DelegationStageOutput, String>> + Send + 'static,
+{
+    let approved = decision_is_approve(&decision);
+    tracing::info!(
+        approved,
+        "[interrupt] resuming durable delegation graph with approval decision"
+    );
+    let mut command = Command::default();
+    command.resume = Some(decision);
+    resume_graph(config, command, run_stage).await
+}
+
+/// Run the delegation graph, resuming from the last checkpoint boundary when the
+/// configured thread has a live, compatible, non-terminal checkpoint, else
+/// starting fresh (issue #3884 — node-level checkpoint & resume).
+///
+/// Classifies the thread's latest checkpoint and routes accordingly:
+/// - **resumable** (a crash/failure left a mid-run boundary) → re-run only the
+///   not-yet-completed nodes from that boundary via [`CompiledGraph::resume`]
+///   with an empty command — never restarting from `plan`, and never re-running
+///   an already-completed step (its [`StepRecord`] is restored from the state);
+/// - **terminal** (already finalized/cancelled) → return the stored final state
+///   without re-running (idempotent re-invocation of a stable thread);
+/// - **absent** (no checkpoint) → a fresh durable run;
+/// - **incompatible** (an undecodable record — e.g. a pre-#3884 `Vec<String>`
+///   `executions` shape) → log, best-effort prune, and start fresh. The decode
+///   error is swallowed into a fresh-run decision, never propagated, never a panic.
+///
+/// Callers that mint a unique `thread_id` per run (today's default) always take
+/// the fresh path unchanged; the resume paths activate only when a caller reuses
+/// a stable `thread_id`, so this is byte-compatible for existing callers while
+/// wiring the resume seam that #3881 (plan-edit resume) builds on.
+pub(crate) async fn run_or_resume_delegation<F, Fut>(
+    config: DelegationConfig,
+    run_stage: F,
+) -> Result<DelegationOutcome, String>
+where
+    F: Fn(DelegationStage, DelegationState) -> Fut + Clone + Send + Sync + 'static,
+    Fut: Future<Output = Result<DelegationStageOutput, String>> + Send + 'static,
+{
+    // Without a checkpointer + thread there is nothing to resume from.
+    let (Some(cp), Some(tid)) = (config.checkpointer.clone(), config.thread_id.clone()) else {
+        return run_delegation_durable(config, run_stage).await;
+    };
+
+    match cp.get(tid.as_str(), None).await {
+        // A checkpoint written under an older state schema (e.g. a pre-#3884
+        // record whose `executions` happened to be empty and so still decoded
+        // into `Vec<StepRecord>`) is expired rather than resumed/returned — its
+        // semantics may not match the current graph. This is what makes
+        // `schema_version` an actual guard, not just documentation, and closes
+        // the empty-`executions` gap that a decode failure alone cannot catch.
+        Ok(Some(checkpoint)) if checkpoint.state.schema_version < CURRENT_SCHEMA_VERSION => {
+            tracing::warn!(
+                thread_id = %tid,
+                schema_version = checkpoint.state.schema_version,
+                current = CURRENT_SCHEMA_VERSION,
+                "[delegation] checkpoint predates the current state schema; pruning and starting fresh"
+            );
+            prune_thread(cp.as_ref(), &tid).await;
+            run_delegation_durable(config, run_stage).await
+        }
+        Ok(Some(checkpoint)) if checkpoint_is_resumable(&checkpoint) => {
+            tracing::info!(
+                thread_id = %tid,
+                "[delegation] resuming durable delegation from its last checkpoint boundary"
+            );
+            // A crash/failure resume carries no decision value — an empty command
+            // simply re-runs the pending node(s) (the crate's `retry` semantics).
+            resume_graph(config, Command::default(), run_stage).await
+        }
+        Ok(Some(checkpoint)) => {
+            // Terminal: return the finalized state without re-running. Defensive:
+            // if this checkpoint still carried an unconsumed interrupt, surface it
+            // instead of silently dropping it. The current routing never produces
+            // this (an interrupt boundary schedules its node and is classified
+            // resumable above), but a future schedule could, and a dropped pause
+            // would strand a run.
+            let pending = checkpoint.interrupts.first().map(|i| PendingApproval {
+                interrupt_id: i.id.clone(),
+                node: i.node.as_str().to_string(),
+                payload: i.payload.clone(),
+                thread_id: tid.clone(),
+            });
+            if pending.is_some() {
+                tracing::warn!(
+                    thread_id = %tid,
+                    "[delegation] terminal-classified checkpoint carried a pending interrupt; surfacing it"
+                );
+            } else {
+                tracing::info!(
+                    thread_id = %tid,
+                    "[delegation] thread already terminal; returning finalized state without re-running"
+                );
+            }
+            Ok(DelegationOutcome {
+                state: checkpoint.state,
+                pending,
+            })
+        }
+        Ok(None) => {
+            tracing::debug!(
+                thread_id = %tid,
+                "[delegation] no checkpoint for thread; starting a fresh durable run"
+            );
+            run_delegation_durable(config, run_stage).await
+        }
+        // Only a *decode / shape-incompatibility* read error expires the
+        // checkpoint. An operational error (SQLite busy / I/O / poisoned lock)
+        // must NOT silently restart a valid resumable run — it is propagated so
+        // durable work is retried by the caller, not dropped.
+        Err(e) if is_incompatible_checkpoint_error(&e) => {
+            tracing::warn!(
+                thread_id = %tid,
+                error = %e,
+                "[delegation] undecodable/incompatible checkpoint; pruning and starting fresh"
+            );
+            prune_thread(cp.as_ref(), &tid).await;
+            run_delegation_durable(config, run_stage).await
+        }
+        Err(e) => {
+            tracing::error!(
+                thread_id = %tid,
+                error = %e,
+                "[delegation] checkpoint read failed (operational); not restarting — propagating error"
+            );
+            Err(format!(
+                "delegation checkpoint read failed for thread {tid}: {e}"
+            ))
+        }
+    }
+}
+
+/// Best-effort prune of a dead/expired checkpoint thread so it is not re-probed
+/// forever. Failure to prune is non-fatal (logged at debug).
+async fn prune_thread(cp: &dyn Checkpointer<DelegationState>, thread_id: &str) {
+    if let Err(e) = cp.delete_thread(thread_id).await {
+        tracing::debug!(
+            thread_id = %thread_id,
+            error = %e,
+            "[delegation] could not prune checkpoint thread (non-fatal)"
+        );
+    }
+}
+
+/// Whether a `Checkpointer::get` error is a decode / shape-incompatibility (safe
+/// to expire the checkpoint) rather than an operational failure (SQLite busy /
+/// I/O / poisoned lock — must not silently restart durable work). The vendored
+/// `SqliteCheckpointer` reports both as `TinyAgentsError::Checkpoint(String)` but
+/// tags decode failures with a `"decode …"` context (`sqlite.rs`:
+/// `decode record` / `decode namespace` / `decode next_nodes`) — the only stable
+/// discriminator it exposes.
+fn is_incompatible_checkpoint_error(e: &tinyagents::TinyAgentsError) -> bool {
+    matches!(e, tinyagents::TinyAgentsError::Checkpoint(msg) if msg.contains("decode"))
+}
+
+/// Whether a loaded checkpoint still has work to resume: a non-finalized,
+/// non-cancelled run that still schedules a real (non-`END`) node.
+fn checkpoint_is_resumable(checkpoint: &Checkpoint<DelegationState>) -> bool {
+    if checkpoint.state.final_output.is_some() || checkpoint.state.cancelled {
+        return false;
+    }
+    checkpoint.next_nodes.iter().any(|n| n.as_str() != END)
+}
+
+/// Rebuild the delegation graph (its node closures are not serializable — only
+/// the typed state is checkpointed) with the same checkpointer + `thread_id`, and
+/// re-enter it at the latest checkpoint's pending node(s) via
+/// [`CompiledGraph::resume`]. `command` carries an approver's decision for the
+/// human-approval interrupt path, or is empty for a plain crash/failure resume.
+async fn resume_graph<F, Fut>(
+    config: DelegationConfig,
+    command: Command<DelegationUpdate>,
+    run_stage: F,
+) -> Result<DelegationOutcome, String>
 where
     F: Fn(DelegationStage, DelegationState) -> Fut + Clone + Send + Sync + 'static,
     Fut: Future<Output = Result<DelegationStageOutput, String>> + Send + 'static,
@@ -295,16 +540,6 @@ where
     )))
     .with_checkpointer(cp);
 
-    let approved = decision_is_approve(&decision);
-    tracing::info!(
-        thread_id = %thread_id,
-        approved,
-        "[interrupt] resuming durable delegation graph with approval decision"
-    );
-
-    let mut command = Command::default();
-    command.resume = Some(decision);
-
     let execution = graph
         .resume(thread_id.clone(), command)
         .await
@@ -385,7 +620,14 @@ where
         ClosureStateReducer::new(|mut s: DelegationState, u: DelegationUpdate| {
             match u {
                 DelegationUpdate::Plan(p) => s.plan = Some(p),
-                DelegationUpdate::Execution(e) => s.executions.push(e),
+                DelegationUpdate::Execution { prompt, result } => {
+                    let index = s.executions.len();
+                    s.executions.push(StepRecord {
+                        index,
+                        prompt,
+                        result,
+                    });
+                }
                 DelegationUpdate::Review { note, approved } => {
                     s.reviews.push(note);
                     s.approved = approved;
@@ -454,7 +696,10 @@ where
                 .map_err(to_node_err)?;
             Ok(NodeResult::Command(
                 Command::default()
-                    .with_update(DelegationUpdate::Execution(out.text))
+                    .with_update(DelegationUpdate::Execution {
+                        prompt: out.prompt.unwrap_or_default(),
+                        result: out.text,
+                    })
                     .with_goto(["review"]),
             ))
         }
@@ -527,7 +772,7 @@ where
                         let payload = json!({
                             "kind": "delegation_review",
                             "reviews": s.reviews,
-                            "executions": s.executions,
+                            "executions": s.executions_texts(),
                             "revisions": s.revisions,
                         });
                         tracing::info!(
@@ -564,7 +809,7 @@ where
             let summary = s
                 .executions
                 .last()
-                .cloned()
+                .map(|r| r.result.clone())
                 .unwrap_or_else(|| "<no execution>".to_string());
             let final_text = if s.cancelled {
                 format!("cancelled after {} execution(s)", s.executions.len())
@@ -672,6 +917,7 @@ mod tests {
                         Ok(DelegationStageOutput {
                             text: format!("review-{n}"),
                             approved: n >= reject_first,
+                            prompt: None,
                         })
                     }
                 }
@@ -686,6 +932,8 @@ mod tests {
             .expect("runs");
         assert_eq!(state.plan.as_deref(), Some("PLAN"));
         assert_eq!(state.executions.len(), 1, "one execution, no revision");
+        assert_eq!(state.executions[0].index, 0);
+        assert_eq!(state.executions[0].result, "EXEC");
         assert_eq!(state.reviews.len(), 1);
         assert_eq!(state.revisions, 0);
         assert!(state.approved);
@@ -848,4 +1096,429 @@ mod tests {
             "at least one super-step boundary checkpoint persisted"
         );
     }
+
+    /// The boxed-future type every inline test runner returns.
+    type BoxedStageFut =
+        std::pin::Pin<Box<dyn Future<Output = Result<DelegationStageOutput, String>> + Send>>;
+
+    #[tokio::test]
+    async fn execution_records_capture_index_and_prompt() {
+        // A worker that surfaces an execute prompt; assert the per-step record
+        // carries index + prompt + result (issue #3884).
+        let runner = move |stage: DelegationStage, _s: DelegationState| {
+            Box::pin(async move {
+                match stage {
+                    DelegationStage::Plan => Ok(DelegationStageOutput::done("PLAN")),
+                    DelegationStage::Execute => Ok(DelegationStageOutput {
+                        text: "EXEC".to_string(),
+                        approved: true,
+                        prompt: Some("EXEC-PROMPT".to_string()),
+                    }),
+                    DelegationStage::Review => Ok(DelegationStageOutput {
+                        text: "APPROVE".to_string(),
+                        approved: true,
+                        prompt: None,
+                    }),
+                }
+            }) as BoxedStageFut
+        };
+        let state = run_delegation(DelegationConfig::default(), runner)
+            .await
+            .expect("runs");
+        assert_eq!(state.executions.len(), 1);
+        assert_eq!(state.executions[0].index, 0);
+        assert_eq!(state.executions[0].result, "EXEC");
+        assert_eq!(state.executions[0].prompt, "EXEC-PROMPT");
+    }
+
+    #[test]
+    fn legacy_string_executions_do_not_deserialize_into_step_records() {
+        // A pre-#3884 checkpoint stored `executions` as a flat string array. It
+        // must fail to load into the new `Vec<StepRecord>` — which is exactly
+        // what makes `run_or_resume_delegation` expire the stale checkpoint
+        // instead of misreading it.
+        let legacy = r#"{"plan":"P","executions":["raw a","raw b"],"reviews":[],"revisions":0,"approved":true,"final_output":null,"cancelled":false}"#;
+        assert!(serde_json::from_str::<DelegationState>(legacy).is_err());
+    }
+
+    #[test]
+    fn schema_version_defaults_to_zero_and_step_records_round_trip() {
+        // A pre-versioned record (no `schema_version`) loads as version 0.
+        let unversioned = r#"{"plan":null,"executions":[],"reviews":[],"revisions":0,"approved":false,"final_output":null,"cancelled":false}"#;
+        let s: DelegationState = serde_json::from_str(unversioned).expect("loads");
+        assert_eq!(s.schema_version, 0);
+
+        // A fresh run stamps the current version and round-trips step records.
+        let mut fresh = DelegationState::new_run();
+        assert_eq!(fresh.schema_version, CURRENT_SCHEMA_VERSION);
+        fresh.executions.push(StepRecord {
+            index: 0,
+            prompt: "P".to_string(),
+            result: "R".to_string(),
+        });
+        let json = serde_json::to_string(&fresh).expect("serializes");
+        let back: DelegationState = serde_json::from_str(&json).expect("round-trips");
+        assert_eq!(back.schema_version, CURRENT_SCHEMA_VERSION);
+        assert_eq!(back.executions[0].result, "R");
+        assert_eq!(back.executions[0].prompt, "P");
+    }
+
+    #[tokio::test]
+    async fn run_or_resume_starts_fresh_without_a_checkpoint() {
+        let dir = tempfile::tempdir().unwrap();
+        let cp: Arc<dyn Checkpointer<DelegationState>> = Arc::new(
+            tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path()),
+        );
+        let config = DelegationConfig {
+            checkpointer: Some(cp),
+            thread_id: Some("fresh-1".to_string()),
+            ..DelegationConfig::default()
+        };
+        let outcome = run_or_resume_delegation(config, flow_runner(0))
+            .await
+            .expect("runs fresh");
+        assert!(outcome.state.final_output.is_some());
+        assert_eq!(outcome.state.executions.len(), 1);
+        assert_eq!(outcome.state.schema_version, CURRENT_SCHEMA_VERSION);
+    }
+
+    #[tokio::test]
+    async fn run_or_resume_continues_from_last_boundary_after_a_crash() {
+        let dir = tempfile::tempdir().unwrap();
+        let cp: Arc<dyn Checkpointer<DelegationState>> = Arc::new(
+            tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path()),
+        );
+        // Count stage invocations across BOTH the crashed run and the resume, to
+        // prove plan/execute are NOT re-run on resume.
+        let plan_runs = Arc::new(AtomicUsize::new(0));
+        let exec_runs = Arc::new(AtomicUsize::new(0));
+        let review_runs = Arc::new(AtomicUsize::new(0));
+        let make_runner = |crash_first_review: bool| {
+            let plan_runs = plan_runs.clone();
+            let exec_runs = exec_runs.clone();
+            let review_runs = review_runs.clone();
+            move |stage: DelegationStage, _s: DelegationState| {
+                let plan_runs = plan_runs.clone();
+                let exec_runs = exec_runs.clone();
+                let review_runs = review_runs.clone();
+                Box::pin(async move {
+                    match stage {
+                        DelegationStage::Plan => {
+                            plan_runs.fetch_add(1, Ordering::SeqCst);
+                            Ok(DelegationStageOutput::done("PLAN"))
+                        }
+                        DelegationStage::Execute => {
+                            exec_runs.fetch_add(1, Ordering::SeqCst);
+                            Ok(DelegationStageOutput {
+                                text: "EXEC".to_string(),
+                                approved: true,
+                                prompt: Some("EXEC-PROMPT".to_string()),
+                            })
+                        }
+                        DelegationStage::Review => {
+                            let n = review_runs.fetch_add(1, Ordering::SeqCst);
+                            if crash_first_review && n == 0 {
+                                Err("simulated crash during review".to_string())
+                            } else {
+                                Ok(DelegationStageOutput {
+                                    text: "APPROVE".to_string(),
+                                    approved: true,
+                                    prompt: None,
+                                })
+                            }
+                        }
+                    }
+                }) as BoxedStageFut
+            }
+        };
+        let make_config = || DelegationConfig {
+            checkpointer: Some(cp.clone()),
+            thread_id: Some("resume-1".to_string()),
+            ..DelegationConfig::default()
+        };
+
+        // Run 1: crashes during the first review. plan + execute completed and
+        // were checkpointed at their super-step boundaries; the run returns Err.
+        let crashed = run_or_resume_delegation(make_config(), make_runner(true)).await;
+        assert!(crashed.is_err(), "first run crashes at review");
+        assert_eq!(plan_runs.load(Ordering::SeqCst), 1);
+        assert_eq!(exec_runs.load(Ordering::SeqCst), 1);
+
+        // Run 2 (resume): a resumable checkpoint exists → re-enter at the pending
+        // node (review), NOT plan; plan/execute must not run again.
+        let resumed = run_or_resume_delegation(make_config(), make_runner(false))
+            .await
+            .expect("resumes");
+        assert!(
+            resumed.state.final_output.is_some(),
+            "resumed run finalizes"
+        );
+        assert_eq!(
+            plan_runs.load(Ordering::SeqCst),
+            1,
+            "plan not re-run on resume"
+        );
+        assert_eq!(
+            exec_runs.load(Ordering::SeqCst),
+            1,
+            "execute not re-run on resume"
+        );
+        assert_eq!(
+            resumed.state.executions.len(),
+            1,
+            "the pre-crash execution survived; no duplicate"
+        );
+        assert_eq!(resumed.state.executions[0].result, "EXEC");
+    }
+
+    #[tokio::test]
+    async fn run_or_resume_is_idempotent_on_a_finalized_thread() {
+        let dir = tempfile::tempdir().unwrap();
+        let cp: Arc<dyn Checkpointer<DelegationState>> = Arc::new(
+            tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path()),
+        );
+        let stage_runs = Arc::new(AtomicUsize::new(0));
+        let make_runner = || {
+            let stage_runs = stage_runs.clone();
+            move |stage: DelegationStage, _s: DelegationState| {
+                let stage_runs = stage_runs.clone();
+                Box::pin(async move {
+                    stage_runs.fetch_add(1, Ordering::SeqCst);
+                    match stage {
+                        DelegationStage::Review => Ok(DelegationStageOutput {
+                            text: "APPROVE".to_string(),
+                            approved: true,
+                            prompt: None,
+                        }),
+                        _ => Ok(DelegationStageOutput::done("X")),
+                    }
+                }) as BoxedStageFut
+            }
+        };
+        let make_config = || DelegationConfig {
+            checkpointer: Some(cp.clone()),
+            thread_id: Some("done-1".to_string()),
+            ..DelegationConfig::default()
+        };
+
+        let first = run_or_resume_delegation(make_config(), make_runner())
+            .await
+            .expect("runs");
+        assert!(first.state.final_output.is_some());
+        let after_first = stage_runs.load(Ordering::SeqCst);
+        assert!(after_first > 0, "the first run invoked stage workers");
+
+        // Re-invoke the SAME thread: a terminal checkpoint → return the finalized
+        // state without re-running any stage worker.
+        let second = run_or_resume_delegation(make_config(), make_runner())
+            .await
+            .expect("idempotent");
+        assert!(second.state.final_output.is_some());
+        assert_eq!(
+            stage_runs.load(Ordering::SeqCst),
+            after_first,
+            "no stage worker re-ran on a finalized thread"
+        );
+    }
+
+    #[tokio::test]
+    async fn incompatible_checkpoint_expires_to_a_fresh_run() {
+        // A pre-#3884 checkpoint (executions as `Vec<String>`) left in the store
+        // must not crash a resume: `run_or_resume_delegation` expires it and runs
+        // fresh, never panicking or surfacing the decode error.
+        #[derive(Clone, Debug, Serialize, Deserialize)]
+        struct LegacyState {
+            plan: Option<String>,
+            executions: Vec<String>,
+            reviews: Vec<String>,
+            revisions: usize,
+            approved: bool,
+            final_output: Option<String>,
+            cancelled: bool,
+        }
+        let dir = tempfile::tempdir().unwrap();
+        // Seed the store as the OLD state type under the thread.
+        let legacy_cp: tinyagents::graph::checkpoint::FileCheckpointer<LegacyState> =
+            tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path());
+        let legacy = Checkpoint {
+            thread_id: "legacy-1".to_string(),
+            checkpoint_id: "cp-legacy".to_string(),
+            run_id: None,
+            parent_checkpoint_id: None,
+            namespace: vec![],
+            state: LegacyState {
+                plan: Some("old".to_string()),
+                executions: vec!["a".to_string(), "b".to_string()],
+                reviews: vec![],
+                revisions: 0,
+                approved: true,
+                final_output: None,
+                cancelled: false,
+            },
+            next_nodes: vec![],
+            completed_tasks: vec![],
+            pending_writes: vec![],
+            interrupts: vec![],
+            pending_activations: None,
+            barrier_arrivals: vec![],
+            metadata: json!({}),
+        };
+        legacy_cp.put(legacy).await.expect("seed legacy checkpoint");
+
+        // Reopen the SAME store as the current state type and resume: the
+        // undecodable record is expired and a fresh run completes.
+        let cp: Arc<dyn Checkpointer<DelegationState>> =
+            Arc::new(tinyagents::graph::checkpoint::FileCheckpointer::<
+                DelegationState,
+            >::new(dir.path()));
+        let config = DelegationConfig {
+            checkpointer: Some(cp),
+            thread_id: Some("legacy-1".to_string()),
+            ..DelegationConfig::default()
+        };
+        let outcome = run_or_resume_delegation(config, flow_runner(0))
+            .await
+            .expect("expires stale checkpoint and runs fresh");
+        assert!(outcome.state.final_output.is_some(), "fresh run completed");
+        assert_eq!(outcome.state.executions.len(), 1);
+        assert_eq!(outcome.state.executions[0].result, "EXEC");
+    }
+
+    #[tokio::test]
+    async fn checkpoint_below_current_schema_version_expires_to_fresh_run() {
+        // A decodable but OLD-schema checkpoint (schema_version defaults to 0 —
+        // e.g. a pre-#3884 record whose executions happened to be empty) must be
+        // expired, not resumed: `schema_version` is a real guard, not just a doc.
+        let dir = tempfile::tempdir().unwrap();
+        let seed: tinyagents::graph::checkpoint::FileCheckpointer<DelegationState> =
+            tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path());
+        let checkpoint = Checkpoint {
+            thread_id: "old-schema".to_string(),
+            checkpoint_id: "cp-old".to_string(),
+            run_id: None,
+            parent_checkpoint_id: None,
+            namespace: vec![],
+            state: DelegationState {
+                plan: Some("stale".to_string()),
+                ..Default::default()
+            },
+            next_nodes: vec![],
+            completed_tasks: vec![],
+            pending_writes: vec![],
+            interrupts: vec![],
+            pending_activations: None,
+            barrier_arrivals: vec![],
+            metadata: json!({}),
+        };
+        assert_eq!(
+            checkpoint.state.schema_version, 0,
+            "an un-stamped record is version 0"
+        );
+        seed.put(checkpoint)
+            .await
+            .expect("seed old-schema checkpoint");
+
+        let cp: Arc<dyn Checkpointer<DelegationState>> =
+            Arc::new(tinyagents::graph::checkpoint::FileCheckpointer::<
+                DelegationState,
+            >::new(dir.path()));
+        let config = DelegationConfig {
+            checkpointer: Some(cp),
+            thread_id: Some("old-schema".to_string()),
+            ..DelegationConfig::default()
+        };
+        let outcome = run_or_resume_delegation(config, flow_runner(0))
+            .await
+            .expect("expires + fresh");
+        assert!(outcome.state.final_output.is_some(), "fresh run completed");
+        assert_eq!(
+            outcome.state.schema_version, CURRENT_SCHEMA_VERSION,
+            "fresh run stamped the current version"
+        );
+        assert_eq!(
+            outcome.state.plan.as_deref(),
+            Some("PLAN"),
+            "re-planned from scratch, not resumed with the stale plan"
+        );
+    }
+
+    #[test]
+    fn incompatible_checkpoint_error_matches_decode_not_operational() {
+        use tinyagents::TinyAgentsError;
+        // Decode / shape-incompatibility → safe to expire.
+        assert!(is_incompatible_checkpoint_error(
+            &TinyAgentsError::Checkpoint(
+                "sqlite checkpointer: decode record: invalid type: string".to_string()
+            )
+        ));
+        assert!(is_incompatible_checkpoint_error(
+            &TinyAgentsError::Checkpoint("sqlite checkpointer: decode next_nodes: eof".to_string())
+        ));
+        // Operational failures must NOT be treated as incompatible (they must
+        // propagate, not silently restart durable work).
+        assert!(!is_incompatible_checkpoint_error(
+            &TinyAgentsError::Checkpoint(
+                "sqlite checkpointer: query latest checkpoint: database is locked".to_string()
+            )
+        ));
+        assert!(!is_incompatible_checkpoint_error(
+            &TinyAgentsError::Checkpoint(
+                "sqlite checkpointer: connection lock poisoned".to_string()
+            )
+        ));
+        assert!(!is_incompatible_checkpoint_error(&TinyAgentsError::Resume(
+            "no checkpoint".to_string()
+        )));
+    }
+
+    #[tokio::test]
+    async fn terminal_checkpoint_with_a_pending_interrupt_surfaces_it() {
+        // A terminal-classified checkpoint that still carries an interrupt must
+        // surface it, not silently drop `pending`. (The live routing never
+        // produces this shape; the terminal branch is defensive.)
+        let dir = tempfile::tempdir().unwrap();
+        let seed: tinyagents::graph::checkpoint::FileCheckpointer<DelegationState> =
+            tinyagents::graph::checkpoint::FileCheckpointer::new(dir.path());
+        let mut state = DelegationState::new_run();
+        state.final_output = Some("done".to_string());
+        let checkpoint = Checkpoint {
+            thread_id: "terminal-interrupt".to_string(),
+            checkpoint_id: "cp-ti".to_string(),
+            run_id: None,
+            parent_checkpoint_id: None,
+            namespace: vec![],
+            state,
+            next_nodes: vec![],
+            completed_tasks: vec![],
+            pending_writes: vec![],
+            interrupts: vec![Interrupt::with_id(
+                "intr-1",
+                "approval",
+                json!({ "kind": "delegation_review" }),
+            )],
+            pending_activations: None,
+            barrier_arrivals: vec![],
+            metadata: json!({}),
+        };
+        seed.put(checkpoint).await.expect("seed terminal+interrupt");
+
+        let cp: Arc<dyn Checkpointer<DelegationState>> =
+            Arc::new(tinyagents::graph::checkpoint::FileCheckpointer::<
+                DelegationState,
+            >::new(dir.path()));
+        let config = DelegationConfig {
+            checkpointer: Some(cp),
+            thread_id: Some("terminal-interrupt".to_string()),
+            ..DelegationConfig::default()
+        };
+        let outcome = run_or_resume_delegation(config, flow_runner(0))
+            .await
+            .expect("terminal");
+        let pending = outcome
+            .pending
+            .expect("the carried interrupt is surfaced, not dropped");
+        assert_eq!(pending.node, "approval");
+        assert_eq!(pending.interrupt_id, "intr-1");
+        assert_eq!(outcome.state.final_output.as_deref(), Some("done"));
+    }
 }