feat(orchestration): resumable + migratable workflow checkpoints (Phase 4) (#54)

A workflow journals its completed steps so an interrupted run picks up from
the last completed step — on this node or, since the checkpoint is
serializable and the executor is a parameter, on another one (host-driven
migration). Step-boundary analogue of LoopCheckpoint (which checkpoints at
tool-round boundaries one level down).

- WorkflowCheckpoint { schema_version, workflow_id, steps, checkpoint_ms } +
  ensure_loadable() (rejects future schema versions, mirrors LoopCheckpoint).
- SessionStore gains save/load/delete_workflow_checkpoint (default no-ops);
  file store writes crash-atomically (temp+fsync+rename) and rejects future
  versions on load; memory store mirrors it.
- execute_steps_parallel_resumable: loads prior progress, skips completed
  steps (reusing cached outcomes), runs only the rest, rewrites the checkpoint
  at each step boundary, merges in spec order, and clears on full success.
  Records ONLY successful steps — a failed step retries on resume (its effect
  didn't complete) while a succeeded step's work is never redone.

Tests: checkpoint round-trip + future-version rejection + pre-v1 default;
resumable skips-completed + clears-on-success (with a fresh executor = the
migration path) + retains a successes-only checkpoint on partial failure; a
real-LLM \#[ignore] resumable workflow passing against .a3s/config.acl. Store
suite (36) + orchestration (14) green; fmt + clippy --lib --bins clean.

Co-authored-by: Claude <claude@anthropic.com>

Author: ZhiXiao-Lin

core/src/lib.rs

@@ -136,8 +136,9 @@ pub use llm::{
     Message, OpenAiClient, TokenUsage,
 };
 pub use orchestration::{
-    execute_pipeline, execute_steps_parallel, AgentExecutor, AgentStepSpec, PipelineStage,
-    StepOutcome,
+    execute_pipeline, execute_steps_parallel, execute_steps_parallel_resumable, AgentExecutor,
+    AgentStepSpec, PipelineStage, StepOutcome, WorkflowCheckpoint, WorkflowStepRecord,
+    WORKFLOW_CHECKPOINT_SCHEMA_VERSION,
 };
 pub use prompts::{AgentStyle, DetectionConfidence, PlanningMode, SystemPromptSlots};
 pub use run::{

core/src/orchestration/checkpoint.rs

@@ -0,0 +1,139 @@
+//! Workflow-level checkpoints: journal completed steps so an interrupted
+//! orchestration resumes from the longest completed prefix — on this node or,
+//! because the checkpoint is serializable and the executor is pluggable, on
+//! another one (host-driven migration).
+//!
+//! This is the step-boundary analogue of [`LoopCheckpoint`](crate::loop_checkpoint::LoopCheckpoint),
+//! which checkpoints at tool-round boundaries one level down.
+
+use super::executor::StepOutcome;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+/// Schema version. Bumped on incompatible format changes; loads from a future
+/// version are rejected (see [`WorkflowCheckpoint::ensure_loadable`]).
+pub const WORKFLOW_CHECKPOINT_SCHEMA_VERSION: u32 = 1;
+
+/// One completed step within a workflow.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct WorkflowStepRecord {
+    /// Matches the [`AgentStepSpec::task_id`](super::AgentStepSpec) of the
+    /// step that produced this outcome.
+    pub task_id: String,
+    /// The completed step's result.
+    pub outcome: StepOutcome,
+}
+
+/// Snapshot of a workflow's completed steps at a step boundary.
+///
+/// (`StepOutcome` contains a `serde_json::Value`, which is not `Eq`, so this
+/// derives `PartialEq` only.)
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct WorkflowCheckpoint {
+    /// Schema version — see [`WORKFLOW_CHECKPOINT_SCHEMA_VERSION`].
+    #[serde(default)]
+    pub schema_version: u32,
+    /// Logical workflow identifier the checkpoint is keyed by.
+    pub workflow_id: String,
+    /// The steps completed so far. A resuming run skips these and re-dispatches
+    /// only the rest.
+    pub steps: Vec<WorkflowStepRecord>,
+    /// Wall-clock timestamp when the checkpoint was written (Unix epoch ms).
+    pub checkpoint_ms: u64,
+}
+
+impl WorkflowCheckpoint {
+    /// Build a checkpoint from a map of completed `task_id -> outcome`.
+    pub fn from_completed(
+        workflow_id: impl Into<String>,
+        completed: &HashMap<String, StepOutcome>,
+        checkpoint_ms: u64,
+    ) -> Self {
+        let steps = completed
+            .iter()
+            .map(|(task_id, outcome)| WorkflowStepRecord {
+                task_id: task_id.clone(),
+                outcome: outcome.clone(),
+            })
+            .collect();
+        Self {
+            schema_version: WORKFLOW_CHECKPOINT_SCHEMA_VERSION,
+            workflow_id: workflow_id.into(),
+            steps,
+            checkpoint_ms,
+        }
+    }
+
+    /// The completed steps as a `task_id -> outcome` map.
+    pub fn completed(&self) -> HashMap<String, StepOutcome> {
+        self.steps
+            .iter()
+            .map(|r| (r.task_id.clone(), r.outcome.clone()))
+            .collect()
+    }
+
+    /// Reject a checkpoint written by a *newer*, incompatible schema version
+    /// than this build understands — mirrors
+    /// [`LoopCheckpoint::ensure_loadable`](crate::loop_checkpoint::LoopCheckpoint::ensure_loadable).
+    /// Field additions are absorbed by `#[serde(default)]`, so older (incl.
+    /// pre-v1 `0`) checkpoints always remain loadable.
+    pub fn ensure_loadable(&self) -> anyhow::Result<()> {
+        if self.schema_version > WORKFLOW_CHECKPOINT_SCHEMA_VERSION {
+            anyhow::bail!(
+                "workflow checkpoint {} has schema version {} but this build supports at \
+                 most {}; refusing to resume from an incompatible future checkpoint",
+                self.workflow_id,
+                self.schema_version,
+                WORKFLOW_CHECKPOINT_SCHEMA_VERSION
+            );
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn outcome(id: &str) -> StepOutcome {
+        StepOutcome {
+            task_id: id.to_string(),
+            session_id: format!("task-run-{id}"),
+            agent: "a".to_string(),
+            output: "o".to_string(),
+            success: true,
+            structured: None,
+        }
+    }
+
+    #[test]
+    fn round_trips_and_exposes_completed_map() {
+        let mut completed = HashMap::new();
+        completed.insert("t1".to_string(), outcome("t1"));
+        let cp = WorkflowCheckpoint::from_completed("wf", &completed, 123);
+        let back: WorkflowCheckpoint =
+            serde_json::from_str(&serde_json::to_string(&cp).unwrap()).unwrap();
+        assert_eq!(back, cp);
+        assert_eq!(back.schema_version, WORKFLOW_CHECKPOINT_SCHEMA_VERSION);
+        assert_eq!(back.checkpoint_ms, 123);
+        assert_eq!(back.completed().get("t1").unwrap().task_id, "t1");
+    }
+
+    #[test]
+    fn ensure_loadable_rejects_only_future_versions() {
+        let mut cp = WorkflowCheckpoint::from_completed("wf", &HashMap::new(), 0);
+        cp.ensure_loadable().expect("current version loadable");
+        cp.schema_version = 0;
+        cp.ensure_loadable().expect("pre-v1 loadable");
+        cp.schema_version = WORKFLOW_CHECKPOINT_SCHEMA_VERSION + 1;
+        let err = cp.ensure_loadable().unwrap_err();
+        assert!(err.to_string().contains("schema version"), "got: {err}");
+    }
+
+    #[test]
+    fn pre_v1_payload_without_schema_version_loads() {
+        let json = r#"{"workflow_id":"wf","steps":[],"checkpoint_ms":0}"#;
+        let cp: WorkflowCheckpoint = serde_json::from_str(json).unwrap();
+        assert_eq!(cp.schema_version, 0);
+    }
+}