clawmates/crates/cm-orchestrator/src/workflow.rs

//! 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![],
                spend: Default::default(),
            })
        }
    }

    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());
    }
}