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

//! Topology runtime: execute a task across a [`TopologyGraph`] by sequencing
//! safe agent *turns* according to the topology's pattern.
//!
//! **Safety by construction (§15):** this engine can only *invoke turns* via a
//! [`TurnExecutor`]; it never performs a side effect itself. Every
//! sandbox-leaving action is gated inside the turn (the real executor wraps
//! `cm-runtime`, which enforces §15 approvals + secret broker + audit).
//! Switching topology therefore cannot escalate an agent's authority — the
//! orchestrator has no capability beyond running turns.
//!
//! v1 ships three executors (hierarchical / pipeline / swarm) over a generic
//! [`TurnExecutor`], so the control flow is fully testable with a scripted
//! executor and is independent of the LLM/runtime.

mod evolve;
mod harness;
#[cfg(feature = "provider")]
mod judge;
mod plan;
#[cfg(feature = "provider")]
mod provider_executor;
mod workflow;

pub use evolve::{evolve, evolve_all, EliteCell, Evolution};
pub use harness::{compare, Comparison, Scorer, TopologyResult};
#[cfg(feature = "provider")]
pub use judge::JudgeScorer;
#[cfg(feature = "provider")]
pub use provider_executor::ProviderExecutor;
pub use workflow::{run_workflow, WorkflowRecord};

use cm_domain::GatedCategory;
use cm_topology::{TopologyGraph, TopologyKind};
use serde::{Deserialize, Serialize};

/// Errors from planning or running a topology.
#[derive(Debug, thiserror::Error)]
pub enum OrchestratorError {
    /// The graph could not be turned into a runnable plan.
    #[error("malformed topology: {0}")]
    Malformed(String),
    /// The underlying turn executor failed.
    #[error("turn executor failed: {0}")]
    Executor(String),
}

/// 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, Deserialize)]
pub struct GatedAction {
    /// The §15 category that required approval.
    pub category: GatedCategory,
    /// Human summary of the action (the previewed payload).
    pub summary: String,
    /// Whether a human approved it (false = blocked, nothing executed).
    pub approved: bool,
}

/// Aggregate cost/safety counters for a run (feeds the Phase 4 harness/paper).
#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize)]
pub struct RunMetrics {
    /// Total model tokens (cost proxy).
    pub tokens: u64,
    /// Number of sandbox-leaving actions attempted.
    pub gated_actions: u32,
    /// Of those, how many a human approved and were executed.
    pub approvals_granted: u32,
    /// Of those, how many were blocked (nothing executed).
    pub approvals_blocked: u32,
    /// Number of turns run.
    pub turns: u32,
}

/// Inputs handed to a single agent turn.
#[derive(Debug, Clone)]
pub struct TurnRequest {
    /// Topology node id (bound to a real claw by the executor).
    pub node_id: String,
    /// The node's role.
    pub role: String,
    /// Optional explicit agent/model alias for this node, from the graph
    /// (`node.attrs["agent"]`). When set, the executor binds this node to this
    /// alias directly — letting one request pin a different model per role
    /// (heterogeneous topologies) without reconfiguring the server. Falls back
    /// to the role→alias map when absent.
    pub agent: Option<String>,
    /// The full free-form node attributes (`node.attrs`). Carries the binding a
    /// recursive executor needs — `attrs["team_id"]` (company tier) or
    /// `attrs["company_id"]` (org tier) — so a "turn" can resolve and run the
    /// sub-topology one tier down. Leaf (claw) executors ignore this.
    pub attrs: std::collections::BTreeMap<String, String>,
    /// The top-level task.
    pub task: String,
    /// Upstream context (task and/or prior step outputs) for this turn.
    pub context: Vec<String>,
}

/// The result of a single agent turn.
#[derive(Debug, Clone)]
pub struct TurnOutcome {
    /// The turn's textual output.
    pub output: String,
    /// Model tokens spent (cost proxy).
    pub tokens: u64,
    /// Any sandbox-leaving actions attempted during the turn.
    pub gated: Vec<GatedAction>,
}

/// Runs one safe agent turn. The real impl wraps `cm-runtime::Runtime`
/// (which enforces §15); tests use a scripted executor.
#[allow(async_fn_in_trait)]
pub trait TurnExecutor {
    /// Execute a single turn and return its outcome.
    async fn run_turn(&self, req: TurnRequest) -> Result<TurnOutcome, OrchestratorError>;
}

/// The phase a step plays in its topology.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StepPhase {
    /// Top-down decomposition.
    Plan,
    /// Doing the work.
    Work,
    /// Bottom-up synthesis by a parent.
    Synth,
    /// Combining many parallel outputs.
    Aggregate,
}

/// A journaled record of one executed step.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StepRecord {
    /// Node that acted.
    pub node_id: String,
    /// Its role.
    pub role: String,
    /// The phase it played.
    pub phase: StepPhase,
    /// Its output.
    pub output: String,
    /// Gated actions it attempted.
    pub gated: Vec<GatedAction>,
    /// Tokens it spent.
    pub tokens: u64,
}

/// The full record of a topology run (journal + final output + totals).
#[derive(Debug, Clone, Serialize)]
pub struct RunRecord {
    /// The topology that was run.
    pub kind: TopologyKind,
    /// Ordered steps.
    pub steps: Vec<StepRecord>,
    /// The run's final output.
    pub final_output: String,
    /// Aggregate metrics.
    pub totals: RunMetrics,
}

/// A resumable checkpoint of a topology run: the steps completed so far plus the
/// state later steps depend on. Serialized into the durable job's `checkpoint`
/// so a crashed/restarted run continues from the next step. The step *plan* is
/// re-derived from the graph on resume (planners are deterministic), so only the
/// completed outputs/records/metrics need to persist.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct RunProgress {
    /// Number of plan steps already completed.
    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 => plan::swarm(graph)?,
        TopologyKind::Mesh | TopologyKind::Blackboard => plan::mesh(graph)?,
        TopologyKind::Debate => plan::debate(graph)?,
    })
}

/// Execute `task` over `graph` using `executor`, returning a full journal.
/// Runs the whole topology to completion in one go (the synchronous path).
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
}

/// 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 {
            context.push(task.to_string());
        }
        for &i in &ps.ctx_from {
            context.push(outputs[i].clone());
        }

        let outcome = executor
            .run_turn(TurnRequest {
                node_id: node.id.clone(),
                role: node.role.clone(),
                agent: node.attrs.get("agent").cloned(),
                attrs: node.attrs.clone(),
                task: task.to_string(),
                context,
            })
            .await?;

        totals.turns += 1;
        totals.tokens += outcome.tokens;
        totals.gated_actions += outcome.gated.len() as u32;
        totals.approvals_granted += outcome.gated.iter().filter(|g| g.approved).count() as u32;
        totals.approvals_blocked += outcome.gated.iter().filter(|g| !g.approved).count() as u32;

        outputs.push(outcome.output.clone());
        records.push(StepRecord {
            node_id: node.id.clone(),
            role: node.role.clone(),
            phase: ps.phase,
            output: outcome.output,
            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 {
        kind: graph.kind,
        final_output: outputs.last().cloned().unwrap_or_default(),
        steps: records,
        totals,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use cm_topology::{Edge, EdgeKind, Node, TopologyGraph};

    /// Deterministic executor: echoes role/node/context so threading is
    /// observable; any node id starting with "risky" attempts one blocked
    /// gated action.
    struct Echo;

    impl TurnExecutor for Echo {
        async fn run_turn(&self, req: TurnRequest) -> Result<TurnOutcome, OrchestratorError> {
            let gated = if req.node_id.starts_with("risky") {
                vec![GatedAction {
                    category: GatedCategory::OutboundMessage,
                    summary: "send email".into(),
                    approved: false,
                }]
            } else {
                vec![]
            };
            Ok(TurnOutcome {
                output: format!("{}({})<{}>", req.role, req.node_id, req.context.join("|")),
                tokens: 10,
                gated,
            })
        }
    }

    fn g(kind: TopologyKind, ids: &[&str], edges: &[(&str, &str)]) -> TopologyGraph {
        TopologyGraph::new(
            kind,
            ids.iter().map(|i| Node::new(*i, "worker")).collect(),
            edges
                .iter()
                .map(|(a, b)| Edge {
                    from: (*a).into(),
                    to: (*b).into(),
                    kind: EdgeKind::DelegatesTo,
                })
                .collect(),
        )
        .unwrap()
    }

    #[tokio::test]
    async fn hierarchical_delegates_then_synthesizes() {
        let graph = g(
            TopologyKind::Hierarchical,
            &["root", "a", "b"],
            &[("root", "a"), ("root", "b")],
        );
        let rec = execute(&graph, "task", &Echo).await.unwrap();
        let phases: Vec<_> = rec.steps.iter().map(|s| s.phase).collect();
        assert_eq!(
            phases,
            vec![
                StepPhase::Plan,
                StepPhase::Work,
                StepPhase::Work,
                StepPhase::Synth
            ]
        );
        // The synthesis step saw both children's outputs.
        let synth = rec.steps.last().unwrap();
        assert!(synth.output.contains("(a)"));
        assert!(synth.output.contains("(b)"));
        assert_eq!(rec.totals.turns, 4);
        assert_eq!(rec.totals.tokens, 40);
    }

    #[tokio::test]
    async fn pipeline_threads_output_forward() {
        let graph = g(
            TopologyKind::Pipeline,
            &["a", "b", "c"],
            &[("a", "b"), ("b", "c")],
        );
        let rec = execute(&graph, "task", &Echo).await.unwrap();
        assert_eq!(rec.steps.len(), 3);
        // c's context contains b's output, which contains a's output.
        assert!(rec.final_output.contains("(b)"));
        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(TopologyKind::Swarm, &["a", "b", "coord"], &[]);
        // give "coord" a coordinator role so it aggregates
        let mut graph = graph;
        graph.nodes[2].role = "coordinator".into();
        let rec = execute(&graph, "task", &Echo).await.unwrap();
        // 3 workers + 1 aggregate
        assert_eq!(rec.steps.len(), 4);
        assert_eq!(rec.steps[3].phase, StepPhase::Aggregate);
        let agg = rec.steps.last().unwrap();
        assert!(
            agg.output.contains("(a)")
                && agg.output.contains("(b)")
                && agg.output.contains("(coord)")
        );
    }

    #[tokio::test]
    async fn blocked_gated_action_is_recorded_not_executed() {
        let graph = g(TopologyKind::Pipeline, &["risky1", "b"], &[("risky1", "b")]);
        let rec = execute(&graph, "task", &Echo).await.unwrap();
        assert_eq!(rec.totals.gated_actions, 1);
        assert_eq!(rec.totals.approvals_blocked, 1);
        assert_eq!(rec.totals.approvals_granted, 0);
        // The orchestrator surfaced it but performed no side effect (by construction).
        assert!(!rec.steps[0].gated[0].approved);
    }

    #[tokio::test]
    async fn mesh_and_debate_execute() {
        // mesh: 2 rounds of n peers + 1 aggregate.
        let mesh = g(TopologyKind::Mesh, &["a", "b"], &[("a", "b")]);
        let rec = execute(&mesh, "task", &Echo).await.unwrap();
        assert_eq!(rec.steps.len(), 5);
        assert_eq!(rec.steps.last().unwrap().phase, StepPhase::Aggregate);

        // debate: propose, critique, revise, judge.
        let debate = g(TopologyKind::Debate, &["p", "c", "j"], &[]);
        let rec2 = execute(&debate, "task", &Echo).await.unwrap();
        let phases: Vec<_> = rec2.steps.iter().map(|s| s.phase).collect();
        assert_eq!(
            phases,
            vec![
                StepPhase::Work,
                StepPhase::Work,
                StepPhase::Synth,
                StepPhase::Aggregate
            ]
        );
    }

    #[tokio::test]
    async fn every_topology_kind_runs() {
        // The full catalog executes (no kind is unsupported).
        for kind in TopologyKind::ALL {
            let graph = g(kind, &["x", "y", "z"], &[("x", "y"), ("y", "z")]);
            let rec = execute(&graph, "task", &Echo).await.unwrap();
            assert!(rec.totals.turns >= 1, "{} ran no turns", kind.as_str());
        }
    }
}