feat(topology): cm-llm-backed ProviderExecutor (Phase 2b)

Add a real TurnExecutor (behind the `provider` feature) that runs each
topology turn as a single tool-free LLM call via any cm-llm provider
(Anthropic / OpenAI-compat / scripted). Topologies now execute real model
calls and produce real metrics (tokens), feeding the comparison harness.

- ProviderExecutor builds a per-role system prompt + threads upstream context
  into the user message; collects TextDelta → output, Usage → tokens.
- Tool-free reasoning turns take no sandbox-leaving actions (gated = []); §15
  remains satisfied. Tool-using turns will route through a cm-runtime adapter.
- Core crate stays dependency-light; cm-llm/futures are optional (feature).
- Tests: pipeline + hierarchical run over the deterministic scripted provider
  (7 tests with --features provider). Clippy clean.

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

crates/cm-orchestrator/src/provider_executor.rs

@@ -0,0 +1,152 @@
+//! A [`TurnExecutor`] that runs each turn as a single, tool-free LLM call via
+//! a `cm-llm` provider.
+//!
+//! Pure-reasoning turns take no sandbox-leaving actions, so §15 is trivially
+//! satisfied (no `GatedAction`s). Tool-using turns — which need real §15
+//! approvals + the secret broker — will go through a `cm-runtime`-backed
+//! executor in a later step; this one is enough to drive real model calls for
+//! the multi-topology comparison harness.
+
+use std::sync::Arc;
+
+use cm_llm::{ChatMessage, ChatRequest, ChatRole, ContentPart, LlmEvent, LlmProvider};
+use futures::StreamExt;
+
+use crate::{OrchestratorError, TurnExecutor, TurnOutcome, TurnRequest};
+
+/// Runs topology turns against any `cm-llm` provider (Anthropic, OpenAI-compat,
+/// or the deterministic scripted provider).
+pub struct ProviderExecutor {
+    provider: Arc<dyn LlmProvider>,
+    model: String,
+    max_tokens: u32,
+}
+
+impl ProviderExecutor {
+    /// Build an executor over a shared provider, model id, and token budget.
+    pub fn new(provider: Arc<dyn LlmProvider>, model: impl Into<String>, max_tokens: u32) -> Self {
+        ProviderExecutor {
+            provider,
+            model: model.into(),
+            max_tokens,
+        }
+    }
+
+    fn system_for(role: &str) -> String {
+        format!(
+            "You are the \"{role}\" agent in a multi-agent system. Do your part of the task \
+             concisely and return only your result."
+        )
+    }
+
+    fn user_message(req: &TurnRequest) -> String {
+        let mut s = format!("Task: {}", req.task);
+        if !req.context.is_empty() {
+            s.push_str("\n\nContext from upstream agents:\n");
+            for (i, c) in req.context.iter().enumerate() {
+                s.push_str(&format!("[{i}] {c}\n"));
+            }
+        }
+        s
+    }
+}
+
+impl TurnExecutor for ProviderExecutor {
+    async fn run_turn(&self, req: TurnRequest) -> Result<TurnOutcome, OrchestratorError> {
+        let request = ChatRequest {
+            system: Self::system_for(&req.role),
+            messages: vec![ChatMessage {
+                role: ChatRole::User,
+                parts: vec![ContentPart::text(Self::user_message(&req))],
+            }],
+            tools: vec![],
+            model: self.model.clone(),
+            max_tokens: self.max_tokens,
+        };
+
+        let mut stream = self
+            .provider
+            .stream(request)
+            .await
+            .map_err(|e| OrchestratorError::Executor(e.to_string()))?;
+
+        let mut output = String::new();
+        let mut tokens: u64 = 0;
+        while let Some(event) = stream.next().await {
+            match event.map_err(|e| OrchestratorError::Executor(e.to_string()))? {
+                LlmEvent::TextDelta(t) => output.push_str(&t),
+                LlmEvent::Usage {
+                    input_tokens,
+                    output_tokens,
+                } => tokens += u64::from(input_tokens) + u64::from(output_tokens),
+                LlmEvent::ToolUse { .. } | LlmEvent::Stop(_) => {}
+            }
+        }
+
+        Ok(TurnOutcome {
+            output: output.trim().to_string(),
+            tokens,
+            // Tool-free reasoning turns leave the sandbox nowhere.
+            gated: vec![],
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::execute;
+    use cm_topology::{Edge, EdgeKind, Node, TopologyGraph, TopologyKind};
+
+    fn scripted() -> Arc<dyn LlmProvider> {
+        // No scenarios → deterministic echo with word-count token accounting.
+        Arc::new(cm_llm::ScriptedProvider::from_toml("").unwrap())
+    }
+
+    #[tokio::test]
+    async fn runs_a_pipeline_with_real_provider_calls() {
+        let exec = ProviderExecutor::new(scripted(), "test-model", 256);
+        let graph = TopologyGraph::new(
+            TopologyKind::Pipeline,
+            vec![Node::new("a", "researcher"), Node::new("b", "writer")],
+            vec![Edge {
+                from: "a".into(),
+                to: "b".into(),
+                kind: EdgeKind::PipesTo,
+            }],
+        )
+        .unwrap();
+
+        let rec = execute(&graph, "Summarize the quarterly plan", &exec)
+            .await
+            .unwrap();
+
+        assert_eq!(rec.steps.len(), 2);
+        assert_eq!(rec.totals.turns, 2);
+        assert!(rec.totals.tokens > 0, "scripted provider meters tokens");
+        assert_eq!(rec.totals.gated_actions, 0, "reasoning turns are tool-free");
+        assert!(!rec.final_output.is_empty());
+    }
+
+    #[tokio::test]
+    async fn hierarchical_runs_over_provider() {
+        let exec = ProviderExecutor::new(scripted(), "test-model", 256);
+        let graph = TopologyGraph::new(
+            TopologyKind::Hierarchical,
+            vec![
+                Node::new("lead", "coordinator"),
+                Node::new("w1", "worker"),
+                Node::new("w2", "worker"),
+            ],
+            vec![
+                Edge { from: "lead".into(), to: "w1".into(), kind: EdgeKind::DelegatesTo },
+                Edge { from: "lead".into(), to: "w2".into(), kind: EdgeKind::DelegatesTo },
+            ],
+        )
+        .unwrap();
+
+        let rec = execute(&graph, "Plan a launch", &exec).await.unwrap();
+        assert_eq!(rec.totals.turns, 4); // plan + 2 workers + synth
+        assert!(rec.totals.tokens > 0);
+    }
+}