Use command line to run sub agent and sub recipe (in sequence or parallel)

crates/goose/src/agents/agent.rs

@@ -9,6 +9,9 @@ use futures::{stream, FutureExt, Stream, StreamExt, TryStreamExt};
 use mcp_core::protocol::JsonRpcMessage;
 
 use crate::agents::final_output_tool::{FINAL_OUTPUT_CONTINUATION_MESSAGE, FINAL_OUTPUT_TOOL_NAME};
+use crate::agents::sub_recipe_execution_tool::sub_recipe_execute_task_tool::{
+    self, SUB_RECIPE_EXECUTE_TASK_TOOL_NAME,
+};
 use crate::agents::sub_recipe_manager::SubRecipeManager;
 use crate::config::{Config, ExtensionConfigManager, PermissionManager};
 use crate::message::Message;
@@ -286,11 +289,12 @@ impl Agent {
 
         let extension_manager = self.extension_manager.read().await;
         let sub_recipe_manager = self.sub_recipe_manager.lock().await;
-
         let result: ToolCallResult = if sub_recipe_manager.is_sub_recipe_tool(&tool_call.name) {
             sub_recipe_manager
                 .dispatch_sub_recipe_tool_call(&tool_call.name, tool_call.arguments.clone())
                 .await
+        } else if tool_call.name == SUB_RECIPE_EXECUTE_TASK_TOOL_NAME {
+            sub_recipe_execute_task_tool::run_tasks(tool_call.arguments.clone()).await
         } else if tool_call.name == PLATFORM_READ_RESOURCE_TOOL_NAME {
             // Check if the tool is read_resource and handle it separately
             ToolCallResult::from(
@@ -574,6 +578,8 @@ impl Agent {
             if let Some(final_output_tool) = self.final_output_tool.lock().await.as_ref() {
                 prefixed_tools.push(final_output_tool.tool());
             }
+            prefixed_tools
+                .push(sub_recipe_execute_task_tool::create_sub_recipe_execute_task_tool());
         }
 
         prefixed_tools

crates/goose/src/agents/mod.rs

@@ -11,6 +11,7 @@ mod reply_parts;
 mod router_tool_selector;
 mod router_tools;
 mod schedule_tool;
+pub mod sub_recipe_execution_tool;
 pub mod sub_recipe_manager;
 pub mod subagent;
 pub mod subagent_handler;

crates/goose/src/agents/recipe_tools/sub_recipe_tools.rs

@@ -3,25 +3,20 @@ use std::{collections::HashMap, fs};
 use anyhow::Result;
 use mcp_core::tool::{Tool, ToolAnnotations};
 use serde_json::{json, Map, Value};
-use tokio::io::{AsyncBufReadExt, BufReader};
-use tokio::process::Command;
 
+use crate::agents::sub_recipe_execution_tool::lib::Task;
 use crate::recipe::{Recipe, RecipeParameter, RecipeParameterRequirement, SubRecipe};
 
-pub const SUB_RECIPE_TOOL_NAME_PREFIX: &str = "subrecipe__run_";
+pub const SUB_RECIPE_TASK_TOOL_NAME_PREFIX: &str = "subrecipe__create_task";
 
-pub fn create_sub_recipe_tool(sub_recipe: &SubRecipe) -> Tool {
+pub fn create_sub_recipe_task_tool(sub_recipe: &SubRecipe) -> Tool {
     let input_schema = get_input_schema(sub_recipe).unwrap();
     Tool::new(
-        format!("{}_{}", SUB_RECIPE_TOOL_NAME_PREFIX, sub_recipe.name),
-        "Run a sub recipe.
-        Use this tool when you need to run a sub-recipe.
-        The sub recipe will be run with the provided parameters 
-        and return the output of the sub recipe."
-            .to_string(),
+        format!("{}_{}", SUB_RECIPE_TASK_TOOL_NAME_PREFIX, sub_recipe.name),
+        "Before running this sub recipe, you should first create a task with this tool and then pass the task to the task executor".to_string(),
         input_schema,
         Some(ToolAnnotations {
-            title: Some(format!("run sub recipe {}", sub_recipe.name)),
+            title: Some(format!("create sub recipe task {}", sub_recipe.name)),
             read_only_hint: false,
             destructive_hint: true,
             idempotent_hint: false,
@@ -99,68 +94,23 @@ fn prepare_command_params(
     Ok(sub_recipe_params)
 }
 
-pub async fn run_sub_recipe(sub_recipe: &SubRecipe, params: Value) -> Result<String> {
+pub async fn create_sub_recipe_task(sub_recipe: &SubRecipe, params: Value) -> Result<String> {
     let command_params = prepare_command_params(sub_recipe, params)?;
-
-    let mut command = Command::new("goose");
-    command.arg("run").arg("--recipe").arg(&sub_recipe.path);
-
-    for (key, value) in command_params {
-        command.arg("--params").arg(format!("{}={}", key, value));
-    }
-
-    command.stdout(std::process::Stdio::piped());
-    command.stderr(std::process::Stdio::piped());
-
-    let mut child = command
-        .spawn()
-        .map_err(|e| anyhow::anyhow!("Failed to spawn: {}", e))?;
-
-    let stdout = child.stdout.take().expect("Failed to capture stdout");
-    let stderr = child.stderr.take().expect("Failed to capture stderr");
-
-    let mut stdout_reader = BufReader::new(stdout).lines();
-    let mut stderr_reader = BufReader::new(stderr).lines();
-    let stdout_sub_recipe_name = sub_recipe.name.clone();
-    let stderr_sub_recipe_name = sub_recipe.name.clone();
-
-    // Spawn background tasks to read from stdout and stderr
-    let stdout_task = tokio::spawn(async move {
-        let mut buffer = String::new();
-        while let Ok(Some(line)) = stdout_reader.next_line().await {
-            println!("[sub-recipe {}] {}", stdout_sub_recipe_name, line);
-            buffer.push_str(&line);
-            buffer.push('\n');
+    let payload = json!({
+        "sub_recipe": {
+            "name": sub_recipe.name.clone(),
+            "command_parameters": command_params,
+            "recipe_path": sub_recipe.path.clone(),
         }
-        buffer
     });
-
-    let stderr_task = tokio::spawn(async move {
-        let mut buffer = String::new();
-        while let Ok(Some(line)) = stderr_reader.next_line().await {
-            eprintln!(
-                "[stderr for sub-recipe {}] {}",
-                stderr_sub_recipe_name, line
-            );
-            buffer.push_str(&line);
-            buffer.push('\n');
-        }
-        buffer
-    });
-
-    let status = child
-        .wait()
-        .await
-        .map_err(|e| anyhow::anyhow!("Failed to wait for process: {}", e))?;
-
-    let stdout_output = stdout_task.await.unwrap();
-    let stderr_output = stderr_task.await.unwrap();
-
-    if status.success() {
-        Ok(stdout_output)
-    } else {
-        Err(anyhow::anyhow!("Command failed:\n{}", stderr_output))
-    }
+    let task = Task {
+        id: uuid::Uuid::new_v4().to_string(),
+        task_type: "sub_recipe".to_string(),
+        payload,
+    };
+    let task_json = serde_json::to_string(&task)
+        .map_err(|e| anyhow::anyhow!("Failed to serialize Task: {}", e))?;
+    Ok(task_json)
 }
 
 #[cfg(test)]

crates/goose/src/agents/sub_recipe_execution_tool/executor.rs

@@ -0,0 +1,103 @@
+use std::sync::atomic::{AtomicBool, AtomicUsize};
+use std::sync::Arc;
+use tokio::sync::mpsc;
+use tokio::time::Instant;
+
+use crate::agents::sub_recipe_execution_tool::lib::{
+    Config, ExecutionResponse, ExecutionStats, Task, TaskResult,
+};
+use crate::agents::sub_recipe_execution_tool::tasks::process_task;
+use crate::agents::sub_recipe_execution_tool::workers::{run_scaler, spawn_worker, SharedState};
+
+pub async fn execute_single_task(task: &Task, config: Config) -> ExecutionResponse {
+    let start_time = Instant::now();
+    let result = process_task(task, config.timeout_seconds).await;
+
+    let execution_time = start_time.elapsed().as_millis();
+    let completed = if result.status == "success" { 1 } else { 0 };
+    let failed = if result.status == "failed" { 1 } else { 0 };
+
+    ExecutionResponse {
+        status: "completed".to_string(),
+        results: vec![result],
+        stats: ExecutionStats {
+            total_tasks: 1,
+            completed,
+            failed,
+            execution_time_ms: execution_time,
+        },
+    }
+}
+
+// Main parallel execution function
+pub async fn parallel_execute(tasks: Vec<Task>, config: Config) -> ExecutionResponse {
+    let start_time = Instant::now();
+    let task_count = tasks.len();
+
+    // Create channels
+    let (task_tx, task_rx) = mpsc::channel::<Task>(task_count);
+    let (result_tx, mut result_rx) = mpsc::channel::<TaskResult>(task_count);
+
+    // Initialize shared state
+    let shared_state = Arc::new(SharedState {
+        task_receiver: Arc::new(tokio::sync::Mutex::new(task_rx)),
+        result_sender: result_tx,
+        active_workers: Arc::new(AtomicUsize::new(0)),
+        should_stop: Arc::new(AtomicBool::new(false)),
+        completed_tasks: Arc::new(AtomicUsize::new(0)),
+    });
+
+    // Send all tasks to the queue
+    for task in tasks.clone() {
+        let _ = task_tx.send(task).await;
+    }
+    // Close sender so workers know when queue is empty
+    drop(task_tx);
+
+    // Start initial workers
+    let mut worker_handles = Vec::new();
+    for i in 0..config.initial_workers {
+        let handle = spawn_worker(shared_state.clone(), i, config.timeout_seconds);
+        worker_handles.push(handle);
+    }
+
+    // Start the scaler
+    let scaler_state = shared_state.clone();
+    let scaler_handle = tokio::spawn(async move {
+        run_scaler(
+            scaler_state,
+            task_count,
+            config.max_workers,
+            config.timeout_seconds,
+        )
+        .await;
+    });
+
+    // Collect results
+    let mut results = Vec::new();
+    while let Some(result) = result_rx.recv().await {
+        results.push(result);
+        if results.len() >= task_count {
+            break;
+        }
+    }
+
+    // Wait for scaler to finish
+    let _ = scaler_handle.await;
+
+    // Calculate stats
+    let execution_time = start_time.elapsed().as_millis();
+    let completed = results.iter().filter(|r| r.status == "success").count();
+    let failed = results.iter().filter(|r| r.status == "failed").count();
+
+    ExecutionResponse {
+        status: "completed".to_string(),
+        results,
+        stats: ExecutionStats {
+            total_tasks: task_count,
+            completed,
+            failed,
+            execution_time_ms: execution_time,
+        },
+    }
+}

crates/goose/src/agents/sub_recipe_execution_tool/lib.rs

@@ -0,0 +1,38 @@
+use crate::agents::sub_recipe_execution_tool::executor::execute_single_task;
+pub use crate::agents::sub_recipe_execution_tool::executor::parallel_execute;
+pub use crate::agents::sub_recipe_execution_tool::types::{
+    Config, ExecutionResponse, ExecutionStats, Task, TaskResult,
+};
+
+use serde_json::Value;
+
+pub async fn execute_tasks(input: Value, execution_mode: &str) -> Result<Value, String> {
+    let tasks: Vec<Task> =
+        serde_json::from_value(input.get("tasks").ok_or("Missing tasks field")?.clone())
+            .map_err(|e| format!("Failed to parse tasks: {}", e))?;
+
+    let config: Config = if let Some(config_value) = input.get("config") {
+        serde_json::from_value(config_value.clone())
+            .map_err(|e| format!("Failed to parse config: {}", e))?
+    } else {
+        Config::default()
+    };
+    let task_count = tasks.len();
+    match execution_mode {
+        "sequential" => {
+            if task_count == 1 {
+                let response = execute_single_task(&tasks[0], config).await;
+                serde_json::to_value(response)
+                    .map_err(|e| format!("Failed to serialize response: {}", e))
+            } else {
+                Err("Sequential execution mode requires exactly one task".to_string())
+            }
+        }
+        "parallel" => {
+            let response = parallel_execute(tasks, config).await;
+            serde_json::to_value(response)
+                .map_err(|e| format!("Failed to serialize response: {}", e))
+        }
+        _ => Err("Invalid execution mode".to_string()),
+    }
+}

crates/goose/src/agents/sub_recipe_execution_tool/mod.rs

@@ -0,0 +1,6 @@
+mod executor;
+pub mod lib;
+pub mod sub_recipe_execute_task_tool;
+mod tasks;
+mod types;
+mod workers;