feat(topology): cm-orchestrator topology runtime engine (Phase 2)

A pure async control-flow engine that executes a task across a TopologyGraph
by sequencing safe agent turns. Safety by construction: the engine can only
invoke turns via a generic TurnExecutor — it performs no side effects itself,
so §15 gating (inside each turn) is inherited and switching topology cannot
escalate authority.

- TurnExecutor trait + TurnRequest/TurnOutcome (real impl will wrap
  cm-runtime::Runtime; tests use a scripted Echo executor).
- Pure planners (plan.rs): hierarchical (delegate down / synthesize up),
  pipeline (topo-ordered threading), swarm (parallel attempts + aggregate).
- RunRecord journal (per-step + RunMetrics: tokens, gated actions, approvals
  granted/blocked, turns) — feeds the Phase 4 comparison harness/paper.
- Unsupported kinds return an error (no panic). 5 tests, clippy clean.

Next (Phase 2b): a real TurnExecutor adapter over cm-runtime::send_message.

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

Cargo.lock

@@ -618,6 +618,17 @@ dependencies = [
  "toml",
 ]
 
+[[package]]
+name = "cm-orchestrator"
+version = "0.1.0"
+dependencies = [
+ "cm-domain",
+ "cm-topology",
+ "serde",
+ "thiserror",
+ "tokio",
+]
+
 [[package]]
 name = "cm-runtime"
 version = "0.1.0"

Cargo.toml

@@ -3,6 +3,7 @@ resolver = "2"
 members = [
     "crates/cm-domain",
     "crates/cm-topology",
+    "crates/cm-orchestrator",
     "crates/cm-config",
     "crates/cm-db",
     "crates/cm-llm",

crates/cm-orchestrator/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "cm-orchestrator"
+version = "0.1.0"
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+publish.workspace = true
+
+[dependencies]
+cm-topology = { path = "../cm-topology" }
+cm-domain = { path = "../cm-domain" }
+serde = { workspace = true }
+thiserror = { workspace = true }
+
+[dev-dependencies]
+tokio = { workspace = true }
+
+[lints]
+workspace = true

crates/cm-orchestrator/src/lib.rs

@@ -0,0 +1,318 @@
+//! 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 plan;
+
+use cm_domain::GatedCategory;
+use cm_topology::{TopologyGraph, TopologyKind};
+use serde::Serialize;
+
+/// Errors from planning or running a topology.
+#[derive(Debug, thiserror::Error)]
+pub enum OrchestratorError {
+    /// No executor exists yet for this topology kind.
+    #[error("topology kind not yet supported by the runtime: {0:?}")]
+    Unsupported(TopologyKind),
+    /// 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)]
+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)]
+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,
+    /// 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)]
+#[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)]
+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,
+}
+
+/// Execute `task` over `graph` using `executor`, returning a full journal.
+pub async fn execute<E: TurnExecutor>(
+    graph: &TopologyGraph,
+    task: &str,
+    executor: &E,
+) -> Result<RunRecord, OrchestratorError> {
+    let steps = match graph.kind {
+        TopologyKind::Hierarchical => plan::hierarchical(graph)?,
+        TopologyKind::Pipeline => plan::pipeline(graph)?,
+        TopologyKind::Swarm => plan::swarm(graph)?,
+        other => return Err(OrchestratorError::Unsupported(other)),
+    };
+
+    let mut outputs: Vec<String> = Vec::with_capacity(steps.len());
+    let mut records: Vec<StepRecord> = Vec::with_capacity(steps.len());
+    let mut totals = RunMetrics::default();
+
+    for ps in &steps {
+        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(),
+                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,
+        });
+    }
+
+    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 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 unsupported_kind_errors() {
+        let graph = g(TopologyKind::Mesh, &["a", "b"], &[("a", "b")]);
+        let err = execute(&graph, "task", &Echo).await.unwrap_err();
+        assert!(matches!(err, OrchestratorError::Unsupported(TopologyKind::Mesh)));
+    }
+}

crates/cm-orchestrator/src/plan.rs

@@ -0,0 +1,173 @@
+//! Pure, synchronous planners: turn a [`TopologyGraph`] into an ordered list
+//! of steps. Separated from execution so the control flow is unit-testable
+//! without any async/LLM machinery.
+
+use std::collections::{HashMap, HashSet, VecDeque};
+
+use cm_topology::TopologyGraph;
+
+use crate::{OrchestratorError, StepPhase};
+
+/// One planned step: which node acts, in what phase, and which prior step
+/// outputs (plus optionally the top-level task) form its context.
+pub(crate) struct PlanStep {
+    pub node_idx: usize,
+    pub phase: StepPhase,
+    pub use_task: bool,
+    pub ctx_from: Vec<usize>,
+}
+
+fn index_map(g: &TopologyGraph) -> HashMap<&str, usize> {
+    g.nodes
+        .iter()
+        .enumerate()
+        .map(|(i, n)| (n.id.as_str(), i))
+        .collect()
+}
+
+fn in_degrees(g: &TopologyGraph, idx: &HashMap<&str, usize>) -> Vec<usize> {
+    let mut indeg = vec![0usize; g.nodes.len()];
+    let mut seen = HashSet::new();
+    for e in &g.edges {
+        if let (Some(&a), Some(&b)) = (idx.get(e.from.as_str()), idx.get(e.to.as_str())) {
+            if a != b && seen.insert((a, b)) {
+                indeg[b] += 1;
+            }
+        }
+    }
+    indeg
+}
+
+/// Hierarchical: root plans (top-down), direct children work, root synthesizes
+/// (bottom-up). Deeper trees are flattened to one level for v1.
+pub(crate) fn hierarchical(g: &TopologyGraph) -> Result<Vec<PlanStep>, OrchestratorError> {
+    let idx = index_map(g);
+    let n = g.nodes.len();
+    let indeg = in_degrees(g, &idx);
+    let root = (0..n).find(|&i| indeg[i] == 0).unwrap_or(0);
+
+    let mut children = Vec::new();
+    let mut seen = HashSet::new();
+    for e in &g.edges {
+        if e.from == g.nodes[root].id {
+            if let Some(&c) = idx.get(e.to.as_str()) {
+                if c != root && seen.insert(c) {
+                    children.push(c);
+                }
+            }
+        }
+    }
+
+    let mut steps = Vec::new();
+    if children.is_empty() {
+        steps.push(PlanStep {
+            node_idx: root,
+            phase: StepPhase::Work,
+            use_task: true,
+            ctx_from: vec![],
+        });
+        return Ok(steps);
+    }
+    steps.push(PlanStep {
+        node_idx: root,
+        phase: StepPhase::Plan,
+        use_task: true,
+        ctx_from: vec![],
+    });
+    let mut child_steps = Vec::new();
+    for c in children {
+        steps.push(PlanStep {
+            node_idx: c,
+            phase: StepPhase::Work,
+            use_task: true,
+            ctx_from: vec![0],
+        });
+        child_steps.push(steps.len() - 1);
+    }
+    steps.push(PlanStep {
+        node_idx: root,
+        phase: StepPhase::Synth,
+        use_task: false,
+        ctx_from: child_steps,
+    });
+    Ok(steps)
+}
+
+/// Pipeline: topological order; each stage takes the previous stage's output.
+pub(crate) fn pipeline(g: &TopologyGraph) -> Result<Vec<PlanStep>, OrchestratorError> {
+    let idx = index_map(g);
+    let n = g.nodes.len();
+    let mut succ = vec![Vec::new(); n];
+    let mut indeg = vec![0usize; n];
+    let mut seen = HashSet::new();
+    for e in &g.edges {
+        if let (Some(&a), Some(&b)) = (idx.get(e.from.as_str()), idx.get(e.to.as_str())) {
+            if a != b && seen.insert((a, b)) {
+                succ[a].push(b);
+                indeg[b] += 1;
+            }
+        }
+    }
+    let mut queue: VecDeque<usize> = (0..n).filter(|&i| indeg[i] == 0).collect();
+    let mut order = Vec::new();
+    while let Some(u) = queue.pop_front() {
+        order.push(u);
+        for &v in &succ[u] {
+            indeg[v] -= 1;
+            if indeg[v] == 0 {
+                queue.push_back(v);
+            }
+        }
+    }
+    // Any nodes left in a cycle: append in node order so we still run them.
+    for i in 0..n {
+        if !order.contains(&i) {
+            order.push(i);
+        }
+    }
+
+    let steps = order
+        .iter()
+        .enumerate()
+        .map(|(i, &node)| PlanStep {
+            node_idx: node,
+            phase: StepPhase::Work,
+            use_task: i == 0,
+            ctx_from: if i == 0 { vec![] } else { vec![i - 1] },
+        })
+        .collect();
+    Ok(steps)
+}
+
+/// Swarm: every node attempts the task independently, then an aggregator
+/// (a "coordinator" role, else a source node) synthesizes all outputs.
+pub(crate) fn swarm(g: &TopologyGraph) -> Result<Vec<PlanStep>, OrchestratorError> {
+    let n = g.nodes.len();
+    if n == 0 {
+        return Err(OrchestratorError::Malformed("empty topology".into()));
+    }
+    let idx = index_map(g);
+    let indeg = in_degrees(g, &idx);
+
+    let mut steps: Vec<PlanStep> = (0..n)
+        .map(|i| PlanStep {
+            node_idx: i,
+            phase: StepPhase::Work,
+            use_task: true,
+            ctx_from: vec![],
+        })
+        .collect();
+
+    let aggregator = (0..n)
+        .find(|&i| g.nodes[i].role.to_lowercase().contains("coordinator"))
+        .or_else(|| (0..n).find(|&i| indeg[i] == 0))
+        .unwrap_or(0);
+
+    steps.push(PlanStep {
+        node_idx: aggregator,
+        phase: StepPhase::Aggregate,
+        use_task: false,
+        ctx_from: (0..n).collect(),
+    });
+    Ok(steps)
+}