Durable topology jobs (2/4): resumable orchestrator execute()

Split execute() into a thin wrapper over new execute_resumable(), which starts
from a prior RunProgress checkpoint (completed step count + outputs + records +
metrics) and invokes an async on_step callback after each newly completed step.
The step plan is re-derived from the graph (planners are deterministic), so only
completed outputs need persisting; the callback owns persistence, keeping the
orchestrator storage-agnostic. RunProgress + the journal types (StepRecord,
RunMetrics, StepPhase, GatedAction) gain Deserialize for JSONB round-trip.

New test: resume-from-checkpoint runs only the remaining steps and reproduces
the full run's output. 14 tests pass, clippy clean.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

crates/cm-orchestrator/src/lib.rs

@@ -31,7 +31,7 @@ pub use provider_executor::ProviderExecutor;
 
 use cm_domain::GatedCategory;
 use cm_topology::{TopologyGraph, TopologyKind};
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 
 /// Errors from planning or running a topology.
 #[derive(Debug, thiserror::Error)]
@@ -46,7 +46,7 @@ pub enum OrchestratorError {
 
 /// A sandbox-leaving action a turn attempted, and whether it was approved.
 /// (Reported by the executor; the orchestrator only aggregates it.)
-#[derive(Debug, Clone, Serialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct GatedAction {
     /// The §15 category that required approval.
     pub category: GatedCategory,
@@ -57,7 +57,7 @@ pub struct GatedAction {
 }
 
 /// Aggregate cost/safety counters for a run (feeds the Phase 4 harness/paper).
-#[derive(Debug, Clone, Copy, Default, Serialize)]
+#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)]
 pub struct RunMetrics {
     /// Total model tokens (cost proxy).
     pub tokens: u64,
@@ -110,7 +110,7 @@ pub trait TurnExecutor {
 }
 
 /// The phase a step plays in its topology.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum StepPhase {
     /// Top-down decomposition.
@@ -124,7 +124,7 @@ pub enum StepPhase {
 }
 
 /// A journaled record of one executed step.
-#[derive(Debug, Clone, Serialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct StepRecord {
     /// Node that acted.
     pub node_id: String,
@@ -153,32 +153,77 @@ pub struct RunRecord {
     pub totals: RunMetrics,
 }
 
-/// Execute `task` over `graph` using `executor`, returning a full journal.
+/// A resumable checkpoint of a topology run: the steps completed so far plus the
-pub async fn execute<E: TurnExecutor>(
+/// state later steps depend on. Serialized into the durable job's `checkpoint`
-    graph: &TopologyGraph,
+/// so a crashed/restarted run continues from the next step. The step *plan* is
-    task: &str,
+/// re-derived from the graph on resume (planners are deterministic), so only the
-    executor: &E,
+/// completed outputs/records/metrics need to persist.
-) -> Result<RunRecord, OrchestratorError> {
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
-    // Map every topology kind onto one of five execution patterns. The match
+pub struct RunProgress {
-    // is exhaustive, so adding a TopologyKind upstream forces a decision here.
+    /// Number of plan steps already completed.
-    let steps = match graph.kind {
+    pub completed: usize,
+    /// Output of each completed step, in plan order (later steps index into this).
+    pub outputs: Vec<String>,
+    /// Journaled step records so far.
+    pub records: Vec<StepRecord>,
+    /// Accumulated metrics so far.
+    pub totals: RunMetrics,
+}
+
+/// Map every topology kind onto one of five execution patterns. The match is
+/// exhaustive, so adding a `TopologyKind` upstream forces a decision here.
+fn plan_steps(graph: &TopologyGraph) -> Result<Vec<plan::PlanStep>, OrchestratorError> {
+    Ok(match graph.kind {
         TopologyKind::Hierarchical
         | TopologyKind::HubSpoke
         | TopologyKind::StarMoe
         | TopologyKind::Market => plan::hierarchical(graph)?,
         TopologyKind::Pipeline | TopologyKind::Ring => plan::pipeline(graph)?,
-        TopologyKind::Swarm | TopologyKind::Flat | TopologyKind::Holacratic => {
+        TopologyKind::Swarm | TopologyKind::Flat | TopologyKind::Holacratic => plan::swarm(graph)?,
-            plan::swarm(graph)?
-        }
         TopologyKind::Mesh | TopologyKind::Blackboard => plan::mesh(graph)?,
         TopologyKind::Debate => plan::debate(graph)?,
-    };
+    })
+}
 
-    let mut outputs: Vec<String> = Vec::with_capacity(steps.len());
+/// Execute `task` over `graph` using `executor`, returning a full journal.
-    let mut records: Vec<StepRecord> = Vec::with_capacity(steps.len());
+/// Runs the whole topology to completion in one go (the synchronous path).
-    let mut totals = RunMetrics::default();
+pub async fn execute<E: TurnExecutor>(
+    graph: &TopologyGraph,
+    task: &str,
+    executor: &E,
+) -> Result<RunRecord, OrchestratorError> {
+    execute_resumable(graph, task, executor, RunProgress::default(), |_| async {
+        Ok(())
+    })
+    .await
+}
 
-    for ps in &steps {
+/// Resumable execution: start from a prior [`RunProgress`] checkpoint (empty for
+/// a fresh run) and invoke `on_step` after each newly completed step with the
+/// updated progress. The caller persists that snapshot durably, so a worker that
+/// dies mid-run can reload the checkpoint and call this again to continue from
+/// the next step. Keeps the orchestrator storage-agnostic — persistence lives in
+/// the callback.
+pub async fn execute_resumable<E, F, Fut>(
+    graph: &TopologyGraph,
+    task: &str,
+    executor: &E,
+    progress: RunProgress,
+    mut on_step: F,
+) -> Result<RunRecord, OrchestratorError>
+where
+    E: TurnExecutor,
+    F: FnMut(RunProgress) -> Fut,
+    Fut: std::future::Future<Output = Result<(), OrchestratorError>>,
+{
+    let steps = plan_steps(graph)?;
+    let start = progress.completed.min(steps.len());
+
+    let mut outputs: Vec<String> = progress.outputs;
+    let mut records: Vec<StepRecord> = progress.records;
+    let mut totals = progress.totals;
+
+    for ps in steps.iter().skip(start) {
         let node = &graph.nodes[ps.node_idx];
         let mut context = Vec::new();
         if ps.use_task {
@@ -213,6 +258,15 @@ pub async fn execute<E: TurnExecutor>(
             gated: outcome.gated,
             tokens: outcome.tokens,
         });
+
+        // Hand the caller a durable snapshot to persist before the next turn.
+        on_step(RunProgress {
+            completed: records.len(),
+            outputs: outputs.clone(),
+            records: records.clone(),
+            totals,
+        })
+        .await?;
     }
 
     Ok(RunRecord {
@@ -303,6 +357,58 @@ mod tests {
         assert!(rec.final_output.contains("(a)"));
     }
 
+    #[tokio::test]
+    async fn resumes_from_checkpoint_without_rerunning_done_steps() {
+        use std::cell::RefCell;
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        // Echo that also counts how many turns it actually executes.
+        struct CountingEcho(AtomicUsize);
+        impl TurnExecutor for CountingEcho {
+            async fn run_turn(&self, req: TurnRequest) -> Result<TurnOutcome, OrchestratorError> {
+                self.0.fetch_add(1, Ordering::SeqCst);
+                Ok(TurnOutcome {
+                    output: format!("{}({})<{}>", req.role, req.node_id, req.context.join("|")),
+                    tokens: 10,
+                    gated: vec![],
+                })
+            }
+        }
+
+        let graph = g(
+            TopologyKind::Pipeline,
+            &["a", "b", "c"],
+            &[("a", "b"), ("b", "c")],
+        );
+
+        // Run #1: capture a progress snapshot after every step.
+        let snaps: RefCell<Vec<RunProgress>> = RefCell::new(Vec::new());
+        let c1 = CountingEcho(AtomicUsize::new(0));
+        let full = execute_resumable(&graph, "task", &c1, RunProgress::default(), |p| {
+            snaps.borrow_mut().push(p);
+            async { Ok(()) }
+        })
+        .await
+        .unwrap();
+        assert_eq!(c1.0.load(Ordering::SeqCst), 3, "fresh run executes all 3 steps");
+        let snaps = snaps.into_inner();
+        assert_eq!(snaps.len(), 3);
+
+        // Resume from the checkpoint taken after step 1 (simulating a crash).
+        let mid = snaps[0].clone();
+        assert_eq!(mid.completed, 1);
+        let c2 = CountingEcho(AtomicUsize::new(0));
+        let resumed = execute_resumable(&graph, "task", &c2, mid, |_| async { Ok(()) })
+            .await
+            .unwrap();
+
+        // Only the remaining 2 steps re-run; the result matches the full run.
+        assert_eq!(c2.0.load(Ordering::SeqCst), 2, "resume runs only remaining steps");
+        assert_eq!(resumed.steps.len(), 3);
+        assert_eq!(resumed.totals.turns, 3);
+        assert_eq!(resumed.final_output, full.final_output);
+    }
+
     #[tokio::test]
     async fn swarm_aggregates_all_workers() {
         let graph = g(