feat(topology): multi-topology comparison harness (Phase 4 core)

compare(graphs, task, executor, scorer) runs the same task across a set of
topologies on the same executor, scores each, and returns a Comparison:
- per-topology results (quality, tokens, turns, blocked approvals, output),
- a leaderboard (quality desc),
- a quality/cost Pareto front (on_pareto flags),
- best_quality and best_value (quality-per-token) picks.

This is the "which patterns yield better results" engine and the structured
output the paper's benchmark tables consume. Generic over TurnExecutor +
Scorer (pluggable LLM-judge later); pure core, no new deps. 9 tests, clippy clean.

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

crates/cm-orchestrator/src/harness.rs

@@ -0,0 +1,207 @@
+//! Multi-topology comparison harness — milestone B / paper data.
+//!
+//! Run the *same task* across a set of topologies on the *same executor*, score
+//! each outcome, and surface a leaderboard + a quality/cost Pareto front. This
+//! is what reveals "which patterns yield better results" for a task type, and
+//! the structured [`Comparison`] is exactly what the paper's benchmark tables
+//! consume.
+
+use serde::Serialize;
+
+use cm_topology::{TopologyGraph, TopologyKind};
+
+use crate::{execute, OrchestratorError, RunRecord, TurnExecutor};
+
+/// Scores the quality of a run's final output in `[0,1]`. Real deployments use
+/// an LLM judge; tests/benchmarks can use a deterministic scorer.
+pub trait Scorer {
+    /// Quality of `record` for `task`, in `[0,1]` (higher is better).
+    fn score(&self, task: &str, record: &RunRecord) -> f64;
+}
+
+/// One topology's result within a comparison.
+#[derive(Debug, Clone, Serialize)]
+pub struct TopologyResult {
+    /// The topology that produced this result.
+    pub kind: TopologyKind,
+    /// Quality in `[0,1]`.
+    pub quality: f64,
+    /// Total tokens (cost proxy).
+    pub tokens: u64,
+    /// Turns run.
+    pub turns: u32,
+    /// Sandbox-leaving actions that were blocked (safety signal).
+    pub approvals_blocked: u32,
+    /// The final output text.
+    pub final_output: String,
+    /// Whether this result is on the quality/cost Pareto front.
+    pub on_pareto: bool,
+}
+
+/// The outcome of comparing several topologies on one task.
+#[derive(Debug, Clone, Serialize)]
+pub struct Comparison {
+    /// The task that was run.
+    pub task: String,
+    /// One result per input topology, in input order.
+    pub results: Vec<TopologyResult>,
+    /// Indices into `results`, best quality first.
+    pub leaderboard: Vec<usize>,
+    /// Index of the highest-quality result, if any.
+    pub best_quality: Option<usize>,
+    /// Index of the best quality-per-token ("value") result, if any.
+    pub best_value: Option<usize>,
+}
+
+/// "Value" = quality per token (bias toward quality when free).
+fn value(r: &TopologyResult) -> f64 {
+    if r.tokens == 0 {
+        r.quality * 1000.0
+    } else {
+        r.quality / r.tokens as f64
+    }
+}
+
+/// Run `task` across every topology in `graphs`, score each, and rank them.
+pub async fn compare<E: TurnExecutor, S: Scorer>(
+    graphs: &[TopologyGraph],
+    task: &str,
+    executor: &E,
+    scorer: &S,
+) -> Result<Comparison, OrchestratorError> {
+    let mut results: Vec<TopologyResult> = Vec::with_capacity(graphs.len());
+    for g in graphs {
+        let rec = execute(g, task, executor).await?;
+        let quality = scorer.score(task, &rec).clamp(0.0, 1.0);
+        results.push(TopologyResult {
+            kind: rec.kind,
+            quality,
+            tokens: rec.totals.tokens,
+            turns: rec.totals.turns,
+            approvals_blocked: rec.totals.approvals_blocked,
+            final_output: rec.final_output,
+            on_pareto: false,
+        });
+    }
+
+    // Pareto: maximize quality, minimize tokens. A result is dominated if some
+    // other is at least as good on both and strictly better on one.
+    for i in 0..results.len() {
+        let dominated = results.iter().enumerate().any(|(j, o)| {
+            j != i
+                && o.quality >= results[i].quality
+                && o.tokens <= results[i].tokens
+                && (o.quality > results[i].quality || o.tokens < results[i].tokens)
+        });
+        results[i].on_pareto = !dominated;
+    }
+
+    let mut leaderboard: Vec<usize> = (0..results.len()).collect();
+    leaderboard.sort_by(|&a, &b| {
+        results[b]
+            .quality
+            .partial_cmp(&results[a].quality)
+            .unwrap_or(std::cmp::Ordering::Equal)
+    });
+
+    let best_quality = leaderboard.first().copied();
+    let best_value = (0..results.len()).max_by(|&a, &b| {
+        value(&results[a])
+            .partial_cmp(&value(&results[b]))
+            .unwrap_or(std::cmp::Ordering::Equal)
+    });
+
+    Ok(Comparison {
+        task: task.to_string(),
+        results,
+        leaderboard,
+        best_quality,
+        best_value,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{GatedAction, TurnOutcome, TurnRequest};
+    use cm_domain::GatedCategory;
+    use cm_topology::{Edge, EdgeKind, Node};
+
+    /// Echoes context so richer collaboration yields longer output; "risky"
+    /// nodes attempt a 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".into(),
+                    approved: false,
+                }]
+            } else {
+                vec![]
+            };
+            Ok(TurnOutcome {
+                output: format!("{}:{}", req.role, req.context.join(" ")),
+                tokens: 10,
+                gated,
+            })
+        }
+    }
+
+    /// Deterministic quality proxy: longer (richer) output scores higher.
+    struct LengthScorer;
+    impl Scorer for LengthScorer {
+        fn score(&self, _task: &str, record: &RunRecord) -> f64 {
+            (record.final_output.len() as f64 / 200.0).min(1.0)
+        }
+    }
+
+    fn pipeline() -> TopologyGraph {
+        TopologyGraph::new(
+            TopologyKind::Pipeline,
+            vec![Node::new("a", "r"), Node::new("b", "w")],
+            vec![Edge { from: "a".into(), to: "b".into(), kind: EdgeKind::PipesTo }],
+        )
+        .unwrap()
+    }
+    fn flat() -> TopologyGraph {
+        TopologyGraph::new(
+            TopologyKind::Swarm,
+            vec![Node::new("a", "r"), Node::new("b", "w"), Node::new("c", "coordinator")],
+            vec![],
+        )
+        .unwrap()
+    }
+
+    #[tokio::test]
+    async fn compares_and_ranks_topologies() {
+        let graphs = vec![pipeline(), flat()];
+        let cmp = compare(&graphs, "do the thing", &Echo, &LengthScorer)
+            .await
+            .unwrap();
+
+        assert_eq!(cmp.results.len(), 2);
+        assert_eq!(cmp.leaderboard.len(), 2);
+        // leaderboard is sorted by quality desc
+        let q0 = cmp.results[cmp.leaderboard[0]].quality;
+        let q1 = cmp.results[cmp.leaderboard[1]].quality;
+        assert!(q0 >= q1);
+        assert!(cmp.best_quality.is_some());
+        assert!(cmp.best_value.is_some());
+        // at least one topology is non-dominated
+        assert!(cmp.results.iter().any(|r| r.on_pareto));
+    }
+
+    #[tokio::test]
+    async fn surfaces_blocked_approvals_per_topology() {
+        let risky = TopologyGraph::new(
+            TopologyKind::Pipeline,
+            vec![Node::new("risky1", "r"), Node::new("b", "w")],
+            vec![Edge { from: "risky1".into(), to: "b".into(), kind: EdgeKind::PipesTo }],
+        )
+        .unwrap();
+        let cmp = compare(&[risky], "x", &Echo, &LengthScorer).await.unwrap();
+        assert_eq!(cmp.results[0].approvals_blocked, 1);
+    }
+}

crates/cm-orchestrator/src/lib.rs

@@ -12,9 +12,12 @@
 //! [`TurnExecutor`], so the control flow is fully testable with a scripted
 //! executor and is independent of the LLM/runtime.
 
+mod harness;
 mod plan;
 #[cfg(feature = "provider")]
 mod provider_executor;
+
+pub use harness::{compare, Comparison, Scorer, TopologyResult};
 #[cfg(feature = "provider")]
 pub use provider_executor::ProviderExecutor;