//! Director - Multi-Agent Orchestration (Experimental) //! //! The Director manages multi-agent conversations by: //! - Determining which agent speaks next //! - Managing conversation state and turn order //! - Supporting multiple scheduling strategies //! - Coordinating agent responses //! //! **Status**: This module is fully implemented but gated behind the `multi-agent` feature. //! The desktop build does not currently enable this feature. When multi-agent support //! is ready for production, add Tauri commands to create and interact with the Director, //! and enable the feature in `desktop/src-tauri/Cargo.toml`. use std::sync::Arc; use serde::{Deserialize, Serialize}; use tokio::sync::{RwLock, Mutex, mpsc}; use zclaw_types::{AgentId, Result, ZclawError}; use zclaw_protocols::{A2aEnvelope, A2aMessageType, A2aRecipient, A2aRouter, A2aAgentProfile, A2aCapability}; use zclaw_runtime::{LlmDriver, CompletionRequest}; /// Director configuration #[derive(Debug, Clone, Serialize, Deserialize)] pub struct DirectorConfig { /// Maximum turns before ending conversation pub max_turns: usize, /// Scheduling strategy pub strategy: ScheduleStrategy, /// Whether to include user in the loop pub include_user: bool, /// Timeout for agent response (seconds) pub response_timeout: u64, /// Whether to allow parallel agent responses pub allow_parallel: bool, } impl Default for DirectorConfig { fn default() -> Self { Self { max_turns: 50, strategy: ScheduleStrategy::Priority, include_user: true, response_timeout: 30, allow_parallel: false, } } } /// Scheduling strategy for determining next speaker #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum ScheduleStrategy { /// Round-robin through all agents RoundRobin, /// Priority-based selection (higher priority speaks first) Priority, /// LLM decides who speaks next LlmDecision, /// Random selection Random, /// Manual (external controller decides) Manual, } /// Agent role in the conversation #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)] #[serde(rename_all = "snake_case")] pub enum AgentRole { /// Main teacher/instructor Teacher, /// Teaching assistant Assistant, /// Student participant Student, /// Moderator/facilitator Moderator, /// Expert consultant Expert, /// Observer (receives messages but doesn't speak) Observer, } impl AgentRole { /// Get default priority for this role pub fn default_priority(&self) -> u8 { match self { AgentRole::Teacher => 10, AgentRole::Moderator => 9, AgentRole::Expert => 8, AgentRole::Assistant => 7, AgentRole::Student => 5, AgentRole::Observer => 0, } } } /// Agent configuration for director #[derive(Debug, Clone, Serialize, Deserialize)] pub struct DirectorAgent { /// Agent ID pub id: AgentId, /// Display name pub name: String, /// Agent role pub role: AgentRole, /// Priority (higher = speaks first) pub priority: u8, /// System prompt / persona pub persona: String, /// Whether this agent is active pub active: bool, /// Maximum turns this agent can speak consecutively pub max_consecutive_turns: usize, } impl DirectorAgent { /// Create a new director agent pub fn new(id: AgentId, name: impl Into, role: AgentRole, persona: impl Into) -> Self { let priority = role.default_priority(); Self { id, name: name.into(), role, priority, persona: persona.into(), active: true, max_consecutive_turns: 2, } } } /// Conversation state #[derive(Debug, Clone, Default)] pub struct ConversationState { /// Current turn number pub turn: usize, /// Current speaker ID pub current_speaker: Option, /// Turn history (agent_id, message_summary) pub history: Vec<(AgentId, String)>, /// Consecutive turns by current agent pub consecutive_turns: usize, /// Whether conversation is active pub active: bool, /// Conversation topic/goal pub topic: Option, } impl ConversationState { /// Create new conversation state pub fn new() -> Self { Self { turn: 0, current_speaker: None, history: Vec::new(), consecutive_turns: 0, active: false, topic: None, } } /// Record a turn pub fn record_turn(&mut self, agent_id: AgentId, summary: String) { if self.current_speaker == Some(agent_id) { self.consecutive_turns += 1; } else { self.consecutive_turns = 1; self.current_speaker = Some(agent_id); } self.history.push((agent_id, summary)); self.turn += 1; } /// Get last N turns pub fn get_recent_history(&self, n: usize) -> &[(AgentId, String)] { let start = self.history.len().saturating_sub(n); &self.history[start..] } /// Check if agent has spoken too many consecutive turns pub fn is_over_consecutive_limit(&self, agent_id: &AgentId, max: usize) -> bool { if self.current_speaker == Some(*agent_id) { self.consecutive_turns >= max } else { false } } } /// The Director orchestrates multi-agent conversations pub struct Director { /// Director configuration config: DirectorConfig, /// Registered agents agents: Arc>>, /// Conversation state state: Arc>, /// A2A router for messaging router: Arc, /// Agent ID for the director itself director_id: AgentId, /// Optional LLM driver for intelligent scheduling llm_driver: Option>, /// Inbox for receiving responses (stores pending request IDs and their response channels) pending_requests: Arc>>>, /// Receiver for incoming messages inbox: Arc>>>, } impl Director { /// Create a new director pub fn new(config: DirectorConfig) -> Self { let director_id = AgentId::new(); let router = Arc::new(A2aRouter::new(director_id.clone())); Self { config, agents: Arc::new(RwLock::new(Vec::new())), state: Arc::new(RwLock::new(ConversationState::new())), router, director_id, llm_driver: None, pending_requests: Arc::new(Mutex::new(std::collections::HashMap::new())), inbox: Arc::new(Mutex::new(None)), } } /// Create director with existing router pub fn with_router(config: DirectorConfig, router: Arc) -> Self { let director_id = AgentId::new(); Self { config, agents: Arc::new(RwLock::new(Vec::new())), state: Arc::new(RwLock::new(ConversationState::new())), router, director_id, llm_driver: None, pending_requests: Arc::new(Mutex::new(std::collections::HashMap::new())), inbox: Arc::new(Mutex::new(None)), } } /// Initialize the director's inbox (must be called after creation) pub async fn initialize(&self) -> Result<()> { let profile = A2aAgentProfile { id: self.director_id.clone(), name: "Director".to_string(), description: "Multi-agent conversation orchestrator".to_string(), capabilities: vec![A2aCapability { name: "orchestration".to_string(), description: "Multi-agent conversation management".to_string(), input_schema: None, output_schema: None, requires_approval: false, version: "1.0.0".to_string(), tags: vec!["orchestration".to_string()], }], protocols: vec!["a2a".to_string()], role: "orchestrator".to_string(), priority: 10, metadata: Default::default(), groups: vec![], last_seen: 0, }; let rx = self.router.register_agent(profile).await; *self.inbox.lock().await = Some(rx); Ok(()) } /// Set LLM driver for intelligent scheduling pub fn with_llm_driver(mut self, driver: Arc) -> Self { self.llm_driver = Some(driver); self } /// Set LLM driver (mutable) pub fn set_llm_driver(&mut self, driver: Arc) { self.llm_driver = Some(driver); } /// Register an agent pub async fn register_agent(&self, agent: DirectorAgent) { let mut agents = self.agents.write().await; agents.push(agent); // Sort by priority (descending) agents.sort_by(|a, b| b.priority.cmp(&a.priority)); } /// Remove an agent pub async fn remove_agent(&self, agent_id: &AgentId) { let mut agents = self.agents.write().await; agents.retain(|a| &a.id != agent_id); } /// Get all registered agents pub async fn get_agents(&self) -> Vec { self.agents.read().await.clone() } /// Get active agents sorted by priority pub async fn get_active_agents(&self) -> Vec { self.agents .read() .await .iter() .filter(|a| a.active) .cloned() .collect() } /// Start a new conversation pub async fn start_conversation(&self, topic: Option) { let mut state = self.state.write().await; state.turn = 0; state.current_speaker = None; state.history.clear(); state.consecutive_turns = 0; state.active = true; state.topic = topic; } /// End the conversation pub async fn end_conversation(&self) { let mut state = self.state.write().await; state.active = false; } /// Get current conversation state pub async fn get_state(&self) -> ConversationState { self.state.read().await.clone() } /// Select the next speaker based on strategy pub async fn select_next_speaker(&self) -> Option { let agents = self.get_active_agents().await; let state = self.state.read().await; if agents.is_empty() || state.turn >= self.config.max_turns { return None; } match self.config.strategy { ScheduleStrategy::RoundRobin => { // Round-robin through active agents let idx = state.turn % agents.len(); Some(agents[idx].clone()) } ScheduleStrategy::Priority => { // Select highest priority agent that hasn't exceeded consecutive limit for agent in &agents { if !state.is_over_consecutive_limit(&agent.id, agent.max_consecutive_turns) { return Some(agent.clone()); } } // If all exceeded, pick the highest priority anyway agents.first().cloned() } ScheduleStrategy::Random => { // Random selection use std::time::{SystemTime, UNIX_EPOCH}; let now = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap() .as_nanos(); let idx = (now as usize) % agents.len(); Some(agents[idx].clone()) } ScheduleStrategy::LlmDecision => { // LLM-based decision making self.select_speaker_with_llm(&agents, &state).await .or_else(|| agents.first().cloned()) } ScheduleStrategy::Manual => { // External controller decides None } } } /// Use LLM to select the next speaker async fn select_speaker_with_llm( &self, agents: &[DirectorAgent], state: &ConversationState, ) -> Option { let driver = self.llm_driver.as_ref()?; // Build context for LLM decision let agent_descriptions: String = agents .iter() .enumerate() .map(|(i, a)| format!("{}. {} ({}) - {}", i + 1, a.name, a.role.as_str(), a.persona)) .collect::>() .join("\n"); let recent_history: String = state .get_recent_history(5) .iter() .map(|(id, msg)| { let agent = agents.iter().find(|a| &a.id == id); let name = agent.map(|a| a.name.as_str()).unwrap_or("Unknown"); format!("- {}: {}", name, msg) }) .collect::>() .join("\n"); let topic = state.topic.as_deref().unwrap_or("General discussion"); let prompt = format!( r#"You are a conversation director. Select the best agent to speak next. Topic: {} Available Agents: {} Recent Conversation: {} Current turn: {} Last speaker: {} Instructions: 1. Consider the conversation flow and topic 2. Choose the agent who should speak next to advance the conversation 3. Avoid having the same agent speak too many times consecutively 4. Consider which role would be most valuable at this point Respond with ONLY the number (1-{}) of the agent who should speak next. No explanation."#, topic, agent_descriptions, recent_history, state.turn, state.current_speaker .and_then(|id| agents.iter().find(|a| a.id == id)) .map(|a| &a.name) .unwrap_or(&"None".to_string()), agents.len() ); let request = CompletionRequest { model: "default".to_string(), system: Some("You are a conversation director. You respond with only a single number.".to_string()), messages: vec![zclaw_types::Message::User { content: prompt }], tools: vec![], max_tokens: Some(10), temperature: Some(0.3), stop: vec![], stream: false, thinking_enabled: false, reasoning_effort: None, plan_mode: false, }; match driver.complete(request).await { Ok(response) => { // Extract text from response let text = response.content.iter() .filter_map(|block| match block { zclaw_runtime::ContentBlock::Text { text } => Some(text.clone()), _ => None, }) .collect::>() .join(""); // Parse the number if let Ok(idx) = text.trim().parse::() { if idx >= 1 && idx <= agents.len() { return Some(agents[idx - 1].clone()); } } // Fallback to first agent agents.first().cloned() } Err(e) => { tracing::warn!("LLM speaker selection failed: {}", e); agents.first().cloned() } } } /// Send message to selected agent and wait for response pub async fn send_to_agent( &self, agent: &DirectorAgent, message: String, ) -> Result { // Create a response channel for this request let (_response_tx, mut _response_rx) = mpsc::channel::(1); let envelope = A2aEnvelope::new( self.director_id.clone(), A2aRecipient::Direct { agent_id: agent.id.clone() }, A2aMessageType::Request, serde_json::json!({ "message": message, "persona": agent.persona.clone(), "role": agent.role.clone(), }), ); // Store the request ID with its response channel let request_id = envelope.id.clone(); { let mut pending = self.pending_requests.lock().await; pending.insert(request_id.clone(), _response_tx); } // Send the request self.router.route(envelope).await?; // Wait for response with timeout let timeout_duration = std::time::Duration::from_secs(self.config.response_timeout); let request_id_clone = request_id.clone(); let response = tokio::time::timeout(timeout_duration, async { // Poll the inbox for responses let mut inbox_guard = self.inbox.lock().await; if let Some(ref mut rx) = *inbox_guard { while let Some(msg) = rx.recv().await { // Check if this is a response to our request if msg.message_type == A2aMessageType::Response { if let Some(ref reply_to) = msg.reply_to { if reply_to == &request_id_clone { // Found our response return Some(msg); } } } // Not our response, continue waiting // (In a real implementation, we'd re-queue non-matching messages) } } None }).await; // Clean up pending request { let mut pending = self.pending_requests.lock().await; pending.remove(&request_id); } match response { Ok(Some(envelope)) => { // Extract response text from payload let response_text = envelope.payload .get("response") .and_then(|v: &serde_json::Value| v.as_str()) .unwrap_or(&format!("[{}] Response from {}", agent.role.as_str(), agent.name)) .to_string(); Ok(response_text) } Ok(None) => { Err(ZclawError::Timeout("No response received".into())) } Err(_) => { Err(ZclawError::Timeout(format!( "Agent {} did not respond within {} seconds", agent.name, self.config.response_timeout ))) } } } /// Broadcast message to all agents pub async fn broadcast(&self, message: String) -> Result<()> { let envelope = A2aEnvelope::new( self.director_id, A2aRecipient::Broadcast, A2aMessageType::Notification, serde_json::json!({ "message": message }), ); self.router.route(envelope).await } /// Run one turn of the conversation pub async fn run_turn(&self, input: Option) -> Result> { let state = self.state.read().await; if !state.active { return Err(ZclawError::InvalidInput("Conversation not active".into())); } drop(state); // Select next speaker let speaker = self.select_next_speaker().await; if let Some(ref agent) = speaker { // Build context from recent history let state = self.state.read().await; let context = Self::build_context(&state, &input); // Send message to agent let response = self.send_to_agent(agent, context).await?; // Update state let mut state = self.state.write().await; let summary = if response.len() > 100 { format!("{}...", &response[..100]) } else { response }; state.record_turn(agent.id, summary); } Ok(speaker) } /// Build context string for agent fn build_context(state: &ConversationState, input: &Option) -> String { let mut context = String::new(); if let Some(ref topic) = state.topic { context.push_str(&format!("Topic: {}\n\n", topic)); } if let Some(ref user_input) = input { context.push_str(&format!("User: {}\n\n", user_input)); } // Add recent history if !state.history.is_empty() { context.push_str("Recent conversation:\n"); for (agent_id, summary) in state.get_recent_history(5) { context.push_str(&format!("- {}: {}\n", agent_id, summary)); } } context } /// Run full conversation until complete pub async fn run_conversation( &self, topic: String, initial_input: Option, ) -> Result> { self.start_conversation(Some(topic.clone())).await; let mut input = initial_input; let mut results = Vec::new(); loop { let state = self.state.read().await; // Check termination conditions if state.turn >= self.config.max_turns { break; } if !state.active { break; } drop(state); // Run one turn match self.run_turn(input.take()).await { Ok(Some(_agent)) => { let state = self.state.read().await; if let Some((agent_id, summary)) = state.history.last() { results.push((*agent_id, summary.clone())); } } Ok(None) => { // Manual mode or no speaker selected break; } Err(e) => { tracing::error!("Turn error: {}", e); break; } } // In a real implementation, we would wait for user input here // if config.include_user is true } self.end_conversation().await; Ok(results) } /// Get the director's agent ID pub fn director_id(&self) -> &AgentId { &self.director_id } } impl AgentRole { /// Get role as string pub fn as_str(&self) -> &'static str { match self { AgentRole::Teacher => "teacher", AgentRole::Assistant => "assistant", AgentRole::Student => "student", AgentRole::Moderator => "moderator", AgentRole::Expert => "expert", AgentRole::Observer => "observer", } } /// Parse role from string pub fn from_str(s: &str) -> Option { match s.to_lowercase().as_str() { "teacher" | "instructor" => Some(AgentRole::Teacher), "assistant" | "ta" => Some(AgentRole::Assistant), "student" => Some(AgentRole::Student), "moderator" | "facilitator" => Some(AgentRole::Moderator), "expert" | "consultant" => Some(AgentRole::Expert), "observer" => Some(AgentRole::Observer), _ => None, } } } /// Builder for creating director configurations pub struct DirectorBuilder { config: DirectorConfig, agents: Vec, } impl DirectorBuilder { /// Create a new builder pub fn new() -> Self { Self { config: DirectorConfig::default(), agents: Vec::new(), } } /// Set scheduling strategy pub fn strategy(mut self, strategy: ScheduleStrategy) -> Self { self.config.strategy = strategy; self } /// Set max turns pub fn max_turns(mut self, max_turns: usize) -> Self { self.config.max_turns = max_turns; self } /// Include user in conversation pub fn include_user(mut self, include: bool) -> Self { self.config.include_user = include; self } /// Add a teacher agent pub fn teacher(mut self, id: AgentId, name: impl Into, persona: impl Into) -> Self { let mut agent = DirectorAgent::new(id, name, AgentRole::Teacher, persona); agent.priority = 10; self.agents.push(agent); self } /// Add an assistant agent pub fn assistant(mut self, id: AgentId, name: impl Into, persona: impl Into) -> Self { let mut agent = DirectorAgent::new(id, name, AgentRole::Assistant, persona); agent.priority = 7; self.agents.push(agent); self } /// Add a student agent pub fn student(mut self, id: AgentId, name: impl Into, persona: impl Into) -> Self { let mut agent = DirectorAgent::new(id, name, AgentRole::Student, persona); agent.priority = 5; self.agents.push(agent); self } /// Add a custom agent pub fn agent(mut self, agent: DirectorAgent) -> Self { self.agents.push(agent); self } /// Build the director pub async fn build(self) -> Director { let director = Director::new(self.config); for agent in self.agents { director.register_agent(agent).await; } director } } impl Default for DirectorBuilder { fn default() -> Self { Self::new() } } #[cfg(test)] mod tests { use super::*; #[tokio::test] async fn test_director_creation() { let director = Director::new(DirectorConfig::default()); let agents = director.get_agents().await; assert!(agents.is_empty()); } #[tokio::test] async fn test_register_agents() { let director = Director::new(DirectorConfig::default()); director.register_agent(DirectorAgent::new( AgentId::new(), "Teacher", AgentRole::Teacher, "You are a helpful teacher.", )).await; director.register_agent(DirectorAgent::new( AgentId::new(), "Student", AgentRole::Student, "You are a curious student.", )).await; let agents = director.get_agents().await; assert_eq!(agents.len(), 2); // Teacher should be first (higher priority) assert_eq!(agents[0].role, AgentRole::Teacher); } #[tokio::test] async fn test_conversation_state() { let mut state = ConversationState::new(); assert_eq!(state.turn, 0); let agent1 = AgentId::new(); let agent2 = AgentId::new(); state.record_turn(agent1, "Hello".to_string()); assert_eq!(state.turn, 1); assert_eq!(state.consecutive_turns, 1); state.record_turn(agent1, "World".to_string()); assert_eq!(state.turn, 2); assert_eq!(state.consecutive_turns, 2); state.record_turn(agent2, "Goodbye".to_string()); assert_eq!(state.turn, 3); assert_eq!(state.consecutive_turns, 1); assert_eq!(state.current_speaker, Some(agent2)); } #[tokio::test] async fn test_select_next_speaker_priority() { let config = DirectorConfig { strategy: ScheduleStrategy::Priority, ..Default::default() }; let director = Director::new(config); let teacher_id = AgentId::new(); let student_id = AgentId::new(); director.register_agent(DirectorAgent::new( teacher_id, "Teacher", AgentRole::Teacher, "Teaching", )).await; director.register_agent(DirectorAgent::new( student_id, "Student", AgentRole::Student, "Learning", )).await; let speaker = director.select_next_speaker().await; assert!(speaker.is_some()); assert_eq!(speaker.unwrap().role, AgentRole::Teacher); } #[tokio::test] async fn test_director_builder() { let director = DirectorBuilder::new() .strategy(ScheduleStrategy::RoundRobin) .max_turns(10) .teacher(AgentId::new(), "AI Teacher", "You teach students.") .student(AgentId::new(), "Curious Student", "You ask questions.") .build() .await; let agents = director.get_agents().await; assert_eq!(agents.len(), 2); let state = director.get_state().await; assert_eq!(state.turn, 0); } #[test] fn test_agent_role_priority() { assert_eq!(AgentRole::Teacher.default_priority(), 10); assert_eq!(AgentRole::Assistant.default_priority(), 7); assert_eq!(AgentRole::Student.default_priority(), 5); assert_eq!(AgentRole::Observer.default_priority(), 0); } #[test] fn test_agent_role_parse() { assert_eq!(AgentRole::from_str("teacher"), Some(AgentRole::Teacher)); assert_eq!(AgentRole::from_str("STUDENT"), Some(AgentRole::Student)); assert_eq!(AgentRole::from_str("unknown"), None); } }