feat(topology): workflow of topologies (run_workflow)

Milestone-A second capability: chain whole topology runs in sequence, threading
each stage's final output into the next stage's task (e.g. swarm brainstorm →
hierarchical execute → debate review). Each stage is a full safe topology run,
so §15 holds at every step. WorkflowRecord aggregates per-stage RunRecords +
totals. Demonstrated in the benchmark example. 14 tests; clippy clean.

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

crates/cm-orchestrator/examples/topology_bench.rs

@@ -15,7 +15,7 @@
 use std::sync::Arc;
 
 use cm_llm::LlmProvider;
-use cm_orchestrator::{compare, ProviderExecutor, RunRecord, Scorer};
+use cm_orchestrator::{compare, run_workflow, ProviderExecutor, RunRecord, Scorer};
 use cm_topology::{Edge, EdgeKind, Node, TopologyGraph, TopologyKind};
 
 /// Deterministic quality proxy: richer (longer) output scores higher.
@@ -130,4 +130,16 @@ async fn main() {
     if let Some(i) = cmp.best_value {
         println!("best value:   {:?} (quality per token)", cmp.results[i].kind);
     }
+
+    // Workflow of topologies: brainstorm (swarm) → execute (hierarchical) → review (debate).
+    let wf = run_workflow(&[swarm(), hierarchical(), debate()], task, &executor)
+        .await
+        .expect("workflow failed");
+    println!("\nworkflow: swarm -> hierarchical -> debate");
+    println!(
+        "  stages: {}   turns: {}   tokens: {}",
+        wf.stages.len(),
+        wf.totals.turns,
+        wf.totals.tokens
+    );
 }

crates/cm-orchestrator/src/lib.rs

@@ -14,12 +14,14 @@
 
 mod harness;
 mod plan;
+mod workflow;
 #[cfg(feature = "provider")]
 mod judge;
 #[cfg(feature = "provider")]
 mod provider_executor;
 
 pub use harness::{compare, Comparison, Scorer, TopologyResult};
+pub use workflow::{run_workflow, WorkflowRecord};
 #[cfg(feature = "provider")]
 pub use judge::JudgeScorer;
 #[cfg(feature = "provider")]

crates/cm-orchestrator/src/workflow.rs

@@ -0,0 +1,117 @@
+//! Workflow of topologies: run several topologies in sequence, threading each
+//! stage's final output into the next stage's task.
+//!
+//! This is the "take my project through a workflow of topologies" capability —
+//! e.g. brainstorm in a `swarm`, execute in a `hierarchy`, review via `debate`.
+//! Each stage is itself a full safe topology run, so §15 still holds at every
+//! step.
+
+use serde::Serialize;
+
+use cm_topology::TopologyGraph;
+
+use crate::{execute, OrchestratorError, RunMetrics, RunRecord, TurnExecutor};
+
+/// The record of a multi-stage workflow run.
+#[derive(Debug, Clone, Serialize)]
+pub struct WorkflowRecord {
+    /// The original task.
+    pub task: String,
+    /// One full topology run per stage, in order.
+    pub stages: Vec<RunRecord>,
+    /// The last stage's final output.
+    pub final_output: String,
+    /// Totals across all stages.
+    pub totals: RunMetrics,
+}
+
+/// Run `stages` in order; each stage after the first receives the previous
+/// stage's final output as additional task context.
+pub async fn run_workflow<E: TurnExecutor>(
+    stages: &[TopologyGraph],
+    task: &str,
+    executor: &E,
+) -> Result<WorkflowRecord, OrchestratorError> {
+    let mut records: Vec<RunRecord> = Vec::with_capacity(stages.len());
+    let mut totals = RunMetrics::default();
+    let mut prior: Option<String> = None;
+
+    for stage in stages {
+        let stage_task = match &prior {
+            None => task.to_string(),
+            Some(p) => format!("{task}\n\nBuilding on the previous stage's result:\n{p}"),
+        };
+        let rec = execute(stage, &stage_task, executor).await?;
+
+        totals.turns += rec.totals.turns;
+        totals.tokens += rec.totals.tokens;
+        totals.gated_actions += rec.totals.gated_actions;
+        totals.approvals_granted += rec.totals.approvals_granted;
+        totals.approvals_blocked += rec.totals.approvals_blocked;
+
+        prior = Some(rec.final_output.clone());
+        records.push(rec);
+    }
+
+    Ok(WorkflowRecord {
+        task: task.to_string(),
+        final_output: prior.unwrap_or_default(),
+        stages: records,
+        totals,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{TurnOutcome, TurnRequest};
+    use cm_topology::{Node, TopologyKind};
+
+    struct Echo;
+    impl TurnExecutor for Echo {
+        async fn run_turn(&self, req: TurnRequest) -> Result<TurnOutcome, OrchestratorError> {
+            Ok(TurnOutcome {
+                output: format!("{}<{}>", req.node_id, req.context.join("|")),
+                tokens: 5,
+                gated: vec![],
+            })
+        }
+    }
+
+    fn stage(kind: TopologyKind, ids: &[&str]) -> TopologyGraph {
+        TopologyGraph::new(
+            kind,
+            ids.iter().map(|i| Node::new(*i, "worker")).collect(),
+            vec![],
+        )
+        .unwrap()
+    }
+
+    #[tokio::test]
+    async fn workflow_threads_stages_and_aggregates_totals() {
+        let stages = vec![
+            stage(TopologyKind::Swarm, &["s1", "s2"]),
+            stage(TopologyKind::Pipeline, &["p1", "p2"]),
+        ];
+        let rec = run_workflow(&stages, "build it", &Echo).await.unwrap();
+
+        assert_eq!(rec.stages.len(), 2);
+        // Stage 2's output reflects stage 1's final output (threaded forward).
+        let s1_final = &rec.stages[0].final_output;
+        assert!(
+            rec.final_output.contains(&s1_final[..s1_final.len().min(4)]),
+            "stage 2 should build on stage 1"
+        );
+        // Totals are the sum of both stages.
+        let summed: u32 = rec.stages.iter().map(|s| s.totals.turns).sum();
+        assert_eq!(rec.totals.turns, summed);
+        assert!(rec.totals.tokens > 0);
+    }
+
+    #[tokio::test]
+    async fn empty_workflow_is_empty() {
+        let rec = run_workflow(&[], "x", &Echo).await.unwrap();
+        assert!(rec.stages.is_empty());
+        assert!(rec.final_output.is_empty());
+    }
+}