refactor: unify subagent and subrecipe tools into single tool (#5893)

This commit is contained in:
tlongwell-block
2025-12-13 13:50:20 -05:00
committed by GitHub
parent e6e5ba52ab
commit a131b08817
32 changed files with 796 additions and 3443 deletions
+32 -59
View File
@@ -273,7 +273,7 @@ fn render_tool_request(req: &ToolRequest, theme: Theme, debug: bool) {
Ok(call) => match call.name.to_string().as_str() {
"developer__text_editor" => render_text_editor_request(call, debug),
"developer__shell" => render_shell_request(call, debug),
"dynamic_task__create_task" => render_dynamic_task_request(call, debug),
"subagent" => render_subagent_request(call, debug),
"todo__write" => render_todo_request(call, debug),
_ => render_default_request(call, debug),
},
@@ -447,69 +447,42 @@ fn render_shell_request(call: &CallToolRequestParam, debug: bool) {
println!();
}
fn render_dynamic_task_request(call: &CallToolRequestParam, debug: bool) {
fn render_subagent_request(call: &CallToolRequestParam, debug: bool) {
print_tool_header(call);
// Print task_parameters array
if let Some(task_parameters) = call
.arguments
.as_ref()
.and_then(|args| args.get("task_parameters"))
.and_then(|v| match v {
Value::Array(arr) => Some(arr),
_ => None,
})
{
println!("{}:", style("task_parameters").dim());
for task_param in task_parameters.iter() {
println!(" -");
if let Some(args) = &call.arguments {
if let Some(Value::String(subrecipe)) = args.get("subrecipe") {
println!("{}: {}", style("subrecipe").dim(), style(subrecipe).cyan());
}
if let Some(param_obj) = task_param.as_object() {
for (key, value) in param_obj {
match value {
Value::String(s) => {
// For strings, print the full content without truncation
println!(" {}: {}", style(key).dim(), style(s).green());
}
Value::Array(arr) => {
// For arrays, print each item on its own line
println!(" {}:", style(key).dim());
for item in arr {
if let Value::String(s) = item {
println!(" - {}", style(s).green());
} else if let Value::Object(_) = item {
// For objects in arrays, print them with indentation
print!(" - ");
if let Value::Object(obj) = item {
print_params(&Some(obj.clone()), 3, debug);
}
} else {
println!(
" - {}",
style(format!("{}", item)).green()
);
}
}
}
Value::Object(_) => {
// For objects, print them with proper indentation
println!(" {}:", style(key).dim());
if let Value::Object(obj) = value {
print_params(&Some(obj.clone()), 2, debug);
}
}
_ => {
// For other types (numbers, booleans, null)
println!(
" {}: {}",
style(key).dim(),
style(format!("{}", value)).green()
);
}
}
}
if let Some(Value::String(instructions)) = args.get("instructions") {
let display = if instructions.len() > 100 && !debug {
safe_truncate(instructions, 100)
} else {
instructions.clone()
};
println!(
"{}: {}",
style("instructions").dim(),
style(display).green()
);
}
if let Some(Value::Object(params)) = args.get("parameters") {
println!("{}:", style("parameters").dim());
print_params(&Some(params.clone()), 1, debug);
}
let skip_keys = ["subrecipe", "instructions", "parameters"];
let mut other_args = serde_json::Map::new();
for (k, v) in args {
if !skip_keys.contains(&k.as_str()) {
other_args.insert(k.clone(), v.clone());
}
}
if !other_args.is_empty() {
print_params(&Some(other_args), 0, debug);
}
}
println!();
+25 -114
View File
@@ -18,18 +18,12 @@ use crate::agents::extension_manager_extension::MANAGE_EXTENSIONS_TOOL_NAME_COMP
use crate::agents::final_output_tool::{FINAL_OUTPUT_CONTINUATION_MESSAGE, FINAL_OUTPUT_TOOL_NAME};
use crate::agents::platform_tools::PLATFORM_MANAGE_SCHEDULE_TOOL_NAME;
use crate::agents::prompt_manager::PromptManager;
use crate::agents::recipe_tools::dynamic_task_tools::{
create_dynamic_task, create_dynamic_task_tool, DYNAMIC_TASK_TOOL_NAME_PREFIX,
};
use crate::agents::retry::{RetryManager, RetryResult};
use crate::agents::router_tools::ROUTER_LLM_SEARCH_TOOL_NAME;
use crate::agents::sub_recipe_manager::SubRecipeManager;
use crate::agents::subagent_execution_tool::lib::ExecutionMode;
use crate::agents::subagent_execution_tool::subagent_execute_task_tool::{
self, SUBAGENT_EXECUTE_TASK_TOOL_NAME,
};
use crate::agents::subagent_execution_tool::tasks_manager::TasksManager;
use crate::agents::subagent_task_config::TaskConfig;
use crate::agents::subagent_tool::{
create_subagent_tool, handle_subagent_tool, SUBAGENT_TOOL_NAME,
};
use crate::agents::tool_route_manager::ToolRouteManager;
use crate::agents::tool_router_index_manager::ToolRouterIndexManager;
use crate::agents::types::SessionConfig;
@@ -92,8 +86,7 @@ pub struct Agent {
pub(super) provider: SharedProvider,
pub extension_manager: Arc<ExtensionManager>,
pub(super) sub_recipe_manager: Mutex<SubRecipeManager>,
pub(super) tasks_manager: TasksManager,
pub(super) sub_recipes: Mutex<HashMap<String, SubRecipe>>,
pub(super) final_output_tool: Arc<Mutex<Option<FinalOutputTool>>>,
pub(super) frontend_tools: Mutex<HashMap<String, FrontendTool>>,
pub(super) frontend_instructions: Mutex<Option<String>>,
@@ -168,8 +161,7 @@ impl Agent {
Self {
provider: provider.clone(),
extension_manager: Arc::new(ExtensionManager::new(provider.clone())),
sub_recipe_manager: Mutex::new(SubRecipeManager::new()),
tasks_manager: TasksManager::new(),
sub_recipes: Mutex::new(HashMap::new()),
final_output_tool: Arc::new(Mutex::new(None)),
frontend_tools: Mutex::new(HashMap::new()),
frontend_instructions: Mutex::new(None),
@@ -407,9 +399,11 @@ impl Agent {
self.extend_system_prompt(final_output_system_prompt).await;
}
pub async fn add_sub_recipes(&self, sub_recipes: Vec<SubRecipe>) {
let mut sub_recipe_manager = self.sub_recipe_manager.lock().await;
sub_recipe_manager.add_sub_recipe_tools(sub_recipes);
pub async fn add_sub_recipes(&self, sub_recipes_to_add: Vec<SubRecipe>) {
let mut sub_recipes = self.sub_recipes.lock().await;
for sr in sub_recipes_to_add {
sub_recipes.insert(sr.name.clone(), sr);
}
}
pub async fn apply_recipe_components(
@@ -438,9 +432,9 @@ impl Agent {
cancellation_token: Option<CancellationToken>,
session: &Session,
) -> (String, Result<ToolCallResult, ErrorData>) {
// Prevent subagents from creating other subagents
if session.session_type == crate::session::SessionType::SubAgent
&& (tool_call.name == DYNAMIC_TASK_TOOL_NAME_PREFIX
|| tool_call.name == SUBAGENT_EXECUTE_TASK_TOOL_NAME)
&& tool_call.name == SUBAGENT_TOOL_NAME
{
return (
request_id,
@@ -486,27 +480,7 @@ impl Agent {
}
debug!("WAITING_TOOL_START: {}", tool_call.name);
let result: ToolCallResult = if self
.sub_recipe_manager
.lock()
.await
.is_sub_recipe_tool(&tool_call.name)
{
let sub_recipe_manager = self.sub_recipe_manager.lock().await;
let arguments = tool_call
.arguments
.clone()
.map(Value::Object)
.unwrap_or(Value::Object(serde_json::Map::new()));
sub_recipe_manager
.dispatch_sub_recipe_tool_call(
&tool_call.name,
arguments,
&self.tasks_manager,
&session.working_dir,
)
.await
} else if tool_call.name == SUBAGENT_EXECUTE_TASK_TOOL_NAME {
let result: ToolCallResult = if tool_call.name == SUBAGENT_TOOL_NAME {
let provider = match self.provider().await {
Ok(p) => p,
Err(_) => {
@@ -521,84 +495,24 @@ impl Agent {
}
};
// Get extensions from the agent's runtime state rather than global config
// This ensures subagents inherit extensions that were dynamically enabled by the parent
let extensions = self.get_extension_configs().await;
let task_config =
TaskConfig::new(provider, &session.id, &session.working_dir, extensions);
let sub_recipes = self.sub_recipes.lock().await.clone();
let arguments = match tool_call.arguments.clone() {
Some(args) => Value::Object(args),
None => {
return (
request_id,
Err(ErrorData::new(
ErrorCode::INVALID_PARAMS,
"Tool call arguments are required".to_string(),
None,
)),
);
}
};
let task_ids: Vec<String> = match arguments.get("task_ids") {
Some(v) => match serde_json::from_value(v.clone()) {
Ok(ids) => ids,
Err(_) => {
return (
request_id,
Err(ErrorData::new(
ErrorCode::INVALID_PARAMS,
"Invalid task_ids format".to_string(),
None,
)),
);
}
},
None => {
return (
request_id,
Err(ErrorData::new(
ErrorCode::INVALID_PARAMS,
"task_ids parameter is required".to_string(),
None,
)),
);
}
};
let execution_mode = arguments
.get("execution_mode")
.and_then(|v| serde_json::from_value::<ExecutionMode>(v.clone()).ok())
.unwrap_or(ExecutionMode::Sequential);
subagent_execute_task_tool::run_tasks(
task_ids,
execution_mode,
task_config,
&self.tasks_manager,
cancellation_token,
)
.await
} else if tool_call.name == DYNAMIC_TASK_TOOL_NAME_PREFIX {
// Get loaded extensions for shortname resolution
let loaded_extensions = self
.extension_manager
.list_extensions()
.await
.unwrap_or_default();
let arguments = tool_call
.arguments
.clone()
.map(Value::Object)
.unwrap_or(Value::Object(serde_json::Map::new()));
create_dynamic_task(
handle_subagent_tool(
arguments,
&self.tasks_manager,
loaded_extensions,
&session.working_dir,
task_config,
sub_recipes,
session.working_dir.clone(),
cancellation_token,
)
.await
} else if self.is_frontend_tool(&tool_call.name).await {
// For frontend tools, return an error indicating we need frontend execution
ToolCallResult::from(Err(ErrorData::new(
@@ -734,21 +648,18 @@ impl Agent {
.unwrap_or_default();
if extension_name.is_none() || extension_name.as_deref() == Some("platform") {
// Add platform tools
// TODO: migrate the manage schedule tool as well
prefixed_tools.extend([platform_tools::manage_schedule_tool()]);
// Dynamic task tool
prefixed_tools.push(create_dynamic_task_tool());
}
if extension_name.is_none() {
let sub_recipe_manager = self.sub_recipe_manager.lock().await;
prefixed_tools.extend(sub_recipe_manager.sub_recipe_tools.values().cloned());
if let Some(final_output_tool) = self.final_output_tool.lock().await.as_ref() {
prefixed_tools.push(final_output_tool.tool());
}
prefixed_tools.push(subagent_execute_task_tool::create_subagent_execute_task_tool());
// Add the unified subagent tool
let sub_recipes = self.sub_recipes.lock().await;
let sub_recipes_vec: Vec<_> = sub_recipes.values().cloned().collect();
prefixed_tools.push(create_subagent_tool(&sub_recipes_vec));
}
prefixed_tools
+1 -2
View File
@@ -10,17 +10,16 @@ pub mod mcp_client;
pub mod moim;
pub mod platform_tools;
pub mod prompt_manager;
pub mod recipe_tools;
mod reply_parts;
pub mod retry;
mod router_tool_selector;
mod router_tools;
mod schedule_tool;
pub(crate) mod skills_extension;
pub mod sub_recipe_manager;
pub mod subagent_execution_tool;
pub mod subagent_handler;
mod subagent_task_config;
pub mod subagent_tool;
pub(crate) mod todo_extension;
mod tool_execution;
mod tool_route_manager;
+2 -2
View File
@@ -6,8 +6,8 @@ use serde_json::Value;
use std::collections::HashMap;
use crate::agents::extension::ExtensionInfo;
use crate::agents::recipe_tools::dynamic_task_tools::should_enabled_subagents;
use crate::agents::router_tools::llm_search_tool_prompt;
use crate::agents::subagent_tool::should_enable_subagents;
use crate::hints::load_hints::{load_hint_files, AGENTS_MD_FILENAME, GOOSE_HINTS_FILENAME};
use crate::{
config::{Config, GooseMode},
@@ -152,7 +152,7 @@ impl<'a> SystemPromptBuilder<'a, PromptManager> {
extension_tool_limits,
goose_mode,
is_autonomous: goose_mode == GooseMode::Auto,
enable_subagents: should_enabled_subagents(self.model_name.as_str()),
enable_subagents: should_enable_subagents(self.model_name.as_str()),
max_extensions: MAX_EXTENSIONS,
max_tools: MAX_TOOLS,
};
@@ -1,379 +0,0 @@
// =======================================
// Module: Dynamic Task Tools
// Handles creation of tasks dynamically without sub-recipes
// =======================================
use crate::agents::extension::ExtensionConfig;
use crate::agents::subagent_execution_tool::tasks_manager::TasksManager;
use crate::agents::subagent_execution_tool::{
lib::ExecutionMode,
task_types::{Task, TaskPayload},
};
use crate::agents::tool_execution::ToolCallResult;
use crate::config::GooseMode;
use crate::recipe::{Recipe, RecipeBuilder};
use crate::session::SessionManager;
use anyhow::{anyhow, Result};
use rmcp::model::{Content, ErrorCode, ErrorData, Tool, ToolAnnotations};
use rmcp::schemars::{schema_for, JsonSchema};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::borrow::Cow;
pub const DYNAMIC_TASK_TOOL_NAME_PREFIX: &str = "dynamic_task__create_task";
#[derive(Debug, Serialize, Deserialize, JsonSchema)]
pub struct CreateDynamicTaskParams {
/// Array of tasks. Each task must have either 'instructions' OR 'prompt' field (at least one is required).
#[schemars(length(min = 1))]
pub task_parameters: Vec<TaskParameter>,
/// How to execute multiple tasks (default: parallel for multiple tasks, sequential for single task)
#[serde(skip_serializing_if = "Option::is_none")]
#[schemars(with = "Option<String>")]
pub execution_mode: Option<ExecutionModeParam>,
}
/// Execution mode for tasks
#[derive(Debug, Serialize, Deserialize, JsonSchema, Clone, Copy)]
#[serde(rename_all = "lowercase")]
pub enum ExecutionModeParam {
Sequential,
Parallel,
}
impl From<ExecutionModeParam> for ExecutionMode {
fn from(mode: ExecutionModeParam) -> Self {
match mode {
ExecutionModeParam::Sequential => ExecutionMode::Sequential,
ExecutionModeParam::Parallel => ExecutionMode::Parallel,
}
}
}
type JsonObject = serde_json::Map<String, Value>;
/// Parameters for a single task
#[derive(Debug, Serialize, Deserialize, JsonSchema)]
pub struct TaskParameter {
#[serde(skip_serializing_if = "Option::is_none")]
pub instructions: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub prompt: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub title: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub description: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub extensions: Option<Vec<JsonObject>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub settings: Option<JsonObject>,
#[serde(skip_serializing_if = "Option::is_none")]
pub parameters: Option<Vec<JsonObject>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub response: Option<JsonObject>,
#[serde(skip_serializing_if = "Option::is_none")]
pub retry: Option<JsonObject>,
#[serde(skip_serializing_if = "Option::is_none")]
pub activities: Option<Vec<String>>,
/// If true, return only the last message from the subagent (default: false, returns full conversation)
#[serde(skip_serializing_if = "Option::is_none")]
pub return_last_only: Option<bool>,
}
pub fn should_enabled_subagents(model_name: &str) -> bool {
let config = crate::config::Config::global();
let is_autonomous = config.get_goose_mode().unwrap_or(GooseMode::Auto) == GooseMode::Auto;
if !is_autonomous {
return false;
}
if model_name.starts_with("gemini") {
return false;
}
true
}
pub fn create_dynamic_task_tool() -> Tool {
let schema = schema_for!(CreateDynamicTaskParams);
let schema_value =
serde_json::to_value(schema).expect("Failed to serialize CreateDynamicTaskParams schema");
let input_schema = schema_value
.as_object()
.expect("Schema should be an object")
.clone();
Tool::new(
DYNAMIC_TASK_TOOL_NAME_PREFIX.to_string(),
"Create tasks with instructions or prompt. For simple tasks, only include the instructions field. Extensions control: omit field = use all current extensions; empty array [] = no extensions; array with names = only those extensions. Specify extensions as shortnames (the prefixes for your tools). Specify return_last_only as true and have your subagent summarize its work in its last message to conserve your own context. Optional: title, description, extensions, settings, retry, response schema, activities. Arrays for multiple tasks.".to_string(),
input_schema,
).annotate(ToolAnnotations {
title: Some("Create Dynamic Tasks".to_string()),
read_only_hint: Some(false),
destructive_hint: Some(false),
idempotent_hint: Some(false),
open_world_hint: Some(true),
})
}
fn process_extensions(
extensions: &Value,
_loaded_extensions: &[String],
) -> Option<Vec<ExtensionConfig>> {
// First try to deserialize as ExtensionConfig array
if let Ok(ext_configs) = serde_json::from_value::<Vec<ExtensionConfig>>(extensions.clone()) {
return Some(ext_configs);
}
// Try to handle mixed array of strings and objects
if let Some(arr) = extensions.as_array() {
// If the array is empty, return an empty Vec (not None)
// This is important: empty array means "no extensions"
if arr.is_empty() {
return Some(Vec::new());
}
let mut converted_extensions = Vec::new();
for ext in arr {
if let Some(name_str) = ext.as_str() {
if let Some(config) = crate::config::get_extension_by_name(name_str) {
if crate::config::is_extension_enabled(&config.key()) {
converted_extensions.push(config);
} else {
tracing::warn!("Extension '{}' is disabled, skipping", name_str);
}
} else {
tracing::warn!("Extension '{}' not found in configuration", name_str);
}
} else if let Ok(ext_config) = serde_json::from_value::<ExtensionConfig>(ext.clone()) {
converted_extensions.push(ext_config);
}
}
// Return the converted extensions even if empty
// (empty means user explicitly wants no extensions)
return Some(converted_extensions);
}
None
}
// Helper function to apply recipe builder methods if value can be deserialized
fn apply_if_ok<T: serde::de::DeserializeOwned>(
builder: RecipeBuilder,
value: Option<&Value>,
f: impl FnOnce(RecipeBuilder, T) -> RecipeBuilder,
) -> RecipeBuilder {
match value.and_then(|v| serde_json::from_value(v.clone()).ok()) {
Some(parsed) => f(builder, parsed),
None => builder,
}
}
pub fn task_params_to_inline_recipe(
task_param: &Value,
loaded_extensions: &[String],
) -> Result<Recipe> {
// Extract and validate core fields
let instructions = task_param.get("instructions").and_then(|v| v.as_str());
let prompt = task_param.get("prompt").and_then(|v| v.as_str());
if instructions.is_none() && prompt.is_none() {
return Err(anyhow!("Either 'instructions' or 'prompt' is required"));
}
// Build recipe with auto-generated defaults
let mut builder = Recipe::builder()
.version("1.0.0")
.title(
task_param
.get("title")
.and_then(|v| v.as_str())
.unwrap_or("Dynamic Task"),
)
.description(
task_param
.get("description")
.and_then(|v| v.as_str())
.unwrap_or("Inline recipe task"),
);
// Set instructions/prompt
if let Some(inst) = instructions {
builder = builder.instructions(inst);
}
if let Some(p) = prompt {
builder = builder.prompt(p);
}
// Handle extensions
if let Some(extensions) = task_param.get("extensions") {
if let Some(ext_configs) = process_extensions(extensions, loaded_extensions) {
builder = builder.extensions(ext_configs);
}
}
// Handle other optional fields
builder = apply_if_ok(builder, task_param.get("settings"), RecipeBuilder::settings);
builder = apply_if_ok(builder, task_param.get("response"), RecipeBuilder::response);
builder = apply_if_ok(builder, task_param.get("retry"), RecipeBuilder::retry);
builder = apply_if_ok(
builder,
task_param.get("activities"),
RecipeBuilder::activities,
);
builder = apply_if_ok(
builder,
task_param.get("parameters"),
RecipeBuilder::parameters,
);
// Build and validate
let recipe = builder
.build()
.map_err(|e| anyhow!("Failed to build recipe: {}", e))?;
// Security validation
if recipe.check_for_security_warnings() {
return Err(anyhow!("Recipe contains potentially harmful content"));
}
// Validate retry config if present
if let Some(ref retry) = recipe.retry {
retry
.validate()
.map_err(|e| anyhow!("Invalid retry config: {}", e))?;
}
Ok(recipe)
}
fn extract_task_parameters(params: &Value) -> Vec<Value> {
params
.get("task_parameters")
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default()
}
fn create_task_execution_payload(tasks: Vec<Task>, execution_mode: ExecutionMode) -> Value {
let task_ids: Vec<String> = tasks.iter().map(|task| task.id.clone()).collect();
json!({
"task_ids": task_ids,
"execution_mode": execution_mode
})
}
pub async fn create_dynamic_task(
params: Value,
tasks_manager: &TasksManager,
loaded_extensions: Vec<String>,
parent_working_dir: &std::path::Path,
) -> ToolCallResult {
let task_params_array = extract_task_parameters(&params);
if task_params_array.is_empty() {
return ToolCallResult::from(Err(ErrorData {
code: ErrorCode::INVALID_PARAMS,
message: Cow::from("task_parameters array cannot be empty"),
data: None,
}));
}
// Convert each parameter set to inline recipe and create tasks
let mut tasks = Vec::new();
for task_param in &task_params_array {
// All tasks must use the new inline recipe path
match task_params_to_inline_recipe(task_param, &loaded_extensions) {
Ok(recipe) => {
// Extract return_last_only flag if present
let return_last_only = task_param
.get("return_last_only")
.and_then(|v| v.as_bool())
.unwrap_or(false);
// Create a session for this task - use its ID as the task ID
let session = match SessionManager::create_session(
parent_working_dir.to_path_buf(),
"Subagent task".to_string(),
crate::session::session_manager::SessionType::SubAgent,
)
.await
{
Ok(s) => s,
Err(e) => {
return ToolCallResult::from(Err(ErrorData {
code: ErrorCode::INTERNAL_ERROR,
message: Cow::from(format!("Failed to create session: {}", e)),
data: None,
}));
}
};
let task = Task {
id: session.id,
payload: TaskPayload {
recipe,
return_last_only,
sequential_when_repeated: false,
parameter_values: None,
},
};
tasks.push(task);
}
Err(e) => {
return ToolCallResult::from(Err(ErrorData {
code: ErrorCode::INVALID_PARAMS,
message: Cow::from(format!("Invalid task parameters: {}", e)),
data: None,
}));
}
}
}
let execution_mode = params
.get("execution_mode")
.and_then(|v| v.as_str())
.map(|s| match s {
"sequential" => ExecutionMode::Sequential,
"parallel" => ExecutionMode::Parallel,
_ => ExecutionMode::Parallel,
})
.unwrap_or_else(|| {
if tasks.len() > 1 {
ExecutionMode::Parallel
} else {
ExecutionMode::Sequential
}
});
let task_execution_payload = create_task_execution_payload(tasks.clone(), execution_mode);
let tasks_json = match serde_json::to_string(&task_execution_payload) {
Ok(json) => json,
Err(e) => {
return ToolCallResult::from(Err(ErrorData {
code: ErrorCode::INTERNAL_ERROR,
message: Cow::from(format!("Failed to serialize task list: {}", e)),
data: None,
}))
}
};
tasks_manager.save_tasks(tasks).await;
ToolCallResult::from(Ok(rmcp::model::CallToolResult {
content: vec![Content::text(tasks_json)],
structured_content: None,
is_error: Some(false),
meta: None,
}))
}
@@ -1,3 +0,0 @@
pub mod dynamic_task_tools;
pub mod param_utils;
pub mod sub_recipe_tools;
@@ -1,38 +0,0 @@
use anyhow::Result;
use serde_json::Value;
use std::collections::HashMap;
use crate::recipe::SubRecipe;
pub fn prepare_command_params(
sub_recipe: &SubRecipe,
params_from_tool_call: Vec<Value>,
) -> Result<Vec<HashMap<String, String>>> {
let base_params = sub_recipe.values.clone().unwrap_or_default();
if params_from_tool_call.is_empty() {
return Ok(vec![base_params]);
}
let result = params_from_tool_call
.into_iter()
.map(|tool_param| {
let mut param_map = base_params.clone();
if let Some(param_obj) = tool_param.as_object() {
for (key, value) in param_obj {
let value_str = value
.as_str()
.map(String::from)
.unwrap_or_else(|| value.to_string());
param_map.entry(key.clone()).or_insert(value_str);
}
}
param_map
})
.collect();
Ok(result)
}
#[cfg(test)]
mod tests;
@@ -1,140 +0,0 @@
use std::collections::HashMap;
use crate::recipe::SubRecipe;
use serde_json::json;
use crate::agents::recipe_tools::param_utils::prepare_command_params;
fn setup_default_sub_recipe() -> SubRecipe {
SubRecipe {
name: "test_sub_recipe".to_string(),
path: "test_sub_recipe.yaml".to_string(),
values: Some(HashMap::from([("key1".to_string(), "value1".to_string())])),
sequential_when_repeated: true,
description: Some("Test subrecipe".to_string()),
}
}
mod prepare_command_params_tests {
use super::*;
#[test]
fn test_return_command_param() {
let parameter_array = vec![json!(HashMap::from([(
"key2".to_string(),
"value2".to_string()
)]))];
let mut sub_recipe = setup_default_sub_recipe();
sub_recipe.values = Some(HashMap::from([("key1".to_string(), "value1".to_string())]));
let result = prepare_command_params(&sub_recipe, parameter_array).unwrap();
assert_eq!(
vec![HashMap::from([
("key1".to_string(), "value1".to_string()),
("key2".to_string(), "value2".to_string())
]),],
result
);
}
#[test]
fn test_return_command_param_when_value_override_passed_param_value() {
let parameter_array = vec![json!(HashMap::from([(
"key2".to_string(),
"different_value".to_string()
)]))];
let mut sub_recipe = setup_default_sub_recipe();
sub_recipe.values = Some(HashMap::from([
("key1".to_string(), "value1".to_string()),
("key2".to_string(), "value2".to_string()),
]));
let result = prepare_command_params(&sub_recipe, parameter_array).unwrap();
assert_eq!(
vec![HashMap::from([
("key1".to_string(), "value1".to_string()),
("key2".to_string(), "value2".to_string())
]),],
result
);
}
#[test]
fn test_return_empty_command_param() {
let parameter_array = vec![];
let mut sub_recipe = setup_default_sub_recipe();
sub_recipe.values = None;
let result = prepare_command_params(&sub_recipe, parameter_array).unwrap();
assert_eq!(result, vec![HashMap::new()]);
}
mod multiple_tool_parameters {
use super::*;
#[test]
fn test_return_command_param_when_all_values_from_tool_call_parameters() {
let parameter_array = vec![
json!(HashMap::from([
("key1".to_string(), "key1_value1".to_string()),
("key2".to_string(), "key2_value1".to_string())
])),
json!(HashMap::from([
("key1".to_string(), "key1_value2".to_string()),
("key2".to_string(), "key2_value2".to_string())
])),
];
let mut sub_recipe = setup_default_sub_recipe();
sub_recipe.values = None;
let result = prepare_command_params(&sub_recipe, parameter_array).unwrap();
assert_eq!(
vec![
HashMap::from([
("key1".to_string(), "key1_value1".to_string()),
("key2".to_string(), "key2_value1".to_string()),
]),
HashMap::from([
("key1".to_string(), "key1_value2".to_string()),
("key2".to_string(), "key2_value2".to_string()),
]),
],
result
);
}
#[test]
fn test_merge_base_values_with_tool_parameters() {
let parameter_array = vec![
json!(HashMap::from([(
"key2".to_string(),
"override_value1".to_string()
)])),
json!(HashMap::from([(
"key2".to_string(),
"override_value2".to_string()
)])),
];
let mut sub_recipe = setup_default_sub_recipe();
sub_recipe.values = Some(HashMap::from([
("key1".to_string(), "base_value".to_string()),
("key2".to_string(), "original_value".to_string()),
]));
let result = prepare_command_params(&sub_recipe, parameter_array).unwrap();
assert_eq!(
vec![
HashMap::from([
("key1".to_string(), "base_value".to_string()),
("key2".to_string(), "original_value".to_string()),
]),
HashMap::from([
("key1".to_string(), "base_value".to_string()),
("key2".to_string(), "original_value".to_string()),
]),
],
result
);
}
}
}
@@ -1,204 +0,0 @@
use std::collections::HashSet;
use std::fs;
use std::sync::Arc;
use anyhow::Result;
use rmcp::model::{Tool, ToolAnnotations};
use serde_json::{json, Map, Value};
use crate::agents::subagent_execution_tool::lib::ExecutionMode;
use crate::agents::subagent_execution_tool::task_types::{Task, TaskPayload};
use crate::agents::subagent_execution_tool::tasks_manager::TasksManager;
use crate::recipe::build_recipe::build_recipe_from_template;
use crate::recipe::local_recipes::load_local_recipe_file;
use crate::recipe::{Recipe, RecipeParameter, RecipeParameterRequirement, SubRecipe};
use crate::session::SessionManager;
use super::param_utils::prepare_command_params;
pub const SUB_RECIPE_TASK_TOOL_NAME_PREFIX: &str = "subrecipe__create_task";
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_TASK_TOOL_NAME_PREFIX, sub_recipe.name),
format!(
"Create one or more tasks to run the '{}' sub recipe. \
Provide an array of parameter sets in the 'task_parameters' field:\n\
- For a single task: provide an array with one parameter set\n\
- For multiple tasks: provide an array with multiple parameter sets, each with different values\n\n\
Each task will run the same sub recipe but with different parameter values. \
This is useful when you need to execute the same sub recipe multiple times with varying inputs. \
After creating the tasks and execution_mode is provided, pass them to the task executor to run these tasks",
sub_recipe.name
),
Arc::new(input_schema.as_object().unwrap().clone())
).annotate(ToolAnnotations {
title: Some(format!(
"create multiple sub recipe tasks for {}",
sub_recipe.name
)),
read_only_hint: Some(false),
destructive_hint: Some(true),
idempotent_hint: Some(false),
open_world_hint: Some(true),
})
}
fn extract_task_parameters(params: &Value) -> Vec<Value> {
params
.get("task_parameters")
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default()
}
async fn create_tasks_from_params(
sub_recipe: &SubRecipe,
command_params: &[std::collections::HashMap<String, String>],
parent_working_dir: &std::path::Path,
) -> Result<Vec<Task>> {
let recipe_file = load_local_recipe_file(&sub_recipe.path)
.map_err(|e| anyhow::anyhow!("Failed to load recipe {}: {}", sub_recipe.path, e))?;
let mut tasks = Vec::new();
for task_command_param in command_params {
let session = SessionManager::create_session(
parent_working_dir.to_path_buf(),
format!("Subagent: {}", sub_recipe.name),
crate::session::session_manager::SessionType::SubAgent,
)
.await?;
let recipe = build_recipe_from_template(
recipe_file.content.clone(),
&recipe_file.parent_dir,
task_command_param
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect(),
None::<fn(&str, &str) -> Result<String, anyhow::Error>>,
)
.map_err(|e| anyhow::anyhow!("Failed to build recipe: {}", e))?;
let task = Task {
id: session.id,
payload: TaskPayload {
recipe,
return_last_only: false,
sequential_when_repeated: sub_recipe.sequential_when_repeated,
parameter_values: Some(task_command_param.clone()),
},
};
tasks.push(task);
}
Ok(tasks)
}
fn create_task_execution_payload(tasks: &[Task], sub_recipe: &SubRecipe) -> Value {
let execution_mode = if tasks.len() == 1 || sub_recipe.sequential_when_repeated {
ExecutionMode::Sequential
} else {
ExecutionMode::Parallel
};
let task_ids: Vec<String> = tasks.iter().map(|task| task.id.clone()).collect();
json!({
"task_ids": task_ids,
"execution_mode": execution_mode,
})
}
pub async fn create_sub_recipe_task(
sub_recipe: &SubRecipe,
params: Value,
tasks_manager: &TasksManager,
parent_working_dir: &std::path::Path,
) -> Result<String> {
let task_params_array = extract_task_parameters(&params);
let command_params = prepare_command_params(sub_recipe, task_params_array.clone())?;
let tasks = create_tasks_from_params(sub_recipe, &command_params, parent_working_dir).await?;
let task_execution_payload = create_task_execution_payload(&tasks, sub_recipe);
let tasks_json = serde_json::to_string(&task_execution_payload)
.map_err(|e| anyhow::anyhow!("Failed to serialize task list: {}", e))?;
tasks_manager.save_tasks(tasks.clone()).await;
Ok(tasks_json)
}
fn get_sub_recipe_parameter_definition(
sub_recipe: &SubRecipe,
) -> Result<Option<Vec<RecipeParameter>>> {
let content = fs::read_to_string(sub_recipe.path.clone())
.map_err(|e| anyhow::anyhow!("Failed to read recipe file {}: {}", sub_recipe.path, e))?;
let recipe = Recipe::from_content(&content)?;
Ok(recipe.parameters)
}
fn get_params_with_values(sub_recipe: &SubRecipe) -> HashSet<String> {
let mut sub_recipe_params_with_values = HashSet::<String>::new();
if let Some(params_with_value) = &sub_recipe.values {
for param_name in params_with_value.keys() {
sub_recipe_params_with_values.insert(param_name.clone());
}
}
sub_recipe_params_with_values
}
fn create_input_schema(param_properties: Map<String, Value>, param_required: Vec<String>) -> Value {
let mut properties = Map::new();
if !param_properties.is_empty() {
properties.insert(
"task_parameters".to_string(),
json!({
"type": "array",
"description": "Array of parameter sets for creating tasks. \
For a single task, provide an array with one element. \
For multiple tasks, provide an array with multiple elements, each with different parameter values. \
If there is no parameter set, provide an empty array.",
"items": {
"type": "object",
"properties": param_properties,
"required": param_required
},
})
);
}
json!({
"type": "object",
"properties": properties,
})
}
fn get_input_schema(sub_recipe: &SubRecipe) -> Result<Value> {
let sub_recipe_params_with_values = get_params_with_values(sub_recipe);
let parameter_definition = get_sub_recipe_parameter_definition(sub_recipe)?;
let mut param_properties = Map::new();
let mut param_required = Vec::new();
if let Some(parameters) = parameter_definition {
for param in parameters {
if sub_recipe_params_with_values.contains(&param.key.clone()) {
continue;
}
param_properties.insert(
param.key.clone(),
json!({
"type": param.input_type.to_string(),
"description": param.description.clone(),
}),
);
if !matches!(param.requirement, RecipeParameterRequirement::Optional) {
param_required.push(param.key);
}
}
}
Ok(create_input_schema(param_properties, param_required))
}
#[cfg(test)]
mod tests;
@@ -1,128 +0,0 @@
use std::collections::HashMap;
use crate::recipe::SubRecipe;
use serde_json::json;
use serde_json::Value;
use tempfile::TempDir;
fn setup_default_sub_recipe() -> SubRecipe {
SubRecipe {
name: "test_sub_recipe".to_string(),
path: "test_sub_recipe.yaml".to_string(),
values: Some(HashMap::from([("key1".to_string(), "value1".to_string())])),
sequential_when_repeated: true,
description: Some("Test subrecipe".to_string()),
}
}
mod get_input_schema {
use super::*;
use crate::agents::recipe_tools::sub_recipe_tools::get_input_schema;
fn prepare_sub_recipe(sub_recipe_file_content: &str) -> (SubRecipe, TempDir) {
let mut sub_recipe = setup_default_sub_recipe();
let temp_dir = tempfile::tempdir().unwrap();
let temp_file = temp_dir.path().join(sub_recipe.path.clone());
std::fs::write(&temp_file, sub_recipe_file_content).unwrap();
sub_recipe.path = temp_file.to_string_lossy().to_string();
(sub_recipe, temp_dir)
}
fn verify_task_parameters(result: Value, expected_task_parameters_items: Value) {
let task_parameters = result
.get("properties")
.unwrap()
.as_object()
.unwrap()
.get("task_parameters")
.unwrap()
.as_object()
.unwrap();
let task_parameters_items = task_parameters.get("items").unwrap();
assert_eq!(&expected_task_parameters_items, task_parameters_items);
}
const SUB_RECIPE_FILE_CONTENT_WITH_TWO_PARAMS: &str = r#"{
"version": "1.0.0",
"title": "Test Recipe",
"description": "A test recipe",
"prompt": "Test prompt",
"parameters": [
{
"key": "key1",
"input_type": "string",
"requirement": "required",
"description": "A test parameter"
},
{
"key": "key2",
"input_type": "number",
"requirement": "optional",
"description": "An optional parameter"
}
]
}"#;
#[test]
fn test_with_one_param_in_tool_input() {
let (mut sub_recipe, _temp_dir) =
prepare_sub_recipe(SUB_RECIPE_FILE_CONTENT_WITH_TWO_PARAMS);
sub_recipe.values = Some(HashMap::from([("key1".to_string(), "value1".to_string())]));
let result = get_input_schema(&sub_recipe).unwrap();
verify_task_parameters(
result,
json!({
"type": "object",
"properties": {
"key2": { "type": "number", "description": "An optional parameter" }
},
"required": []
}),
);
}
#[test]
fn test_without_param_in_tool_input() {
let (mut sub_recipe, _temp_dir) =
prepare_sub_recipe(SUB_RECIPE_FILE_CONTENT_WITH_TWO_PARAMS);
sub_recipe.values = Some(HashMap::from([
("key1".to_string(), "value1".to_string()),
("key2".to_string(), "value2".to_string()),
]));
let result = get_input_schema(&sub_recipe).unwrap();
assert_eq!(
None,
result
.get("properties")
.unwrap()
.as_object()
.unwrap()
.get("task_parameters")
);
}
#[test]
fn test_with_all_params_in_tool_input() {
let (mut sub_recipe, _temp_dir) =
prepare_sub_recipe(SUB_RECIPE_FILE_CONTENT_WITH_TWO_PARAMS);
sub_recipe.values = None;
let result = get_input_schema(&sub_recipe).unwrap();
verify_task_parameters(
result,
json!({
"type": "object",
"properties": {
"key1": { "type": "string", "description": "A test parameter" },
"key2": { "type": "number", "description": "An optional parameter" }
},
"required": ["key1"]
}),
);
}
}
+3 -6
View File
@@ -16,7 +16,7 @@ use crate::providers::toolshim::{
modify_system_prompt_for_tool_json, OllamaInterpreter,
};
use crate::agents::recipe_tools::dynamic_task_tools::should_enabled_subagents;
use crate::agents::subagent_tool::should_enable_subagents;
use crate::session::SessionManager;
#[cfg(test)]
use crate::session::SessionType;
@@ -125,11 +125,8 @@ impl Agent {
let provider = self.provider().await?;
let model_name = provider.get_model_config().model_name;
if !should_enabled_subagents(&model_name) {
tools.retain(|tool| {
tool.name != crate::agents::subagent_execution_tool::subagent_execute_task_tool::SUBAGENT_EXECUTE_TASK_TOOL_NAME
&& tool.name != crate::agents::recipe_tools::dynamic_task_tools::DYNAMIC_TASK_TOOL_NAME_PREFIX
});
if !should_enable_subagents(&model_name) {
tools.retain(|tool| tool.name != crate::agents::subagent_tool::SUBAGENT_TOOL_NAME);
}
}
@@ -1,116 +0,0 @@
use rmcp::model::Tool;
use rmcp::model::{Content, ErrorCode, ErrorData};
use serde_json::Value;
use std::borrow::Cow;
use std::collections::HashMap;
use crate::{
agents::{
recipe_tools::sub_recipe_tools::{
create_sub_recipe_task, create_sub_recipe_task_tool, SUB_RECIPE_TASK_TOOL_NAME_PREFIX,
},
subagent_execution_tool::tasks_manager::TasksManager,
tool_execution::ToolCallResult,
},
recipe::SubRecipe,
};
#[derive(Debug, Clone)]
pub struct SubRecipeManager {
pub sub_recipe_tools: HashMap<String, Tool>,
pub sub_recipes: HashMap<String, SubRecipe>,
}
impl Default for SubRecipeManager {
fn default() -> Self {
Self::new()
}
}
impl SubRecipeManager {
pub fn new() -> Self {
Self {
sub_recipe_tools: HashMap::new(),
sub_recipes: HashMap::new(),
}
}
pub fn add_sub_recipe_tools(&mut self, sub_recipes_to_add: Vec<SubRecipe>) {
for sub_recipe in sub_recipes_to_add {
let sub_recipe_key = format!(
"{}_{}",
SUB_RECIPE_TASK_TOOL_NAME_PREFIX,
sub_recipe.name.clone()
);
let tool = create_sub_recipe_task_tool(&sub_recipe);
self.sub_recipe_tools.insert(sub_recipe_key.clone(), tool);
self.sub_recipes.insert(sub_recipe_key.clone(), sub_recipe);
}
}
pub fn is_sub_recipe_tool(&self, tool_name: &str) -> bool {
self.sub_recipe_tools.contains_key(tool_name)
}
pub async fn dispatch_sub_recipe_tool_call(
&self,
tool_name: &str,
params: Value,
tasks_manager: &TasksManager,
parent_working_dir: &std::path::Path,
) -> ToolCallResult {
let result = self
.call_sub_recipe_tool(tool_name, params, tasks_manager, parent_working_dir)
.await;
match result {
Ok(content) => ToolCallResult::from(Ok(rmcp::model::CallToolResult {
content,
structured_content: None,
is_error: Some(false),
meta: None,
})),
Err(e) => ToolCallResult::from(Err(ErrorData {
code: ErrorCode::INTERNAL_ERROR,
message: Cow::from(e.to_string()),
data: None,
})),
}
}
async fn call_sub_recipe_tool(
&self,
tool_name: &str,
params: Value,
tasks_manager: &TasksManager,
parent_working_dir: &std::path::Path,
) -> Result<Vec<Content>, ErrorData> {
let sub_recipe = self.sub_recipes.get(tool_name).ok_or_else(|| {
let sub_recipe_name = tool_name
.strip_prefix(SUB_RECIPE_TASK_TOOL_NAME_PREFIX)
.and_then(|s| s.strip_prefix("_"))
.ok_or_else(|| ErrorData {
code: ErrorCode::INVALID_PARAMS,
message: Cow::from(format!(
"Invalid sub-recipe tool name format: {}",
tool_name
)),
data: None,
})
.unwrap();
ErrorData {
code: ErrorCode::INVALID_PARAMS,
message: Cow::from(format!("Sub-recipe '{}' not found", sub_recipe_name)),
data: None,
}
})?;
let output = create_sub_recipe_task(sub_recipe, params, tasks_manager, parent_working_dir)
.await
.map_err(|e| ErrorData {
code: ErrorCode::INTERNAL_ERROR,
message: Cow::from(format!("Sub-recipe task creation failed: {}", e)),
data: None,
})?;
Ok(vec![Content::text(output)])
}
}
@@ -1,221 +0,0 @@
use crate::agents::subagent_execution_tool::lib::{
ExecutionResponse, ExecutionStats, SharedState, Task, TaskResult, TaskStatus,
};
use crate::agents::subagent_execution_tool::task_execution_tracker::{
DisplayMode, TaskExecutionTracker,
};
use crate::agents::subagent_execution_tool::tasks::process_task;
use crate::agents::subagent_execution_tool::workers::spawn_worker;
use crate::agents::subagent_task_config::TaskConfig;
use rmcp::model::ServerNotification;
use std::sync::atomic::AtomicUsize;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::sync::mpsc::Sender;
use tokio::time::Instant;
use tokio_util::sync::CancellationToken;
const EXECUTION_STATUS_COMPLETED: &str = "completed";
const DEFAULT_MAX_WORKERS: usize = 10;
pub async fn execute_single_task(
task: &Task,
notifier: mpsc::Sender<ServerNotification>,
task_config: TaskConfig,
cancellation_token: Option<CancellationToken>,
) -> ExecutionResponse {
let start_time = Instant::now();
let task_execution_tracker = Arc::new(TaskExecutionTracker::new(
vec![task.clone()],
DisplayMode::SingleTaskOutput,
notifier,
cancellation_token.clone(),
));
let result = process_task(
task,
task_execution_tracker.clone(),
task_config,
cancellation_token.unwrap_or_default(),
)
.await;
// Complete the task in the tracker
task_execution_tracker
.complete_task(&result.task_id, result.clone())
.await;
let execution_time = start_time.elapsed().as_millis();
let stats = calculate_stats(std::slice::from_ref(&result), execution_time);
ExecutionResponse {
status: EXECUTION_STATUS_COMPLETED.to_string(),
results: vec![result],
stats,
}
}
pub async fn execute_tasks_in_parallel(
tasks: Vec<Task>,
notifier: Sender<ServerNotification>,
task_config: TaskConfig,
cancellation_token: Option<CancellationToken>,
) -> ExecutionResponse {
let task_execution_tracker = Arc::new(TaskExecutionTracker::new(
tasks.clone(),
DisplayMode::MultipleTasksOutput,
notifier,
cancellation_token.clone(),
));
let start_time = Instant::now();
let task_count = tasks.len();
if task_count == 0 {
return create_empty_response();
}
task_execution_tracker.refresh_display().await;
let (task_tx, task_rx, result_tx, mut result_rx) = create_channels(task_count);
if let Err(e) = send_tasks_to_channel(tasks, task_tx).await {
tracing::error!("Task execution failed: {}", e);
return create_error_response(e);
}
let shared_state = create_shared_state(
task_rx,
result_tx,
task_execution_tracker.clone(),
cancellation_token.unwrap_or_default(),
);
let worker_count = std::cmp::min(task_count, DEFAULT_MAX_WORKERS);
let mut worker_handles = Vec::new();
for i in 0..worker_count {
let handle = spawn_worker(shared_state.clone(), i, task_config.clone());
worker_handles.push(handle);
}
let results = collect_results(&mut result_rx, task_execution_tracker.clone(), task_count).await;
for handle in worker_handles {
if let Err(e) = handle.await {
tracing::error!("Worker error: {}", e);
}
}
task_execution_tracker.send_tasks_complete().await;
let execution_time = start_time.elapsed().as_millis();
let stats = calculate_stats(&results, execution_time);
ExecutionResponse {
status: EXECUTION_STATUS_COMPLETED.to_string(),
results,
stats,
}
}
fn calculate_stats(results: &[TaskResult], execution_time_ms: u128) -> ExecutionStats {
let completed = results
.iter()
.filter(|r| matches!(r.status, TaskStatus::Completed))
.count();
let failed = results
.iter()
.filter(|r| matches!(r.status, TaskStatus::Failed))
.count();
ExecutionStats {
total_tasks: results.len(),
completed,
failed,
execution_time_ms,
}
}
fn create_channels(
task_count: usize,
) -> (
mpsc::Sender<Task>,
mpsc::Receiver<Task>,
mpsc::Sender<TaskResult>,
mpsc::Receiver<TaskResult>,
) {
let (task_tx, task_rx) = mpsc::channel::<Task>(task_count);
let (result_tx, result_rx) = mpsc::channel::<TaskResult>(task_count);
(task_tx, task_rx, result_tx, result_rx)
}
fn create_shared_state(
task_rx: mpsc::Receiver<Task>,
result_tx: mpsc::Sender<TaskResult>,
task_execution_tracker: Arc<TaskExecutionTracker>,
cancellation_token: CancellationToken,
) -> Arc<SharedState> {
Arc::new(SharedState {
task_receiver: Arc::new(tokio::sync::Mutex::new(task_rx)),
result_sender: result_tx,
active_workers: Arc::new(AtomicUsize::new(0)),
task_execution_tracker,
cancellation_token,
})
}
async fn send_tasks_to_channel(
tasks: Vec<Task>,
task_tx: mpsc::Sender<Task>,
) -> Result<(), String> {
for task in tasks {
task_tx
.send(task)
.await
.map_err(|e| format!("Failed to queue task: {}", e))?;
}
Ok(())
}
fn create_empty_response() -> ExecutionResponse {
ExecutionResponse {
status: EXECUTION_STATUS_COMPLETED.to_string(),
results: vec![],
stats: ExecutionStats {
total_tasks: 0,
completed: 0,
failed: 0,
execution_time_ms: 0,
},
}
}
async fn collect_results(
result_rx: &mut mpsc::Receiver<TaskResult>,
task_execution_tracker: Arc<TaskExecutionTracker>,
expected_count: usize,
) -> Vec<TaskResult> {
let mut results = Vec::new();
while let Some(result) = result_rx.recv().await {
task_execution_tracker
.complete_task(&result.task_id, result.clone())
.await;
results.push(result);
if results.len() >= expected_count {
break;
}
}
results
}
fn create_error_response(error: String) -> ExecutionResponse {
tracing::error!("Creating error response: {}", error);
ExecutionResponse {
status: "failed".to_string(),
results: vec![],
stats: ExecutionStats {
total_tasks: 0,
completed: 0,
failed: 1,
execution_time_ms: 0,
},
}
}
@@ -1,100 +0,0 @@
use super::{calculate_stats, create_empty_response, create_error_response};
use crate::agents::sub_recipe_execution_tool::lib::{TaskResult, TaskStatus};
use serde_json::json;
fn create_test_task_result(task_id: &str, status: TaskStatus) -> TaskResult {
let is_failed = matches!(status, TaskStatus::Failed);
TaskResult {
task_id: task_id.to_string(),
status,
data: Some(json!({"output": "test output"})),
error: if is_failed {
Some("Test error".to_string())
} else {
None
},
}
}
#[test]
fn test_calculate_stats() {
let results = vec![
create_test_task_result("task1", TaskStatus::Completed),
create_test_task_result("task2", TaskStatus::Completed),
create_test_task_result("task3", TaskStatus::Failed),
create_test_task_result("task4", TaskStatus::Completed),
];
let stats = calculate_stats(&results, 1500);
assert_eq!(stats.total_tasks, 4);
assert_eq!(stats.completed, 3);
assert_eq!(stats.failed, 1);
assert_eq!(stats.execution_time_ms, 1500);
}
#[test]
fn test_calculate_stats_empty_results() {
let results = vec![];
let stats = calculate_stats(&results, 0);
assert_eq!(stats.total_tasks, 0);
assert_eq!(stats.completed, 0);
assert_eq!(stats.failed, 0);
assert_eq!(stats.execution_time_ms, 0);
}
#[test]
fn test_calculate_stats_all_completed() {
let results = vec![
create_test_task_result("task1", TaskStatus::Completed),
create_test_task_result("task2", TaskStatus::Completed),
];
let stats = calculate_stats(&results, 800);
assert_eq!(stats.total_tasks, 2);
assert_eq!(stats.completed, 2);
assert_eq!(stats.failed, 0);
assert_eq!(stats.execution_time_ms, 800);
}
#[test]
fn test_calculate_stats_all_failed() {
let results = vec![
create_test_task_result("task1", TaskStatus::Failed),
create_test_task_result("task2", TaskStatus::Failed),
];
let stats = calculate_stats(&results, 1200);
assert_eq!(stats.total_tasks, 2);
assert_eq!(stats.completed, 0);
assert_eq!(stats.failed, 2);
assert_eq!(stats.execution_time_ms, 1200);
}
#[test]
fn test_create_empty_response() {
let response = create_empty_response();
assert_eq!(response.status, "completed");
assert_eq!(response.results.len(), 0);
assert_eq!(response.stats.total_tasks, 0);
assert_eq!(response.stats.completed, 0);
assert_eq!(response.stats.failed, 0);
assert_eq!(response.stats.execution_time_ms, 0);
}
#[test]
fn test_create_error_response() {
let error_msg = "Test error message";
let response = create_error_response(error_msg.to_string());
assert_eq!(response.status, "failed");
assert_eq!(response.results.len(), 0);
assert_eq!(response.stats.total_tasks, 0);
assert_eq!(response.stats.completed, 0);
assert_eq!(response.stats.failed, 1);
assert_eq!(response.stats.execution_time_ms, 0);
}
@@ -1,117 +0,0 @@
pub use crate::agents::subagent_execution_tool::task_types::{
ExecutionMode, ExecutionResponse, ExecutionStats, SharedState, Task, TaskResult, TaskStatus,
};
use crate::agents::subagent_execution_tool::{
executor::{execute_single_task, execute_tasks_in_parallel},
tasks_manager::TasksManager,
};
use crate::agents::subagent_task_config::TaskConfig;
use rmcp::model::ServerNotification;
use serde_json::{json, Value};
use tokio::sync::mpsc::Sender;
use tokio_util::sync::CancellationToken;
pub async fn execute_tasks(
task_ids: Vec<String>,
execution_mode: ExecutionMode,
notifier: Sender<ServerNotification>,
task_config: TaskConfig,
tasks_manager: &TasksManager,
cancellation_token: Option<CancellationToken>,
) -> Result<Value, String> {
let tasks = tasks_manager.get_tasks(&task_ids).await?;
let task_count = tasks.len();
match execution_mode {
ExecutionMode::Sequential => {
if task_count == 1 {
let response =
execute_single_task(&tasks[0], notifier, task_config, cancellation_token).await;
handle_response(response)
} else {
Err("Sequential execution mode requires exactly one task".to_string())
}
}
ExecutionMode::Parallel => {
let any_sequential = tasks
.iter()
.any(|task| task.payload.sequential_when_repeated);
if any_sequential {
Ok(json!(
{
"execution_mode": ExecutionMode::Sequential,
"task_ids": task_ids,
"results": ["the tasks should be executed sequentially, no matter how user requests it. Please use the subrecipe__execute_task tool to execute the tasks sequentially."]
}
))
} else {
let response: ExecutionResponse = execute_tasks_in_parallel(
tasks,
notifier.clone(),
task_config,
cancellation_token,
)
.await;
handle_response(response)
}
}
}
}
fn extract_failed_tasks(results: &[TaskResult]) -> Vec<String> {
results
.iter()
.filter(|r| matches!(r.status, TaskStatus::Failed))
.map(format_failed_task_error)
.collect()
}
fn format_failed_task_error(result: &TaskResult) -> String {
let error_msg = result.error.as_deref().unwrap_or("Unknown error");
let partial_output = result
.data
.as_ref()
.and_then(|d| d.get("partial_output"))
.and_then(|v| v.as_str())
.filter(|s| !s.trim().is_empty())
.unwrap_or("No output captured");
format!(
"Task '{}' ({}): {}\nOutput: {}",
result.task_id,
get_task_description(result),
error_msg,
partial_output
)
}
fn format_error_summary(
failed_count: usize,
total_count: usize,
failed_tasks: Vec<String>,
) -> String {
format!(
"{}/{} tasks failed:\n{}",
failed_count,
total_count,
failed_tasks.join("\n")
)
}
fn handle_response(response: ExecutionResponse) -> Result<Value, String> {
if response.stats.failed > 0 {
let failed_tasks = extract_failed_tasks(&response.results);
let error_summary = format_error_summary(
response.stats.failed,
response.stats.total_tasks,
failed_tasks,
);
return Err(error_summary);
}
serde_json::to_value(response).map_err(|e| format!("Failed to serialize response: {}", e))
}
fn get_task_description(result: &TaskResult) -> String {
format!("ID: {}", result.task_id)
}
@@ -1,216 +0,0 @@
use super::{
extract_failed_tasks, format_error_summary, format_failed_task_error, get_task_description,
handle_response,
};
use crate::agents::sub_recipe_execution_tool::lib::{
ExecutionResponse, ExecutionStats, TaskResult, TaskStatus,
};
use serde_json::json;
fn create_test_task_result(task_id: &str, status: TaskStatus, error: Option<String>) -> TaskResult {
TaskResult {
task_id: task_id.to_string(),
status,
data: Some(json!({"partial_output": "test output"})),
error,
}
}
fn create_test_execution_response(
results: Vec<TaskResult>,
failed_count: usize,
) -> ExecutionResponse {
ExecutionResponse {
status: "completed".to_string(),
results: results.clone(),
stats: ExecutionStats {
total_tasks: results.len(),
completed: results.len() - failed_count,
failed: failed_count,
execution_time_ms: 1000,
},
}
}
#[test]
fn test_extract_failed_tasks() {
let results = vec![
create_test_task_result("task1", TaskStatus::Completed, None),
create_test_task_result(
"task2",
TaskStatus::Failed,
Some("Error message".to_string()),
),
create_test_task_result("task3", TaskStatus::Completed, None),
create_test_task_result(
"task4",
TaskStatus::Failed,
Some("Another error".to_string()),
),
];
let failed_tasks = extract_failed_tasks(&results);
assert_eq!(failed_tasks.len(), 2);
assert!(failed_tasks[0].contains("task2"));
assert!(failed_tasks[0].contains("Error message"));
assert!(failed_tasks[1].contains("task4"));
assert!(failed_tasks[1].contains("Another error"));
}
#[test]
fn test_extract_failed_tasks_empty() {
let results = vec![
create_test_task_result("task1", TaskStatus::Completed, None),
create_test_task_result("task2", TaskStatus::Completed, None),
];
let failed_tasks = extract_failed_tasks(&results);
assert_eq!(failed_tasks.len(), 0);
}
#[test]
fn test_format_failed_task_error_with_error_message() {
let result = create_test_task_result(
"task1",
TaskStatus::Failed,
Some("Test error message".to_string()),
);
let formatted = format_failed_task_error(&result);
assert!(formatted.contains("task1"));
assert!(formatted.contains("Test error message"));
assert!(formatted.contains("test output"));
assert!(formatted.contains("ID: task1"));
}
#[test]
fn test_format_failed_task_error_without_error_message() {
let result = create_test_task_result("task2", TaskStatus::Failed, None);
let formatted = format_failed_task_error(&result);
assert!(formatted.contains("task2"));
assert!(formatted.contains("Unknown error"));
assert!(formatted.contains("test output"));
}
#[test]
fn test_format_failed_task_error_empty_partial_output() {
let mut result =
create_test_task_result("task3", TaskStatus::Failed, Some("Error".to_string()));
result.data = Some(json!({"partial_output": ""}));
let formatted = format_failed_task_error(&result);
assert!(formatted.contains("No output captured"));
}
#[test]
fn test_format_failed_task_error_no_partial_output() {
let mut result =
create_test_task_result("task4", TaskStatus::Failed, Some("Error".to_string()));
result.data = Some(json!({}));
let formatted = format_failed_task_error(&result);
assert!(formatted.contains("No output captured"));
}
#[test]
fn test_format_failed_task_error_no_data() {
let mut result =
create_test_task_result("task5", TaskStatus::Failed, Some("Error".to_string()));
result.data = None;
let formatted = format_failed_task_error(&result);
assert!(formatted.contains("No output captured"));
}
#[test]
fn test_format_error_summary() {
let failed_tasks = vec![
"Task 'task1': Error 1\nOutput: output1".to_string(),
"Task 'task2': Error 2\nOutput: output2".to_string(),
];
let summary = format_error_summary(2, 5, failed_tasks);
assert_eq!(summary, "2/5 tasks failed:\nTask 'task1': Error 1\nOutput: output1\nTask 'task2': Error 2\nOutput: output2");
}
#[test]
fn test_format_error_summary_single_failure() {
let failed_tasks = vec!["Task 'task1': Error\nOutput: output".to_string()];
let summary = format_error_summary(1, 3, failed_tasks);
assert_eq!(
summary,
"1/3 tasks failed:\nTask 'task1': Error\nOutput: output"
);
}
#[test]
fn test_handle_response_success() {
let results = vec![
create_test_task_result("task1", TaskStatus::Completed, None),
create_test_task_result("task2", TaskStatus::Completed, None),
];
let response = create_test_execution_response(results, 0);
let result = handle_response(response);
assert!(result.is_ok());
let value = result.unwrap();
assert_eq!(value["status"], "completed");
assert_eq!(value["stats"]["failed"], 0);
}
#[test]
fn test_handle_response_with_failures() {
let results = vec![
create_test_task_result("task1", TaskStatus::Completed, None),
create_test_task_result("task2", TaskStatus::Failed, Some("Test error".to_string())),
];
let response = create_test_execution_response(results, 1);
let result = handle_response(response);
assert!(result.is_err());
let error = result.unwrap_err();
assert!(error.contains("1/2 tasks failed"));
assert!(error.contains("task2"));
assert!(error.contains("Test error"));
}
#[test]
fn test_handle_response_all_failures() {
let results = vec![
create_test_task_result("task1", TaskStatus::Failed, Some("Error 1".to_string())),
create_test_task_result("task2", TaskStatus::Failed, Some("Error 2".to_string())),
];
let response = create_test_execution_response(results, 2);
let result = handle_response(response);
assert!(result.is_err());
let error = result.unwrap_err();
assert!(error.contains("2/2 tasks failed"));
assert!(error.contains("task1"));
assert!(error.contains("task2"));
assert!(error.contains("Error 1"));
assert!(error.contains("Error 2"));
}
#[test]
fn test_get_task_description() {
let result = create_test_task_result("test_task_123", TaskStatus::Completed, None);
let description = get_task_description(&result);
assert_eq!(description, "ID: test_task_123");
}
@@ -1,10 +1,5 @@
mod executor;
pub mod lib;
pub mod notification_events;
pub mod subagent_execute_task_tool;
pub mod task_execution_tracker;
pub mod task_types;
pub mod tasks;
pub mod tasks_manager;
pub mod utils;
pub mod workers;
pub mod lib {
pub use super::notification_events::TaskStatus;
}
@@ -1,7 +1,25 @@
use crate::agents::subagent_execution_tool::task_types::TaskStatus;
use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum TaskStatus {
Pending,
Running,
Completed,
Failed,
}
impl std::fmt::Display for TaskStatus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
TaskStatus::Pending => write!(f, "Pending"),
TaskStatus::Running => write!(f, "Running"),
TaskStatus::Completed => write!(f, "Completed"),
TaskStatus::Failed => write!(f, "Failed"),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "subtype")]
pub enum TaskExecutionNotificationEvent {
@@ -1,107 +0,0 @@
use std::borrow::Cow;
use crate::agents::subagent_task_config::TaskConfig;
use crate::agents::{
subagent_execution_tool::lib::execute_tasks,
subagent_execution_tool::task_types::ExecutionMode,
subagent_execution_tool::tasks_manager::TasksManager, tool_execution::ToolCallResult,
};
use rmcp::model::{Content, ErrorCode, ErrorData, ServerNotification, Tool, ToolAnnotations};
use rmcp::object;
use tokio::sync::mpsc;
use tokio_stream;
use tokio_util::sync::CancellationToken;
pub const SUBAGENT_EXECUTE_TASK_TOOL_NAME: &str = "subagent__execute_task";
pub fn create_subagent_execute_task_tool() -> Tool {
Tool::new(
SUBAGENT_EXECUTE_TASK_TOOL_NAME,
"Only use the subagent__execute_task tool when you execute sub recipe task or dynamic task.
EXECUTION STRATEGY DECISION:
1. If the tasks are created with execution_mode, use the execution_mode.
2. Execute tasks sequentially unless user explicitly requests parallel execution. PARALLEL: User uses keywords like 'parallel', 'simultaneously', 'at the same time', 'concurrently'
IMPLEMENTATION:
- Sequential execution: Call this tool multiple times, passing exactly ONE task per call
- Parallel execution: Call this tool once, passing an ARRAY of all tasks
EXAMPLES:
User Intent Based:
- User: 'get weather and tell me a joke' Sequential (2 separate tool calls, 1 task each)
- User: 'get weather and joke in parallel' Parallel (1 tool call with array of 2 tasks)
- User: 'run these simultaneously' Parallel (1 tool call with task array)
- User: 'do task A then task B' Sequential (2 separate tool calls)",
object!({
"type": "object",
"properties": {
"execution_mode": {
"type": "string",
"enum": ["sequential", "parallel"],
"default": "sequential",
"description": "Execution strategy for multiple tasks. Use 'sequential' (default) unless user explicitly requests parallel execution with words like 'parallel', 'simultaneously', 'at the same time', or 'concurrently'."
},
"task_ids": {
"type": "array",
"items": {
"type": "string",
"description": "Unique identifier for the task"
}
}
},
"required": ["task_ids"]
})
).annotate(ToolAnnotations {
title: Some("Run tasks in parallel".to_string()),
read_only_hint: Some(false),
destructive_hint: Some(true),
idempotent_hint: Some(false),
open_world_hint: Some(true),
})
}
pub async fn run_tasks(
task_ids: Vec<String>,
execution_mode: ExecutionMode,
task_config: TaskConfig,
tasks_manager: &TasksManager,
cancellation_token: Option<CancellationToken>,
) -> ToolCallResult {
let (notification_tx, notification_rx) = mpsc::channel::<ServerNotification>(100);
let tasks_manager_clone = tasks_manager.clone();
let result_future = async move {
match execute_tasks(
task_ids,
execution_mode,
notification_tx,
task_config,
&tasks_manager_clone,
cancellation_token,
)
.await
{
Ok(result) => {
let output = serde_json::to_string(&result).unwrap();
Ok(rmcp::model::CallToolResult {
content: vec![Content::text(output)],
structured_content: None,
is_error: Some(false),
meta: None,
})
}
Err(e) => Err(ErrorData {
code: ErrorCode::INTERNAL_ERROR,
message: Cow::from(e.to_string()),
data: None,
}),
}
};
// Convert receiver to stream
let notification_stream = tokio_stream::wrappers::ReceiverStream::new(notification_rx);
ToolCallResult {
result: Box::new(Box::pin(result_future)),
notification_stream: Some(Box::new(notification_stream)),
}
}
@@ -1,303 +0,0 @@
use rmcp::model::{
LoggingLevel, LoggingMessageNotification, LoggingMessageNotificationMethod,
LoggingMessageNotificationParam, ServerNotification,
};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{mpsc, RwLock};
use tokio::time::{sleep, Duration, Instant};
use tokio_util::sync::CancellationToken;
use crate::agents::subagent_execution_tool::notification_events::{
FailedTaskInfo, TaskCompletionStats, TaskExecutionNotificationEvent, TaskExecutionStats,
TaskInfo as EventTaskInfo,
};
use crate::agents::subagent_execution_tool::task_types::{Task, TaskInfo, TaskResult, TaskStatus};
use crate::agents::subagent_execution_tool::utils::{count_by_status, get_task_name};
use crate::utils::is_token_cancelled;
use tokio::sync::mpsc::Sender;
const RECIPE_TASK_TYPE: &str = "recipe";
#[derive(Debug, Clone, PartialEq)]
pub enum DisplayMode {
MultipleTasksOutput,
SingleTaskOutput,
}
const THROTTLE_INTERVAL_MS: u64 = 250;
const COMPLETION_NOTIFICATION_DELAY_MS: u64 = 500;
fn format_task_metadata(task_info: &TaskInfo) -> String {
// If we have parameter values, format them nicely
if let Some(ref params) = task_info.task.payload.parameter_values {
if !params.is_empty() {
let mut param_strs: Vec<String> = params
.iter()
.filter(|(k, _)| k.as_str() != "recipe_dir")
.map(|(k, v)| format!("{}={}", k, v))
.collect();
if !param_strs.is_empty() {
param_strs.sort();
return param_strs.join(", ");
}
}
}
// Fallback to recipe title if no parameters
task_info.task.payload.recipe.title.clone()
}
pub struct TaskExecutionTracker {
tasks: Arc<RwLock<HashMap<String, TaskInfo>>>,
last_refresh: Arc<RwLock<Instant>>,
notifier: mpsc::Sender<ServerNotification>,
display_mode: DisplayMode,
cancellation_token: Option<CancellationToken>,
}
impl TaskExecutionTracker {
pub fn new(
tasks: Vec<Task>,
display_mode: DisplayMode,
notifier: Sender<ServerNotification>,
cancellation_token: Option<CancellationToken>,
) -> Self {
let task_map = tasks
.into_iter()
.map(|task| {
let task_id = task.id.clone();
(
task_id,
TaskInfo {
task,
status: TaskStatus::Pending,
start_time: None,
end_time: None,
result: None,
current_output: String::new(),
},
)
})
.collect();
Self {
tasks: Arc::new(RwLock::new(task_map)),
last_refresh: Arc::new(RwLock::new(Instant::now())),
notifier,
display_mode,
cancellation_token,
}
}
fn is_cancelled(&self) -> bool {
is_token_cancelled(&self.cancellation_token)
}
fn log_notification_error<T>(&self, error: &mpsc::error::TrySendError<T>, context: &str) {
if !self.is_cancelled() {
tracing::warn!("Failed to send {} notification: {}", context, error);
}
}
fn try_send_notification(&self, event: TaskExecutionNotificationEvent, context: &str) {
if let Err(e) = self
.notifier
.try_send(ServerNotification::LoggingMessageNotification(
LoggingMessageNotification {
method: LoggingMessageNotificationMethod,
params: LoggingMessageNotificationParam {
data: event.to_notification_data(),
level: LoggingLevel::Info,
logger: None,
},
extensions: Default::default(),
},
))
{
self.log_notification_error(&e, context);
}
}
pub async fn start_task(&self, task_id: &str) {
let mut tasks = self.tasks.write().await;
if let Some(task_info) = tasks.get_mut(task_id) {
task_info.status = TaskStatus::Running;
task_info.start_time = Some(Instant::now());
}
drop(tasks);
self.force_refresh_display().await;
}
pub async fn complete_task(&self, task_id: &str, result: TaskResult) {
let mut tasks = self.tasks.write().await;
if let Some(task_info) = tasks.get_mut(task_id) {
task_info.status = result.status.clone();
task_info.end_time = Some(Instant::now());
task_info.result = Some(result);
}
drop(tasks);
self.force_refresh_display().await;
}
pub async fn get_current_output(&self, task_id: &str) -> Option<String> {
let tasks = self.tasks.read().await;
tasks
.get(task_id)
.map(|task_info| task_info.current_output.clone())
}
async fn format_line(&self, task_info: Option<&TaskInfo>, line: &str) -> String {
if let Some(task_info) = task_info {
let task_name = get_task_name(task_info);
let metadata = format_task_metadata(task_info);
if metadata.is_empty() {
format!("[{} ({})] {}", task_name, RECIPE_TASK_TYPE, line)
} else {
format!(
"[{} ({}) {}] {}",
task_name, RECIPE_TASK_TYPE, metadata, line
)
}
} else {
line.to_string()
}
}
pub async fn send_live_output(&self, task_id: &str, line: &str) {
match self.display_mode {
DisplayMode::SingleTaskOutput => {
let tasks = self.tasks.read().await;
let task_info = tasks.get(task_id);
let formatted_line = self.format_line(task_info, line).await;
drop(tasks);
let event = TaskExecutionNotificationEvent::line_output(
task_id.to_string(),
formatted_line,
);
self.try_send_notification(event, "live output");
}
DisplayMode::MultipleTasksOutput => {
let mut tasks = self.tasks.write().await;
if let Some(task_info) = tasks.get_mut(task_id) {
task_info.current_output.push_str(line);
task_info.current_output.push('\n');
}
drop(tasks);
if !self.should_throttle_refresh().await {
self.refresh_display().await;
}
}
}
}
async fn should_throttle_refresh(&self) -> bool {
let now = Instant::now();
let mut last_refresh = self.last_refresh.write().await;
if now.duration_since(*last_refresh) > Duration::from_millis(THROTTLE_INTERVAL_MS) {
*last_refresh = now;
false
} else {
true
}
}
async fn send_tasks_update(&self) {
if self.is_cancelled() {
return;
}
let tasks = self.tasks.read().await;
let task_list: Vec<_> = tasks.values().collect();
let (total, pending, running, completed, failed) = count_by_status(&tasks);
let stats = TaskExecutionStats::new(total, pending, running, completed, failed);
let event_tasks: Vec<EventTaskInfo> = task_list
.iter()
.map(|task_info| {
let now = Instant::now();
EventTaskInfo {
id: task_info.task.id.clone(),
status: task_info.status.clone(),
duration_secs: task_info.start_time.map(|start| {
if let Some(end) = task_info.end_time {
end.duration_since(start).as_secs_f64()
} else {
now.duration_since(start).as_secs_f64()
}
}),
current_output: task_info.current_output.clone(),
task_type: RECIPE_TASK_TYPE.to_string(),
task_name: get_task_name(task_info).to_string(),
task_metadata: format_task_metadata(task_info),
error: task_info.error().cloned(),
result_data: task_info.data().cloned(),
}
})
.collect();
let event = TaskExecutionNotificationEvent::tasks_update(stats, event_tasks);
self.try_send_notification(event, "tasks update");
}
pub async fn refresh_display(&self) {
match self.display_mode {
DisplayMode::MultipleTasksOutput => {
self.send_tasks_update().await;
}
DisplayMode::SingleTaskOutput => {
// No dashboard display needed for single task output mode
// Live output is handled via send_live_output method
}
}
}
// Force refresh without throttling - used for important status changes
async fn force_refresh_display(&self) {
match self.display_mode {
DisplayMode::MultipleTasksOutput => {
// Reset throttle timer to allow immediate update
let mut last_refresh = self.last_refresh.write().await;
*last_refresh = Instant::now() - Duration::from_millis(THROTTLE_INTERVAL_MS + 1);
drop(last_refresh);
self.send_tasks_update().await;
}
DisplayMode::SingleTaskOutput => {
// No dashboard display needed for single task output mode
}
}
}
pub async fn send_tasks_complete(&self) {
if self.is_cancelled() {
return;
}
let tasks = self.tasks.read().await;
let (total, _, _, completed, failed) = count_by_status(&tasks);
let stats = TaskCompletionStats::new(total, completed, failed);
let failed_tasks: Vec<FailedTaskInfo> = tasks
.values()
.filter(|task_info| matches!(task_info.status, TaskStatus::Failed))
.map(|task_info| FailedTaskInfo {
id: task_info.task.id.clone(),
name: get_task_name(task_info).to_string(),
error: task_info.error().cloned(),
})
.collect();
let event = TaskExecutionNotificationEvent::tasks_complete(stats, failed_tasks);
self.try_send_notification(event, "tasks complete");
// Wait for the notification to be recieved and displayed before clearing the tasks
sleep(Duration::from_millis(COMPLETION_NOTIFICATION_DELAY_MS)).await;
}
}
@@ -1,115 +0,0 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use crate::agents::subagent_execution_tool::task_execution_tracker::TaskExecutionTracker;
use crate::recipe::Recipe;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
#[serde(rename_all = "lowercase")]
pub enum ExecutionMode {
#[default]
Sequential,
Parallel,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskPayload {
pub recipe: Recipe,
pub return_last_only: bool,
pub sequential_when_repeated: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub parameter_values: Option<HashMap<String, String>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Task {
pub id: String,
pub payload: TaskPayload,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskResult {
pub task_id: String,
pub status: TaskStatus,
#[serde(skip_serializing_if = "Option::is_none")]
pub data: Option<Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum TaskStatus {
Pending,
Running,
Completed,
Failed,
}
impl std::fmt::Display for TaskStatus {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
TaskStatus::Pending => write!(f, "Pending"),
TaskStatus::Running => write!(f, "Running"),
TaskStatus::Completed => write!(f, "Completed"),
TaskStatus::Failed => write!(f, "Failed"),
}
}
}
#[derive(Debug, Clone)]
pub struct TaskInfo {
pub task: Task,
pub status: TaskStatus,
pub start_time: Option<tokio::time::Instant>,
pub end_time: Option<tokio::time::Instant>,
pub result: Option<TaskResult>,
pub current_output: String,
}
impl TaskInfo {
pub fn error(&self) -> Option<&String> {
self.result.as_ref().and_then(|r| r.error.as_ref())
}
pub fn data(&self) -> Option<&Value> {
self.result.as_ref().and_then(|r| r.data.as_ref())
}
}
pub struct SharedState {
pub task_receiver: Arc<tokio::sync::Mutex<mpsc::Receiver<Task>>>,
pub result_sender: mpsc::Sender<TaskResult>,
pub active_workers: Arc<AtomicUsize>,
pub task_execution_tracker: Arc<TaskExecutionTracker>,
pub cancellation_token: CancellationToken,
}
impl SharedState {
pub fn increment_active_workers(&self) {
self.active_workers.fetch_add(1, Ordering::SeqCst);
}
pub fn decrement_active_workers(&self) {
self.active_workers.fetch_sub(1, Ordering::SeqCst);
}
}
#[derive(Debug, Serialize)]
pub struct ExecutionStats {
pub total_tasks: usize,
pub completed: usize,
pub failed: usize,
pub execution_time_ms: u128,
}
#[derive(Debug, Serialize)]
pub struct ExecutionResponse {
pub status: String,
pub results: Vec<TaskResult>,
pub stats: ExecutionStats,
}
@@ -1,94 +0,0 @@
use serde_json::Value;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use crate::agents::subagent_execution_tool::task_execution_tracker::TaskExecutionTracker;
use crate::agents::subagent_execution_tool::task_types::{Task, TaskResult, TaskStatus};
use crate::agents::subagent_task_config::TaskConfig;
pub async fn process_task(
task: &Task,
_task_execution_tracker: Arc<TaskExecutionTracker>,
task_config: TaskConfig,
cancellation_token: CancellationToken,
) -> TaskResult {
match handle_recipe_task(task.clone(), task_config, cancellation_token).await {
Ok(data) => TaskResult {
task_id: task.id.clone(),
status: TaskStatus::Completed,
data: Some(data),
error: None,
},
Err(error) => TaskResult {
task_id: task.id.clone(),
status: TaskStatus::Failed,
data: None,
error: Some(error),
},
}
}
async fn handle_recipe_task(
task: Task,
mut task_config: TaskConfig,
cancellation_token: CancellationToken,
) -> Result<Value, String> {
use crate::agents::subagent_handler::run_complete_subagent_task;
use crate::model::ModelConfig;
use crate::providers;
let recipe = task.payload.recipe;
let return_last_only = task.payload.return_last_only;
if let Some(ref exts) = recipe.extensions {
task_config.extensions = exts.clone();
}
if let Some(ref settings) = recipe.settings {
let new_provider = match (
&settings.goose_provider,
&settings.goose_model,
settings.temperature,
) {
(Some(provider), Some(model), temp) => {
let config = ModelConfig::new_or_fail(model).with_temperature(temp);
Some((provider.clone(), config))
}
(Some(_), None, _) => {
return Err("Recipe specifies provider but no model".to_string());
}
(None, model_or_temp, _)
if model_or_temp.is_some() || settings.temperature.is_some() =>
{
let provider_name = task_config.provider.get_name().to_string();
let mut config = task_config.provider.get_model_config();
if let Some(model) = &settings.goose_model {
config.model_name = model.clone();
}
if let Some(temp) = settings.temperature {
config = config.with_temperature(Some(temp));
}
Some((provider_name, config))
}
_ => None,
};
if let Some((provider_name, model_config)) = new_provider {
task_config.provider = providers::create(&provider_name, model_config)
.await
.map_err(|e| format!("Failed to create provider '{}': {}", provider_name, e))?;
}
}
tokio::select! {
result = run_complete_subagent_task(recipe, task_config, return_last_only, task.id.clone()) => {
result.map(|text| serde_json::json!({"result": text}))
.map_err(|e| format!("Recipe execution failed: {}", e))
}
_ = cancellation_token.cancelled() => {
Err("Task cancelled".to_string())
}
}
}
@@ -1,110 +0,0 @@
use anyhow::Result;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use crate::agents::subagent_execution_tool::task_types::Task;
#[derive(Debug, Clone)]
pub struct TasksManager {
tasks: Arc<RwLock<HashMap<String, Task>>>,
}
impl Default for TasksManager {
fn default() -> Self {
Self::new()
}
}
impl TasksManager {
pub fn new() -> Self {
Self {
tasks: Arc::new(RwLock::new(HashMap::new())),
}
}
pub async fn save_tasks(&self, tasks: Vec<Task>) {
let mut task_map = self.tasks.write().await;
for task in tasks {
task_map.insert(task.id.clone(), task);
}
}
pub async fn get_task(&self, task_id: &str) -> Option<Task> {
let tasks = self.tasks.read().await;
tasks.get(task_id).cloned()
}
pub async fn get_tasks(&self, task_ids: &[String]) -> Result<Vec<Task>, String> {
let mut tasks = Vec::new();
for task_id in task_ids {
match self.get_task(task_id).await {
Some(task) => tasks.push(task),
None => {
return Err(format!(
"Task with ID '{}' not found in TasksManager",
task_id
))
}
}
}
Ok(tasks)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::agents::subagent_execution_tool::task_types::TaskPayload;
use crate::recipe::Recipe;
fn create_test_task(id: &str, sub_recipe_name: &str) -> Task {
let recipe = Recipe::builder()
.version("1.0.0")
.title(sub_recipe_name)
.description("Test recipe")
.instructions("Test instructions")
.build()
.unwrap();
Task {
id: id.to_string(),
payload: TaskPayload {
recipe,
return_last_only: false,
sequential_when_repeated: false,
parameter_values: None,
},
}
}
#[tokio::test]
async fn test_save_and_get_task() {
let manager = TasksManager::new();
let tasks = vec![create_test_task("task1", "weather")];
manager.save_tasks(tasks).await;
let retrieved = manager.get_task("task1").await;
assert!(retrieved.is_some());
assert_eq!(retrieved.unwrap().id, "task1");
}
#[tokio::test]
async fn test_save_multiple_tasks() {
let manager = TasksManager::new();
let tasks = vec![
create_test_task("task1", "weather"),
create_test_task("task2", "news"),
];
manager.save_tasks(tasks).await;
let task1 = manager.get_task("task1").await;
let task2 = manager.get_task("task2").await;
assert!(task1.is_some());
assert!(task2.is_some());
assert_eq!(task1.unwrap().id, "task1");
assert_eq!(task2.unwrap().id, "task2");
}
}
@@ -1,57 +0,0 @@
use std::collections::HashMap;
use crate::agents::subagent_execution_tool::task_types::{TaskInfo, TaskStatus};
pub fn get_task_name(task_info: &TaskInfo) -> &str {
&task_info.task.payload.recipe.title
}
pub fn count_by_status(tasks: &HashMap<String, TaskInfo>) -> (usize, usize, usize, usize, usize) {
let total = tasks.len();
let (pending, running, completed, failed) = tasks.values().fold(
(0, 0, 0, 0),
|(pending, running, completed, failed), task| match task.status {
TaskStatus::Pending => (pending + 1, running, completed, failed),
TaskStatus::Running => (pending, running + 1, completed, failed),
TaskStatus::Completed => (pending, running, completed + 1, failed),
TaskStatus::Failed => (pending, running, completed, failed + 1),
},
);
(total, pending, running, completed, failed)
}
pub fn strip_ansi_codes(text: &str) -> String {
let mut result = String::new();
let mut chars = text.chars();
while let Some(ch) = chars.next() {
if ch == '\x1b' {
if let Some(next_ch) = chars.next() {
if next_ch == '[' {
// This is an ANSI escape sequence, consume until alphabetic character
loop {
match chars.next() {
Some(c) if c.is_ascii_alphabetic() => break,
Some(_) => continue,
None => break,
}
}
} else {
// Not an ANSI sequence, keep both characters
result.push(ch);
result.push(next_ch);
}
} else {
// End of string after \x1b
result.push(ch);
}
} else {
result.push(ch);
}
}
result
}
#[cfg(test)]
mod tests;
@@ -1,152 +0,0 @@
use crate::agents::subagent_execution_tool::task_types::{Task, TaskInfo, TaskPayload, TaskStatus};
use crate::agents::subagent_execution_tool::utils::{
count_by_status, get_task_name, strip_ansi_codes,
};
use crate::recipe::Recipe;
use std::collections::HashMap;
fn create_task_info_with_defaults(task: Task, status: TaskStatus) -> TaskInfo {
TaskInfo {
task,
status,
start_time: None,
end_time: None,
result: None,
current_output: String::new(),
}
}
mod test_get_task_name {
use super::*;
#[test]
fn test_extracts_recipe_title() {
let recipe = Recipe::builder()
.version("1.0.0")
.title("my_recipe")
.description("Test")
.instructions("do something")
.build()
.unwrap();
let task = Task {
id: "task_1".to_string(),
payload: TaskPayload {
recipe,
return_last_only: false,
sequential_when_repeated: false,
parameter_values: None,
},
};
let task_info = create_task_info_with_defaults(task, TaskStatus::Pending);
assert_eq!(get_task_name(&task_info), "my_recipe");
}
}
mod count_by_status {
use super::*;
fn create_test_task(id: &str, status: TaskStatus) -> TaskInfo {
let recipe = Recipe::builder()
.version("1.0.0")
.title("Test Recipe")
.description("Test")
.instructions("Test")
.build()
.unwrap();
let task = Task {
id: id.to_string(),
payload: TaskPayload {
recipe,
return_last_only: false,
sequential_when_repeated: false,
parameter_values: None,
},
};
create_task_info_with_defaults(task, status)
}
#[test]
fn counts_empty_map() {
let tasks = HashMap::new();
let (total, pending, running, completed, failed) = count_by_status(&tasks);
assert_eq!(
(total, pending, running, completed, failed),
(0, 0, 0, 0, 0)
);
}
#[test]
fn counts_single_status() {
let mut tasks = HashMap::new();
tasks.insert(
"task1".to_string(),
create_test_task("task1", TaskStatus::Pending),
);
tasks.insert(
"task2".to_string(),
create_test_task("task2", TaskStatus::Pending),
);
let (total, pending, running, completed, failed) = count_by_status(&tasks);
assert_eq!(
(total, pending, running, completed, failed),
(2, 2, 0, 0, 0)
);
}
#[test]
fn counts_mixed_statuses() {
let mut tasks = HashMap::new();
tasks.insert(
"task1".to_string(),
create_test_task("task1", TaskStatus::Pending),
);
tasks.insert(
"task2".to_string(),
create_test_task("task2", TaskStatus::Running),
);
tasks.insert(
"task3".to_string(),
create_test_task("task3", TaskStatus::Completed),
);
tasks.insert(
"task4".to_string(),
create_test_task("task4", TaskStatus::Failed),
);
tasks.insert(
"task5".to_string(),
create_test_task("task5", TaskStatus::Completed),
);
let (total, pending, running, completed, failed) = count_by_status(&tasks);
assert_eq!(
(total, pending, running, completed, failed),
(5, 1, 1, 2, 1)
);
}
}
mod strip_ansi_codes {
use super::*;
#[test]
fn test_strip_ansi_codes() {
assert_eq!(strip_ansi_codes("hello world"), "hello world");
assert_eq!(strip_ansi_codes("\x1b[31mred text\x1b[0m"), "red text");
assert_eq!(
strip_ansi_codes("\x1b[1;32mbold green\x1b[0m"),
"bold green"
);
assert_eq!(
strip_ansi_codes("normal\x1b[33myellow\x1b[0mnormal"),
"normalyellownormal"
);
assert_eq!(strip_ansi_codes("\x1bhello"), "\x1bhello");
assert_eq!(strip_ansi_codes("hello\x1b"), "hello\x1b");
assert_eq!(strip_ansi_codes(""), "");
}
}
@@ -1,57 +0,0 @@
use crate::agents::subagent_execution_tool::task_types::{SharedState, Task};
use crate::agents::subagent_execution_tool::tasks::process_task;
use crate::agents::subagent_task_config::TaskConfig;
use std::sync::Arc;
async fn receive_task(state: &SharedState) -> Option<Task> {
let mut receiver = state.task_receiver.lock().await;
receiver.recv().await
}
pub fn spawn_worker(
state: Arc<SharedState>,
worker_id: usize,
task_config: TaskConfig,
) -> tokio::task::JoinHandle<()> {
state.increment_active_workers();
tokio::spawn(async move {
worker_loop(state, worker_id, task_config).await;
})
}
async fn worker_loop(state: Arc<SharedState>, _worker_id: usize, task_config: TaskConfig) {
loop {
tokio::select! {
task_option = receive_task(&state) => {
match task_option {
Some(task) => {
state.task_execution_tracker.start_task(&task.id).await;
let result = process_task(
&task,
state.task_execution_tracker.clone(),
task_config.clone(),
state.cancellation_token.clone(),
)
.await;
if let Err(e) = state.result_sender.send(result).await {
// Only log error if not cancelled (channel close is expected during cancellation)
if !state.cancellation_token.is_cancelled() {
tracing::error!("Worker failed to send result: {}", e);
}
break;
}
}
None => break, // No more tasks
}
}
_ = state.cancellation_token.cancelled() => {
tracing::debug!("Worker cancelled");
break;
}
}
}
state.decrement_active_workers();
}
+47 -10
View File
@@ -2,15 +2,27 @@ use crate::{
agents::{subagent_task_config::TaskConfig, AgentEvent, SessionConfig},
conversation::{message::Message, Conversation},
execution::manager::AgentManager,
prompt_template::render_global_file,
recipe::Recipe,
};
use anyhow::{anyhow, Result};
use futures::StreamExt;
use rmcp::model::{ErrorCode, ErrorData};
use serde::Serialize;
use std::future::Future;
use std::pin::Pin;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info};
#[derive(Serialize)]
struct SubagentPromptContext {
max_turns: usize,
subagent_id: String,
task_instructions: String,
tool_count: usize,
available_tools: String,
}
type AgentMessagesFuture =
Pin<Box<dyn Future<Output = Result<(Conversation, Option<String>)>> + Send>>;
@@ -20,16 +32,18 @@ pub async fn run_complete_subagent_task(
task_config: TaskConfig,
return_last_only: bool,
session_id: String,
cancellation_token: Option<CancellationToken>,
) -> Result<String, anyhow::Error> {
let (messages, final_output) = get_agent_messages(recipe, task_config, session_id)
.await
.map_err(|e| {
ErrorData::new(
ErrorCode::INTERNAL_ERROR,
format!("Failed to execute task: {}", e),
None,
)
})?;
let (messages, final_output) =
get_agent_messages(recipe, task_config, session_id, cancellation_token)
.await
.map_err(|e| {
ErrorData::new(
ErrorCode::INTERNAL_ERROR,
format!("Failed to execute task: {}", e),
None,
)
})?;
if let Some(output) = final_output {
return Ok(output);
@@ -100,6 +114,7 @@ fn get_agent_messages(
recipe: Recipe,
task_config: TaskConfig,
session_id: String,
cancellation_token: Option<CancellationToken>,
) -> AgentMessagesFuture {
Box::pin(async move {
let text_instruction = recipe
@@ -137,6 +152,26 @@ fn get_agent_messages(
.apply_recipe_components(recipe.sub_recipes.clone(), recipe.response.clone(), true)
.await;
let tools = agent.list_tools(None).await;
let subagent_prompt = render_global_file(
"subagent_system.md",
&SubagentPromptContext {
max_turns: task_config
.max_turns
.expect("TaskConfig always sets max_turns"),
subagent_id: session_id.clone(),
task_instructions: text_instruction.clone(),
tool_count: tools.len(),
available_tools: tools
.iter()
.map(|t| t.name.to_string())
.collect::<Vec<_>>()
.join(", "),
},
)
.map_err(|e| anyhow!("Failed to render subagent system prompt: {}", e))?;
agent.override_system_prompt(subagent_prompt).await;
let user_message = Message::user().with_text(text_instruction);
let mut conversation = Conversation::new_unvalidated(vec![user_message.clone()]);
@@ -153,7 +188,9 @@ fn get_agent_messages(
};
let mut stream = crate::session_context::with_session_id(Some(session_id.clone()), async {
agent.reply(user_message, session_config, None).await
agent
.reply(user_message, session_config, cancellation_token)
.await
})
.await
.map_err(|e| anyhow!("Failed to get reply from agent: {}", e))?;
+541
View File
@@ -0,0 +1,541 @@
use std::borrow::Cow;
use std::collections::HashMap;
use std::path::PathBuf;
use anyhow::{anyhow, Result};
use futures::FutureExt;
use rmcp::model::{Content, ErrorCode, ErrorData, Tool};
use serde::Deserialize;
use serde_json::{json, Value};
use tokio_util::sync::CancellationToken;
use crate::agents::subagent_handler::run_complete_subagent_task;
use crate::agents::subagent_task_config::TaskConfig;
use crate::agents::tool_execution::ToolCallResult;
use crate::config::GooseMode;
use crate::providers;
use crate::recipe::build_recipe::build_recipe_from_template;
use crate::recipe::local_recipes::load_local_recipe_file;
use crate::recipe::{Recipe, SubRecipe};
use crate::session::SessionManager;
pub const SUBAGENT_TOOL_NAME: &str = "subagent";
const SUMMARY_INSTRUCTIONS: &str = r#"
Important: Your parent agent will only receive your final message as a summary of your work.
Make sure your last message provides a comprehensive summary of:
- What you were asked to do
- What actions you took
- The results or outcomes
- Any important findings or recommendations
Be concise but complete.
"#;
#[derive(Debug, Deserialize)]
pub struct SubagentParams {
pub instructions: Option<String>,
pub subrecipe: Option<String>,
pub parameters: Option<HashMap<String, Value>>,
pub extensions: Option<Vec<String>>,
pub settings: Option<SubagentSettings>,
#[serde(default = "default_summary")]
pub summary: bool,
}
fn default_summary() -> bool {
true
}
#[derive(Debug, Deserialize)]
pub struct SubagentSettings {
pub provider: Option<String>,
pub model: Option<String>,
pub temperature: Option<f32>,
}
pub fn create_subagent_tool(sub_recipes: &[SubRecipe]) -> Tool {
let description = build_tool_description(sub_recipes);
let schema = json!({
"type": "object",
"properties": {
"instructions": {
"type": "string",
"description": "Instructions for the subagent. Required for ad-hoc tasks. For predefined tasks, adds additional context."
},
"subrecipe": {
"type": "string",
"description": "Name of a predefined subrecipe to run."
},
"parameters": {
"type": "object",
"additionalProperties": true,
"description": "Parameters for the subrecipe. Only valid when 'subrecipe' is specified."
},
"extensions": {
"type": "array",
"items": {"type": "string"},
"description": "Extensions to enable. Omit to inherit all, empty array for none."
},
"settings": {
"type": "object",
"properties": {
"provider": {"type": "string", "description": "Override LLM provider"},
"model": {"type": "string", "description": "Override model"},
"temperature": {"type": "number", "description": "Override temperature"}
},
"description": "Override model/provider settings."
},
"summary": {
"type": "boolean",
"default": true,
"description": "If true (default), return only the subagent's final summary."
}
}
});
Tool::new(
SUBAGENT_TOOL_NAME,
description,
schema.as_object().unwrap().clone(),
)
}
fn build_tool_description(sub_recipes: &[SubRecipe]) -> String {
let mut desc = String::from(
"Delegate a task to a subagent that runs independently with its own context.\n\n\
Modes:\n\
1. Ad-hoc: Provide `instructions` for a custom task\n\
2. Predefined: Provide `subrecipe` name to run a predefined task\n\
3. Augmented: Provide both `subrecipe` and `instructions` to add context\n\n\
The subagent has access to the same tools as you by default. \
Use `extensions` to limit which extensions the subagent can use.\n\n\
For parallel execution, make multiple `subagent` tool calls in the same message.",
);
if !sub_recipes.is_empty() {
desc.push_str("\n\nAvailable subrecipes:");
for sr in sub_recipes {
let params_info = get_subrecipe_params_description(sr);
let sequential_hint = if sr.sequential_when_repeated {
" [run sequentially, not in parallel]"
} else {
""
};
desc.push_str(&format!(
"\n• {}{} - {}{}",
sr.name,
sequential_hint,
sr.description.as_deref().unwrap_or("No description"),
if params_info.is_empty() {
String::new()
} else {
format!(" (params: {})", params_info)
}
));
}
}
desc
}
fn get_subrecipe_params_description(sub_recipe: &SubRecipe) -> String {
match load_local_recipe_file(&sub_recipe.path) {
Ok(recipe_file) => match Recipe::from_content(&recipe_file.content) {
Ok(recipe) => {
if let Some(params) = recipe.parameters {
params
.iter()
.filter(|p| {
sub_recipe
.values
.as_ref()
.map(|v| !v.contains_key(&p.key))
.unwrap_or(true)
})
.map(|p| {
let req = match p.requirement {
crate::recipe::RecipeParameterRequirement::Required => "[required]",
_ => "[optional]",
};
format!("{} {}", p.key, req)
})
.collect::<Vec<_>>()
.join(", ")
} else {
String::new()
}
}
Err(_) => String::new(),
},
Err(_) => String::new(),
}
}
/// Note: SubRecipe.sequential_when_repeated is surfaced as a hint in the tool description
/// (e.g., "[run sequentially, not in parallel]") but not enforced. The LLM controls
/// sequencing by making sequential vs parallel tool calls.
pub fn handle_subagent_tool(
params: Value,
task_config: TaskConfig,
sub_recipes: HashMap<String, SubRecipe>,
working_dir: PathBuf,
cancellation_token: Option<CancellationToken>,
) -> ToolCallResult {
let parsed_params: SubagentParams = match serde_json::from_value(params) {
Ok(p) => p,
Err(e) => {
return ToolCallResult::from(Err(ErrorData {
code: ErrorCode::INVALID_PARAMS,
message: Cow::from(format!("Invalid parameters: {}", e)),
data: None,
}));
}
};
if parsed_params.instructions.is_none() && parsed_params.subrecipe.is_none() {
return ToolCallResult::from(Err(ErrorData {
code: ErrorCode::INVALID_PARAMS,
message: Cow::from("Must provide 'instructions' or 'subrecipe' (or both)"),
data: None,
}));
}
if parsed_params.parameters.is_some() && parsed_params.subrecipe.is_none() {
return ToolCallResult::from(Err(ErrorData {
code: ErrorCode::INVALID_PARAMS,
message: Cow::from("'parameters' can only be used with 'subrecipe'"),
data: None,
}));
}
let recipe = match build_recipe(&parsed_params, &sub_recipes) {
Ok(r) => r,
Err(e) => {
return ToolCallResult::from(Err(ErrorData {
code: ErrorCode::INVALID_PARAMS,
message: Cow::from(e.to_string()),
data: None,
}));
}
};
ToolCallResult {
notification_stream: None,
result: Box::new(
execute_subagent(
recipe,
task_config,
parsed_params,
working_dir,
cancellation_token,
)
.boxed(),
),
}
}
async fn execute_subagent(
recipe: Recipe,
task_config: TaskConfig,
params: SubagentParams,
working_dir: PathBuf,
cancellation_token: Option<CancellationToken>,
) -> Result<rmcp::model::CallToolResult, ErrorData> {
let session = SessionManager::create_session(
working_dir,
"Subagent task".to_string(),
crate::session::session_manager::SessionType::SubAgent,
)
.await
.map_err(|e| ErrorData {
code: ErrorCode::INTERNAL_ERROR,
message: Cow::from(format!("Failed to create session: {}", e)),
data: None,
})?;
let task_config = apply_settings_overrides(task_config, &params)
.await
.map_err(|e| ErrorData {
code: ErrorCode::INVALID_PARAMS,
message: Cow::from(e.to_string()),
data: None,
})?;
let result = run_complete_subagent_task(
recipe,
task_config,
params.summary,
session.id,
cancellation_token,
)
.await;
match result {
Ok(text) => Ok(rmcp::model::CallToolResult {
content: vec![Content::text(text)],
structured_content: None,
is_error: Some(false),
meta: None,
}),
Err(e) => Err(ErrorData {
code: ErrorCode::INTERNAL_ERROR,
message: Cow::from(e.to_string()),
data: None,
}),
}
}
fn build_recipe(
params: &SubagentParams,
sub_recipes: &HashMap<String, SubRecipe>,
) -> Result<Recipe> {
let mut recipe = if let Some(subrecipe_name) = &params.subrecipe {
build_subrecipe(subrecipe_name, params, sub_recipes)?
} else {
build_adhoc_recipe(params)?
};
if params.summary {
let current = recipe.instructions.unwrap_or_default();
recipe.instructions = Some(format!("{}\n{}", current, SUMMARY_INSTRUCTIONS));
}
Ok(recipe)
}
fn build_subrecipe(
subrecipe_name: &str,
params: &SubagentParams,
sub_recipes: &HashMap<String, SubRecipe>,
) -> Result<Recipe> {
let sub_recipe = sub_recipes.get(subrecipe_name).ok_or_else(|| {
let available: Vec<_> = sub_recipes.keys().cloned().collect();
anyhow!(
"Unknown subrecipe '{}'. Available: {}",
subrecipe_name,
available.join(", ")
)
})?;
let recipe_file = load_local_recipe_file(&sub_recipe.path)
.map_err(|e| anyhow!("Failed to load subrecipe '{}': {}", subrecipe_name, e))?;
let mut param_values: Vec<(String, String)> = Vec::new();
if let Some(values) = &sub_recipe.values {
for (k, v) in values {
param_values.push((k.clone(), v.clone()));
}
}
if let Some(provided_params) = &params.parameters {
for (k, v) in provided_params {
let value_str = match v {
Value::String(s) => s.clone(),
other => other.to_string(),
};
param_values.push((k.clone(), value_str));
}
}
let mut recipe = build_recipe_from_template(
recipe_file.content,
&recipe_file.parent_dir,
param_values,
None::<fn(&str, &str) -> Result<String, anyhow::Error>>,
)
.map_err(|e| anyhow!("Failed to build subrecipe: {}", e))?;
// Merge prompt into instructions so the subagent gets the actual task.
// The subagent handler uses `instructions` as the user message.
let mut combined = String::new();
if let Some(instructions) = &recipe.instructions {
combined.push_str(instructions);
}
if let Some(prompt) = &recipe.prompt {
if !combined.is_empty() {
combined.push_str("\n\n");
}
combined.push_str(prompt);
}
if let Some(extra_instructions) = &params.instructions {
if !combined.is_empty() {
combined.push_str("\n\n");
}
combined.push_str("Additional context from parent agent:\n");
combined.push_str(extra_instructions);
}
if !combined.is_empty() {
recipe.instructions = Some(combined);
}
Ok(recipe)
}
fn build_adhoc_recipe(params: &SubagentParams) -> Result<Recipe> {
let instructions = params
.instructions
.as_ref()
.ok_or_else(|| anyhow!("Instructions required for ad-hoc task"))?;
let recipe = Recipe::builder()
.version("1.0.0")
.title("Subagent Task")
.description("Ad-hoc subagent task")
.instructions(instructions)
.build()
.map_err(|e| anyhow!("Failed to build recipe: {}", e))?;
if recipe.check_for_security_warnings() {
return Err(anyhow!("Recipe contains potentially harmful content"));
}
Ok(recipe)
}
async fn apply_settings_overrides(
mut task_config: TaskConfig,
params: &SubagentParams,
) -> Result<TaskConfig> {
if let Some(settings) = &params.settings {
if settings.provider.is_some() || settings.model.is_some() || settings.temperature.is_some()
{
let provider_name = settings
.provider
.clone()
.unwrap_or_else(|| task_config.provider.get_name().to_string());
let mut model_config = task_config.provider.get_model_config();
if let Some(model) = &settings.model {
model_config.model_name = model.clone();
}
if let Some(temp) = settings.temperature {
model_config = model_config.with_temperature(Some(temp));
}
task_config.provider = providers::create(&provider_name, model_config)
.await
.map_err(|e| anyhow!("Failed to create provider '{}': {}", provider_name, e))?;
}
}
if let Some(extension_names) = &params.extensions {
if extension_names.is_empty() {
task_config.extensions = Vec::new();
} else {
task_config
.extensions
.retain(|ext| extension_names.contains(&ext.name()));
}
}
Ok(task_config)
}
pub fn should_enable_subagents(model_name: &str) -> bool {
let config = crate::config::Config::global();
let is_autonomous = config.get_goose_mode().unwrap_or(GooseMode::Auto) == GooseMode::Auto;
if !is_autonomous {
return false;
}
if model_name.starts_with("gemini") {
return false;
}
true
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_tool_name() {
assert_eq!(SUBAGENT_TOOL_NAME, "subagent");
}
#[test]
fn test_create_tool_without_subrecipes() {
let tool = create_subagent_tool(&[]);
assert_eq!(tool.name, "subagent");
assert!(tool.description.as_ref().unwrap().contains("Ad-hoc"));
assert!(!tool
.description
.as_ref()
.unwrap()
.contains("Available subrecipes"));
}
#[test]
fn test_create_tool_with_subrecipes() {
let sub_recipes = vec![SubRecipe {
name: "test_recipe".to_string(),
path: "test.yaml".to_string(),
values: None,
sequential_when_repeated: false,
description: Some("A test recipe".to_string()),
}];
let tool = create_subagent_tool(&sub_recipes);
assert!(tool
.description
.as_ref()
.unwrap()
.contains("Available subrecipes"));
assert!(tool.description.as_ref().unwrap().contains("test_recipe"));
}
#[test]
fn test_sequential_hint_in_description() {
let sub_recipes = vec![
SubRecipe {
name: "parallel_ok".to_string(),
path: "test.yaml".to_string(),
values: None,
sequential_when_repeated: false,
description: Some("Can run in parallel".to_string()),
},
SubRecipe {
name: "sequential_only".to_string(),
path: "test.yaml".to_string(),
values: None,
sequential_when_repeated: true,
description: Some("Must run sequentially".to_string()),
},
];
let tool = create_subagent_tool(&sub_recipes);
let desc = tool.description.as_ref().unwrap();
assert!(desc.contains("parallel_ok"));
assert!(!desc.contains("parallel_ok [run sequentially"));
assert!(desc.contains("sequential_only [run sequentially, not in parallel]"));
}
#[test]
fn test_params_deserialization_full() {
let params: SubagentParams = serde_json::from_value(json!({
"instructions": "Extra context",
"subrecipe": "my_recipe",
"parameters": {"key": "value"},
"extensions": ["developer"],
"settings": {"model": "gpt-4"},
"summary": false
}))
.unwrap();
assert_eq!(params.instructions, Some("Extra context".to_string()));
assert_eq!(params.subrecipe, Some("my_recipe".to_string()));
assert!(params.parameters.is_some());
assert_eq!(params.extensions, Some(vec!["developer".to_string()]));
assert!(!params.summary);
}
}
@@ -1,552 +0,0 @@
use goose::agents::recipe_tools::dynamic_task_tools::{
create_dynamic_task, task_params_to_inline_recipe,
};
use serde_json::json;
#[cfg(test)]
mod tests {
use super::*;
// Helper function to create a list of loaded extensions for testing
fn test_loaded_extensions() -> Vec<String> {
vec!["developer".to_string(), "memory".to_string()]
}
#[test]
fn test_minimal_task_with_instructions() {
let params = json!({
"instructions": "Test task"
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert_eq!(recipe.instructions, Some("Test task".to_string()));
assert_eq!(recipe.title, "Dynamic Task");
assert_eq!(recipe.description, "Inline recipe task");
}
#[test]
fn test_minimal_task_with_prompt() {
let params = json!({
"prompt": "Test prompt"
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert_eq!(recipe.prompt, Some("Test prompt".to_string()));
}
#[test]
fn test_missing_required_fields() {
let params = json!({
"title": "Test"
});
let result = task_params_to_inline_recipe(&params, &test_loaded_extensions());
assert!(result.is_err());
assert!(result
.unwrap_err()
.to_string()
.contains("instructions' or 'prompt"));
}
#[test]
fn test_with_recipe_fields() {
let params = json!({
"instructions": "Test",
"title": "Custom Title",
"description": "Custom Description",
"retry": {
"max_retries": 3,
"checks": [
{
"type": "shell",
"command": "echo test"
}
]
},
"response": {
"json_schema": {
"type": "object"
}
}
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert_eq!(recipe.title, "Custom Title");
assert_eq!(recipe.description, "Custom Description");
assert!(recipe.retry.is_some());
assert!(recipe.response.is_some());
// Verify retry config details
let retry = recipe.retry.unwrap();
assert_eq!(retry.max_retries, 3);
assert_eq!(retry.checks.len(), 1);
}
#[test]
fn test_security_validation() {
let params = json!({
"instructions": format!("Test{}", '\u{E0041}') // Harmful Unicode tag
});
let result = task_params_to_inline_recipe(&params, &test_loaded_extensions());
assert!(result.is_err());
assert!(result.unwrap_err().to_string().contains("harmful"));
}
#[tokio::test]
async fn test_create_multiple_tasks() {
use goose::agents::subagent_execution_tool::tasks_manager::TasksManager;
let tasks_manager = TasksManager::new();
let params = json!({
"task_parameters": [
{"instructions": "Task 1"},
{"prompt": "Task 2"}
]
});
let working_dir = std::path::Path::new("/tmp");
let result = create_dynamic_task(
params,
&tasks_manager,
test_loaded_extensions(),
working_dir,
)
.await;
// Check that the result is successful by awaiting the future
let tool_result = result.result.await;
assert!(tool_result.is_ok());
let contents = tool_result.unwrap();
assert!(!contents.content.is_empty());
// Parse the returned JSON to verify task creation
if let Some(text_content) = contents.content.first().and_then(|c| c.as_text()) {
let task_payload: serde_json::Value = serde_json::from_str(&text_content.text).unwrap();
assert!(task_payload.get("task_ids").is_some());
let task_ids = task_payload.get("task_ids").unwrap().as_array().unwrap();
assert_eq!(task_ids.len(), 2);
}
}
#[test]
fn test_return_last_only_flag() {
let params_with_flag = json!({
"instructions": "Test task",
"return_last_only": true
});
let recipe =
task_params_to_inline_recipe(&params_with_flag, &test_loaded_extensions()).unwrap();
assert_eq!(recipe.instructions, Some("Test task".to_string()));
// The flag should not affect the recipe itself, only the task payload
// We can't test the task creation here without async context
let params_without_flag = json!({
"instructions": "Test task"
});
let recipe2 =
task_params_to_inline_recipe(&params_without_flag, &test_loaded_extensions()).unwrap();
assert_eq!(recipe2.instructions, Some("Test task".to_string()));
}
#[tokio::test]
async fn test_text_instruction_not_supported() {
use goose::agents::subagent_execution_tool::tasks_manager::TasksManager;
let tasks_manager = TasksManager::new();
let params = json!({
"task_parameters": [
{"text_instruction": "Legacy task"}
]
});
let working_dir = std::path::Path::new("/tmp");
let result = create_dynamic_task(
params,
&tasks_manager,
test_loaded_extensions(),
working_dir,
)
.await;
// Check that the result fails since text_instruction is no longer supported
let tool_result = result.result.await;
assert!(tool_result.is_err());
// Verify the error message indicates missing required fields
if let Err(err) = tool_result {
let error_msg = err.message.to_string();
assert!(error_msg.contains("instructions") || error_msg.contains("prompt"));
}
}
#[test]
fn test_with_extensions() {
let params = json!({
"instructions": "Test",
"extensions": [
{
"type": "builtin",
"name": "developer",
"description": "developer"
}
]
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert!(recipe.extensions.is_some());
let extensions = recipe.extensions.unwrap();
assert_eq!(extensions.len(), 1);
}
#[test]
fn test_with_context_and_activities() {
let params = json!({
"instructions": "Test",
"context": ["context1", "context2"],
"activities": ["activity1", "activity2"]
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert!(recipe.activities.is_some());
assert_eq!(recipe.activities.unwrap(), vec!["activity1", "activity2"]);
}
#[test]
fn test_invalid_retry_config() {
// Test with max_retries = 0 (invalid)
let params = json!({
"instructions": "Test",
"retry": {
"max_retries": 0, // Invalid: must be > 0
"checks": []
}
});
let result = task_params_to_inline_recipe(&params, &test_loaded_extensions());
assert!(result.is_err());
assert!(result
.unwrap_err()
.to_string()
.contains("Invalid retry config"));
}
#[test]
fn test_invalid_retry_config_missing_checks() {
// Test with missing required field 'checks'
let params = json!({
"instructions": "Test",
"retry": {
"max_retries": 3
// Missing 'checks' field
}
});
let result = task_params_to_inline_recipe(&params, &test_loaded_extensions());
// This should fail during deserialization since 'checks' is required
assert!(result.is_ok()); // But retry field will be None due to failed deserialization
let recipe = result.unwrap();
assert!(recipe.retry.is_none());
}
// Additional edge case tests
#[test]
fn test_both_instructions_and_prompt() {
// Test that both instructions and prompt can be provided
let params = json!({
"instructions": "Test instructions",
"prompt": "Test prompt"
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert_eq!(recipe.instructions, Some("Test instructions".to_string()));
assert_eq!(recipe.prompt, Some("Test prompt".to_string()));
}
#[test]
fn test_empty_task_parameters_array() {
// This test is for the create_dynamic_task function
// We can't test it here without async, but we document the expected behavior
// Empty task_parameters array should return an error
}
#[test]
fn test_invalid_json_in_optional_fields() {
// Test that invalid JSON in optional fields is gracefully ignored
let params = json!({
"instructions": "Test",
"settings": "not an object", // Invalid: should be object
"extensions": "not an array", // Invalid: should be array
"context": {"not": "an array"}, // Invalid: should be array
"activities": 123 // Invalid: should be array
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert_eq!(recipe.instructions, Some("Test".to_string()));
// Invalid fields should be ignored (None)
assert!(recipe.settings.is_none());
assert!(recipe.extensions.is_none());
assert!(recipe.activities.is_none());
}
#[test]
fn test_with_settings() {
let params = json!({
"instructions": "Test",
"settings": {
"goose_provider": "openai",
"goose_model": "gpt-4",
"temperature": 0.7
}
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert!(recipe.settings.is_some());
let settings = recipe.settings.unwrap();
assert_eq!(settings.goose_provider, Some("openai".to_string()));
assert_eq!(settings.goose_model, Some("gpt-4".to_string()));
assert_eq!(settings.temperature, Some(0.7));
}
#[test]
fn test_with_parameters() {
let params = json!({
"instructions": "Test",
"parameters": [
{
"key": "test_param",
"input_type": "string",
"requirement": "required",
"description": "A test parameter"
}
]
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert!(recipe.parameters.is_some());
let parameters = recipe.parameters.unwrap();
assert_eq!(parameters.len(), 1);
assert_eq!(parameters[0].key, "test_param");
}
#[test]
fn test_empty_strings_for_required_fields() {
// Empty strings should be valid for instructions/prompt
let params = json!({
"instructions": ""
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert_eq!(recipe.instructions, Some("".to_string()));
}
#[test]
fn test_very_long_instruction() {
// Test with a very long instruction string
let long_instruction = "a".repeat(10000);
let params = json!({
"instructions": long_instruction.clone()
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert_eq!(recipe.instructions, Some(long_instruction));
}
#[tokio::test]
async fn test_mixed_valid_and_invalid_tasks() {
use goose::agents::subagent_execution_tool::tasks_manager::TasksManager;
let tasks_manager = TasksManager::new();
let params = json!({
"task_parameters": [
{"instructions": "Valid task"},
{"title": "Invalid - missing instruction"}, // This should cause error
]
});
let working_dir = std::path::Path::new("/tmp");
let result = create_dynamic_task(
params,
&tasks_manager,
test_loaded_extensions(),
working_dir,
)
.await;
// Should fail on the invalid task
let tool_result = result.result.await;
assert!(tool_result.is_err());
}
#[test]
fn test_unicode_in_non_instruction_fields() {
// Unicode tags should be allowed in non-instruction fields
let params = json!({
"instructions": "Test",
"title": format!("Title with unicode {}", '\u{E0041}'),
"description": format!("Description with unicode {}", '\u{E0041}')
});
// This should succeed - only instructions/prompt/activities are checked for security
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert!(recipe.title.contains('\u{E0041}'));
assert!(recipe.description.contains('\u{E0041}'));
}
#[test]
fn test_extension_shortnames() {
// Test that extension shortnames are properly resolved
// Note: This test now depends on actual config, so it may not find all extensions
// if they're not configured in the test environment
let loaded_exts = vec!["developer".to_string(), "memory".to_string()];
let params = json!({
"instructions": "Test",
"extensions": ["developer", "memory"]
});
let recipe = task_params_to_inline_recipe(&params, &loaded_exts).unwrap();
assert!(recipe.extensions.is_some());
let extensions = recipe.extensions.unwrap();
// We can't guarantee both extensions exist in config during tests
// Just check that we got some extensions and they have the right structure
assert!(extensions.len() <= 2);
if !extensions.is_empty() {
// Check that the first one is a valid ExtensionConfig
assert!(matches!(
&extensions[0],
goose::agents::extension::ExtensionConfig::Builtin { .. }
| goose::agents::extension::ExtensionConfig::Stdio { .. }
| goose::agents::extension::ExtensionConfig::Sse { .. }
| goose::agents::extension::ExtensionConfig::StreamableHttp { .. }
| goose::agents::extension::ExtensionConfig::Frontend { .. }
| goose::agents::extension::ExtensionConfig::InlinePython { .. }
));
}
}
#[test]
fn test_mixed_extension_formats() {
// Test mixing shortnames and full configs
// Note: Shortnames depend on config being present, which may not exist in CI
let loaded_exts = vec!["developer".to_string(), "memory".to_string()];
let params = json!({
"instructions": "Test",
"extensions": [
"developer", // Shortname - may not resolve in CI
{
"type": "stdio",
"name": "custom",
"description": "Custom stdio",
"cmd": "echo",
"args": ["test"]
}
]
});
let recipe = task_params_to_inline_recipe(&params, &loaded_exts).unwrap();
assert!(recipe.extensions.is_some());
let extensions = recipe.extensions.unwrap();
// At minimum we should get the full config (stdio), shortname may not resolve
assert!(!extensions.is_empty() && extensions.len() <= 2);
// The last one should always be the stdio config we provided
if let Some(last) = extensions.last() {
match last {
goose::agents::extension::ExtensionConfig::Stdio { name, .. } => {
assert_eq!(name, "custom");
}
_ => {
// If we got 2 extensions, the second should be stdio
if extensions.len() == 2 {
panic!("Expected stdio extension config for 'custom'");
}
}
}
}
}
#[test]
fn test_unknown_extension_shortname() {
// Test that unknown extension shortnames are skipped while valid configs are kept
let loaded_exts = vec!["developer".to_string()];
let params = json!({
"instructions": "Test",
"extensions": [
"unknown_extension_1", // Full config should always work
{
"type": "builtin",
"name": "test_builtin",
"display_name": "Test Builtin",
"description": "Test extension"
},
"unknown_extension_2" // Should be skipped
]
});
let recipe = task_params_to_inline_recipe(&params, &loaded_exts).unwrap();
assert!(recipe.extensions.is_some());
let extensions = recipe.extensions.unwrap();
// Should only get the full config, unknown shortnames should be skipped
assert_eq!(extensions.len(), 1);
// Verify it's the builtin we provided
match &extensions[0] {
goose::agents::extension::ExtensionConfig::Builtin { name, .. } => {
assert_eq!(name, "test_builtin");
}
_ => panic!("Expected builtin extension config"),
}
}
#[test]
fn test_empty_extensions_array() {
// Test that an empty extensions array results in no extensions
let loaded_exts = vec!["developer".to_string(), "memory".to_string()];
let params = json!({
"instructions": "Test",
"extensions": []
});
let recipe = task_params_to_inline_recipe(&params, &loaded_exts).unwrap();
assert!(recipe.extensions.is_some());
let extensions = recipe.extensions.unwrap();
// Empty array should mean no extensions
assert_eq!(extensions.len(), 0);
}
#[test]
fn test_omitted_extensions_field() {
// Test that omitting the extensions field results in None (use all)
let loaded_exts = vec!["developer".to_string(), "memory".to_string()];
let params = json!({
"instructions": "Test"
// No extensions field
});
let recipe = task_params_to_inline_recipe(&params, &loaded_exts).unwrap();
// When extensions field is omitted, recipe.extensions should be None
assert!(recipe.extensions.is_none());
}
#[test]
fn test_null_values_in_optional_fields() {
// Test that null values in optional fields are handled gracefully
let params = json!({
"instructions": "Test",
"title": null,
"description": null,
"extensions": null,
"settings": null
});
let recipe = task_params_to_inline_recipe(&params, &test_loaded_extensions()).unwrap();
assert_eq!(recipe.instructions, Some("Test".to_string()));
// Null values should use defaults or be None
assert_eq!(recipe.title, "Dynamic Task"); // Should use default
assert_eq!(recipe.description, "Inline recipe task"); // Should use default
assert!(recipe.extensions.is_none());
assert!(recipe.settings.is_none());
}
}
+104
View File
@@ -0,0 +1,104 @@
use goose::agents::subagent_tool::{create_subagent_tool, SUBAGENT_TOOL_NAME};
use goose::recipe::{Recipe, SubRecipe};
use std::collections::HashMap;
use tempfile::TempDir;
const RECIPE_TWO_PARAMS: &str = r#"
version: "1.0.0"
title: "Test Task"
description: "A test task"
instructions: "Process {{ first }} and {{ second }}"
parameters:
- key: first
input_type: string
requirement: required
description: "First param"
- key: second
input_type: string
requirement: required
description: "Second param"
"#;
fn write_recipe(temp_dir: &TempDir, name: &str, content: &str) -> String {
let path = temp_dir.path().join(format!("{}.yaml", name));
std::fs::write(&path, content).unwrap();
path.to_string_lossy().to_string()
}
fn make_subrecipe(path: String, name: &str, values: Option<HashMap<String, String>>) -> SubRecipe {
SubRecipe {
name: name.to_string(),
path,
values,
sequential_when_repeated: false,
description: Some(format!("{} description", name)),
}
}
#[test]
fn test_tool_description_includes_subrecipe_params_and_filters_presets() {
let temp_dir = TempDir::new().unwrap();
let path = write_recipe(&temp_dir, "mytask", RECIPE_TWO_PARAMS);
let no_presets = make_subrecipe(path.clone(), "mytask", None);
let tool = create_subagent_tool(&[no_presets]);
let desc = tool.description.as_ref().unwrap();
assert!(desc.contains("mytask"));
assert!(desc.contains("first [required]"));
assert!(desc.contains("second [required]"));
let mut preset = HashMap::new();
preset.insert("second".to_string(), "preset_value".to_string());
let with_presets = make_subrecipe(path, "deploy", Some(preset));
let tool = create_subagent_tool(&[with_presets]);
let params_section = tool
.description
.as_ref()
.unwrap()
.split("(params:")
.nth(1)
.unwrap_or("");
assert!(params_section.contains("first"));
assert!(!params_section.contains("second"));
}
#[test]
fn test_adhoc_recipe_builder_and_security_check() {
let recipe = Recipe::builder()
.version("1.0.0")
.title("Adhoc Task")
.description("An ad-hoc task")
.instructions("Do the thing")
.build()
.unwrap();
assert_eq!(recipe.title, "Adhoc Task");
assert_eq!(recipe.instructions.as_ref().unwrap(), "Do the thing");
assert!(!recipe.check_for_security_warnings());
}
#[test]
fn test_adhoc_tool_schema_properties() {
let tool = create_subagent_tool(&[]);
assert_eq!(tool.name, SUBAGENT_TOOL_NAME);
assert!(tool.description.as_ref().unwrap().contains("Ad-hoc"));
assert!(!tool
.description
.as_ref()
.unwrap()
.contains("Available subrecipes"));
let props = tool
.input_schema
.get("properties")
.unwrap()
.as_object()
.unwrap();
assert!(props.contains_key("instructions"));
assert!(props.contains_key("subrecipe"));
assert!(props.contains_key("parameters"));
assert!(props.contains_key("extensions"));
assert!(props.contains_key("settings"));
assert!(props.contains_key("summary"));
}
+10 -9
View File
@@ -83,10 +83,10 @@ instructions: |
### Phase 3: Subagent Testing (Meta-Recursion)
Create subagents to test yourself recursively:
- Basic subagent creation and execution
- Parallel subagent orchestration
- Parallel subagent execution (multiple subagent calls at once)
- Sequential subagent chains
- Recursive depth testing (subagent creating subagent)
- Test return_last_only optimization
- Test summary mode (default behavior for concise results)
### Phase 4: Advanced Self-Testing
Push boundaries and test limits:
@@ -175,23 +175,23 @@ prompt: |
## 🤖 PHASE 3: Subagent Meta-Testing
### Basic Subagent Test
Create a simple subagent task:
Use the `subagent` tool with instructions to create a simple task:
```
Task: "Create a file called subagent_test.txt with 'Hello from subagent'"
subagent(instructions: "Create a file called subagent_test.txt with 'Hello from subagent'")
```
### Parallel Subagent Test
{% if parallel_tests == "true" %}
Create 3 parallel subagents:
Create 3 subagent calls simultaneously (parallel execution):
1. Count files in current directory
2. Get current timestamp
3. Create a test file
Use execution_mode: "parallel" and verify all complete.
Make all three `subagent` tool calls at once to execute them in parallel.
{% endif %}
### Sequential Chain Test
Create dependent subagents:
Create dependent subagents (one after another):
1. First: Create a Python file
2. Second: Analyze the created file
3. Third: Run the Python file
@@ -202,8 +202,9 @@ prompt: |
Monitor for resource constraints and context window limits.
{% endif %}
### Return Last Only Test
Create subagents with return_last_only=true and verify condensed output.
### Summary Mode Test
Create subagents with summary mode (default) and verify concise output.
Test with `summary: false` to get full conversation history.
Log results to: {{ workspace_dir }}/phase3_subagents.md
{% endif %}
+8 -22
View File
@@ -77,29 +77,23 @@ check_recipe_output() {
local tmpfile=$1
local mode=$2
if grep -q "| subrecipe" "$tmpfile"; then
echo "✓ SUCCESS: Subrecipe tools invoked"
RESULTS+=("✓ Subrecipe tool invocation ($mode)")
# Check for unified subagent tool invocation (new format: "─── subagent |")
if grep -q "─── subagent" "$tmpfile"; then
echo "✓ SUCCESS: Subagent tool invoked"
RESULTS+=("✓ Subagent tool invocation ($mode)")
else
echo "✗ FAILED: No evidence of subrecipe tool invocation"
RESULTS+=("✗ Subrecipe tool invocation ($mode)")
echo "✗ FAILED: No evidence of subagent tool invocation"
RESULTS+=("✗ Subagent tool invocation ($mode)")
fi
if grep -q "file_stats" "$tmpfile" && grep -q "code_patterns" "$tmpfile"; then
# Check that both subrecipes were called (shown as "subrecipe: <name>" in output)
if grep -q "subrecipe:.*file_stats\|file_stats.*subrecipe" "$tmpfile" && grep -q "subrecipe:.*code_patterns\|code_patterns.*subrecipe" "$tmpfile"; then
echo "✓ SUCCESS: Both subrecipes (file_stats, code_patterns) found in output"
RESULTS+=("✓ Both subrecipes present ($mode)")
else
echo "✗ FAILED: Not all subrecipes found in output"
RESULTS+=("✗ Subrecipe names ($mode)")
fi
if grep -q "| subagent" "$tmpfile"; then
echo "✓ SUCCESS: Subagent execution detected"
RESULTS+=("✓ Subagent execution ($mode)")
else
echo "✗ FAILED: No evidence of subagent execution"
RESULTS+=("✗ Subagent execution ($mode)")
fi
}
echo "Running recipe with parallel subrecipes..."
@@ -108,14 +102,6 @@ if (cd "$TESTDIR" && "$SCRIPT_DIR/target/release/goose" run --recipe project_ana
echo "✓ SUCCESS: Recipe completed successfully"
RESULTS+=("✓ Recipe exit code")
check_recipe_output "$TMPFILE" "parallel"
if grep -q "execution_mode: parallel" "$TMPFILE"; then
echo "✓ SUCCESS: Parallel execution mode detected"
RESULTS+=("✓ Parallel execution mode")
else
echo "✗ FAILED: Parallel execution mode not detected"
RESULTS+=("✗ Parallel execution mode")
fi
else
echo "✗ FAILED: Recipe execution failed"
RESULTS+=("✗ Recipe exit code")