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

//! 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,
            web_search: false,
        };

        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);
    }
}