feat(topology): executors for all 12 kinds (mesh + debate; mappings)
ci / gates (push) Has been cancelled
ci / rust (push) Has been cancelled
ci / sandbox-k8s (push) Has been cancelled
ci / frontend (push) Has been cancelled
ci / e2e (push) Has been cancelled

Every TopologyKind now runs, mapped to five execution patterns:
- hierarchical ← hub_spoke, star_moe, market
- pipeline     ← ring
- swarm        ← flat, holacratic
- mesh (new)   ← blackboard   (two peer-exchange rounds + aggregate)
- debate (new)                (propose → critique → revise → judge)

execute()'s match is now exhaustive (adding a kind upstream forces an executor),
so the Unsupported error is gone. Benchmark spans all five distinct patterns.
14 tests with --features provider; clippy clean. Doc updated.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
Omar Sobh
2026-06-15 20:54:12 -07:00
co-authored by Claude Opus 4.8
parent c595cebab0
commit e93a3cfb53
4 changed files with 130 additions and 18 deletions
@@ -63,6 +63,28 @@ fn swarm() -> TopologyGraph {
.unwrap() .unwrap()
} }
fn mesh() -> TopologyGraph {
TopologyGraph::new(
TopologyKind::Mesh,
vec![node("a", "analyst"), node("b", "analyst"), node("c", "analyst")],
vec![
Edge { from: "a".into(), to: "b".into(), kind: EdgeKind::PeersWith },
Edge { from: "b".into(), to: "c".into(), kind: EdgeKind::PeersWith },
Edge { from: "a".into(), to: "c".into(), kind: EdgeKind::PeersWith },
],
)
.unwrap()
}
fn debate() -> TopologyGraph {
TopologyGraph::new(
TopologyKind::Debate,
vec![node("proposer", "proposer"), node("critic", "critic"), node("judge", "judge")],
vec![],
)
.unwrap()
}
#[tokio::main] #[tokio::main]
async fn main() { async fn main() {
let (provider, model): (Arc<dyn LlmProvider>, String) = let (provider, model): (Arc<dyn LlmProvider>, String) =
@@ -78,7 +100,7 @@ async fn main() {
let executor = ProviderExecutor::new(provider, model.clone(), 512); let executor = ProviderExecutor::new(provider, model.clone(), 512);
let task = "Draft a go-to-market launch plan for a new product."; let task = "Draft a go-to-market launch plan for a new product.";
let graphs = vec![hierarchical(), pipeline(), swarm()]; let graphs = vec![hierarchical(), pipeline(), swarm(), mesh(), debate()];
let cmp = compare(&graphs, task, &executor, &LengthScorer) let cmp = compare(&graphs, task, &executor, &LengthScorer)
.await .await
+37 -11
View File
@@ -32,9 +32,6 @@ use serde::Serialize;
/// Errors from planning or running a topology. /// Errors from planning or running a topology.
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum OrchestratorError { 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. /// The graph could not be turned into a runnable plan.
#[error("malformed topology: {0}")] #[error("malformed topology: {0}")]
Malformed(String), Malformed(String),
@@ -152,11 +149,19 @@ pub async fn execute<E: TurnExecutor>(
task: &str, task: &str,
executor: &E, executor: &E,
) -> Result<RunRecord, OrchestratorError> { ) -> Result<RunRecord, OrchestratorError> {
// Map every topology kind onto one of five execution patterns. The match
// is exhaustive, so adding a TopologyKind upstream forces a decision here.
let steps = match graph.kind { let steps = match graph.kind {
TopologyKind::Hierarchical => plan::hierarchical(graph)?, TopologyKind::Hierarchical
TopologyKind::Pipeline => plan::pipeline(graph)?, | TopologyKind::HubSpoke
TopologyKind::Swarm => plan::swarm(graph)?, | TopologyKind::StarMoe
other => return Err(OrchestratorError::Unsupported(other)), | 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)?,
}; };
let mut outputs: Vec<String> = Vec::with_capacity(steps.len()); let mut outputs: Vec<String> = Vec::with_capacity(steps.len());
@@ -321,9 +326,30 @@ mod tests {
} }
#[tokio::test] #[tokio::test]
async fn unsupported_kind_errors() { async fn mesh_and_debate_execute() {
let graph = g(TopologyKind::Mesh, &["a", "b"], &[("a", "b")]); // mesh: 2 rounds of n peers + 1 aggregate.
let err = execute(&graph, "task", &Echo).await.unwrap_err(); let mesh = g(TopologyKind::Mesh, &["a", "b"], &[("a", "b")]);
assert!(matches!(err, OrchestratorError::Unsupported(TopologyKind::Mesh))); 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());
}
} }
} }
+60
View File
@@ -171,3 +171,63 @@ pub(crate) fn swarm(g: &TopologyGraph) -> Result<Vec<PlanStep>, OrchestratorErro
}); });
Ok(steps) Ok(steps)
} }
/// Mesh / blackboard: two rounds of peer exchange (round 2 sees every round-1
/// output), then the first peer aggregates — models iterative convergence.
pub(crate) fn mesh(g: &TopologyGraph) -> Result<Vec<PlanStep>, OrchestratorError> {
let n = g.nodes.len();
if n == 0 {
return Err(OrchestratorError::Malformed("empty topology".into()));
}
let mut steps = Vec::new();
for i in 0..n {
steps.push(PlanStep {
node_idx: i,
phase: StepPhase::Work,
use_task: true,
ctx_from: vec![],
});
}
let round1: Vec<usize> = (0..n).collect();
for i in 0..n {
steps.push(PlanStep {
node_idx: i,
phase: StepPhase::Work,
use_task: false,
ctx_from: round1.clone(),
});
}
let round2: Vec<usize> = (n..2 * n).collect();
steps.push(PlanStep {
node_idx: 0,
phase: StepPhase::Aggregate,
use_task: false,
ctx_from: round2,
});
Ok(steps)
}
/// Debate: proposer drafts, critic critiques, proposer revises, judge decides.
pub(crate) fn debate(g: &TopologyGraph) -> Result<Vec<PlanStep>, OrchestratorError> {
let n = g.nodes.len();
if n == 0 {
return Err(OrchestratorError::Malformed("empty topology".into()));
}
if n == 1 {
return Ok(vec![PlanStep {
node_idx: 0,
phase: StepPhase::Work,
use_task: true,
ctx_from: vec![],
}]);
}
let proposer = 0;
let critic = 1;
let judge = if n >= 3 { 2 } else { 0 };
Ok(vec![
PlanStep { node_idx: proposer, phase: StepPhase::Work, use_task: true, ctx_from: vec![] },
PlanStep { node_idx: critic, phase: StepPhase::Work, use_task: true, ctx_from: vec![0] },
PlanStep { node_idx: proposer, phase: StepPhase::Synth, use_task: false, ctx_from: vec![0, 1] },
PlanStep { node_idx: judge, phase: StepPhase::Aggregate, use_task: false, ctx_from: vec![2, 1] },
])
}
+10 -6
View File
@@ -50,12 +50,16 @@ does today. Switching topology cannot widen an agent's authority. This is the pl
paper's safety contribution. paper's safety contribution.
## Crate / component map ## Crate / component map
- **`crates/cm-topology`** (this phase): pure, offline — kinds, graph, adapter, classifier, heuristics. Tests only. - **`crates/cm-topology`** ✅: pure, offline — kinds, graph, adapter, classifier, heuristics.
- **`crates/cm-orchestrator`** (Phase 2): binds a `TopologyGraph` to real claws and executes a task per - **`crates/cm-orchestrator`** ✅ (Phase 2): executes a task across a `TopologyGraph` by sequencing safe
pattern, gating every sandbox-leaving edge via `cm-safety`; journals steps to `run_events`. turns via a generic `TurnExecutor`. **All 12 kinds run**, mapped to five execution patterns: hierarchical
- **Experiment harness** (Phase 4, in orchestrator): run a task across N topologies; capture (also hub_spoke/star_moe/market), pipeline (also ring), swarm (also flat/holacratic), mesh (also
cost/quality(`cm-llm` judge)/latency/#approvals/#steps; compute Pareto + leaderboard. Seeded + pinned model. blackboard), debate. `ProviderExecutor` (cm-llm) runs real tool-free turns today; a `cm-runtime`
- **`cm-api`**: CRUD topologies; create runs (single / workflow-of-topologies / comparison); fetch results. tool-using executor (full §15 + `run_events`) is the remaining adapter.
- **Experiment harness** ✅ (Phase 4 core, in orchestrator): `compare()` runs a task across N topologies;
captures tokens/quality(`JudgeScorer`)/turns/#blocked-approvals; computes Pareto + leaderboard. Runnable
via `examples/topology_bench.rs`.
- **`cm-api`** (next): CRUD topologies; create runs (single / workflow-of-topologies / comparison); fetch results.
- **`cm-db`**: `topologies`, `topology_runs`, `run_results`, `pareto_snapshots`. - **`cm-db`**: `topologies`, `topology_runs`, `run_results`, `pareto_snapshots`.
- **`frontend/`**: ReactFlow topology builder + Pareto/leaderboard view. - **`frontend/`**: ReactFlow topology builder + Pareto/leaderboard view.