feat: run sub recipe multiple times in parallel (Experimental feature in CLI) #3274
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -31,6 +31,9 @@ ui/desktop/src/bin/goose_llm.dll | |
| # Hermit | ||
| .hermit/ | ||
|
|
||
| # Claude | ||
| .claude | ||
|
|
||
| debug_*.txt | ||
|
|
||
| # Docs | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,247 @@ | ||
| use goose::agents::sub_recipe_execution_tool::lib::TaskStatus; | ||
| use goose::agents::sub_recipe_execution_tool::notification_events::{ | ||
| TaskExecutionNotificationEvent, TaskInfo, | ||
| }; | ||
| use serde_json::Value; | ||
| use std::sync::atomic::{AtomicBool, Ordering}; | ||
|
|
||
| #[cfg(test)] | ||
| mod tests; | ||
|
|
||
| const CLEAR_SCREEN: &str = "\x1b[2J\x1b[H"; | ||
| const MOVE_TO_PROGRESS_LINE: &str = "\x1b[4;1H"; | ||
| const CLEAR_TO_EOL: &str = "\x1b[K"; | ||
| const CLEAR_BELOW: &str = "\x1b[J"; | ||
| pub const TASK_EXECUTION_NOTIFICATION_TYPE: &str = "task_execution"; | ||
|
|
||
| static INITIAL_SHOWN: AtomicBool = AtomicBool::new(false); | ||
|
|
||
| fn format_result_data_for_display(result_data: &Value) -> String { | ||
| match result_data { | ||
| Value::String(s) => strip_ansi_codes(s), | ||
| Value::Object(obj) => { | ||
| if let Some(partial_output) = obj.get("partial_output").and_then(|v| v.as_str()) { | ||
| format!("Partial output: {}", partial_output) | ||
| } else { | ||
| serde_json::to_string_pretty(obj).unwrap_or_default() | ||
| } | ||
| } | ||
| Value::Array(arr) => serde_json::to_string_pretty(arr).unwrap_or_default(), | ||
| Value::Bool(b) => b.to_string(), | ||
| Value::Number(n) => n.to_string(), | ||
| Value::Null => "null".to_string(), | ||
| } | ||
| } | ||
|
|
||
| fn process_output_for_display(output: &str) -> String { | ||
| const MAX_OUTPUT_LINES: usize = 2; | ||
| const OUTPUT_PREVIEW_LENGTH: usize = 100; | ||
|
|
||
| let lines: Vec<&str> = output.lines().collect(); | ||
| let recent_lines = if lines.len() > MAX_OUTPUT_LINES { | ||
| &lines[lines.len() - MAX_OUTPUT_LINES..] | ||
| } else { | ||
| &lines | ||
| }; | ||
|
|
||
| let clean_output = recent_lines.join(" ... "); | ||
| let stripped = strip_ansi_codes(&clean_output); | ||
| truncate_with_ellipsis(&stripped, OUTPUT_PREVIEW_LENGTH) | ||
| } | ||
|
|
||
| fn truncate_with_ellipsis(text: &str, max_len: usize) -> String { | ||
| if text.len() > max_len { | ||
| let mut end = max_len.saturating_sub(3); | ||
| while end > 0 && !text.is_char_boundary(end) { | ||
| end -= 1; | ||
| } | ||
| format!("{}...", &text[..end]) | ||
| } else { | ||
| text.to_string() | ||
| } | ||
| } | ||
|
|
||
| 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 | ||
| } | ||
|
|
||
| pub fn format_task_execution_notification( | ||
| data: &Value, | ||
| ) -> Option<(String, Option<String>, Option<String>)> { | ||
| if let Ok(event) = serde_json::from_value::<TaskExecutionNotificationEvent>(data.clone()) { | ||
| return Some(match event { | ||
| TaskExecutionNotificationEvent::LineOutput { output, .. } => ( | ||
| format!("{}\n", output), | ||
| None, | ||
| Some(TASK_EXECUTION_NOTIFICATION_TYPE.to_string()), | ||
| ), | ||
| TaskExecutionNotificationEvent::TasksUpdate { .. } => { | ||
| let formatted_display = format_tasks_update_from_event(&event); | ||
| ( | ||
| formatted_display, | ||
| None, | ||
| Some(TASK_EXECUTION_NOTIFICATION_TYPE.to_string()), | ||
| ) | ||
| } | ||
| TaskExecutionNotificationEvent::TasksComplete { .. } => { | ||
| let formatted_summary = format_tasks_complete_from_event(&event); | ||
| ( | ||
| formatted_summary, | ||
| None, | ||
| Some(TASK_EXECUTION_NOTIFICATION_TYPE.to_string()), | ||
| ) | ||
| } | ||
| }); | ||
| } | ||
| None | ||
| } | ||
|
|
||
| fn format_tasks_update_from_event(event: &TaskExecutionNotificationEvent) -> String { | ||
| if let TaskExecutionNotificationEvent::TasksUpdate { stats, tasks } = event { | ||
| let mut display = String::new(); | ||
|
|
||
| if !INITIAL_SHOWN.swap(true, Ordering::SeqCst) { | ||
| display.push_str(CLEAR_SCREEN); | ||
| display.push_str("🎯 Task Execution Dashboard\n"); | ||
| display.push_str("═══════════════════════════\n\n"); | ||
| } else { | ||
| display.push_str(MOVE_TO_PROGRESS_LINE); | ||
| } | ||
|
|
||
| display.push_str(&format!( | ||
| "📊 Progress: {} total | ⏳ {} pending | 🏃 {} running | ✅ {} completed | ❌ {} failed", | ||
| stats.total, stats.pending, stats.running, stats.completed, stats.failed | ||
| )); | ||
| display.push_str(&format!("{}\n\n", CLEAR_TO_EOL)); | ||
|
|
||
| let mut sorted_tasks = tasks.clone(); | ||
| sorted_tasks.sort_by(|a, b| a.id.cmp(&b.id)); | ||
|
|
||
| for task in sorted_tasks { | ||
| display.push_str(&format_task_display(&task)); | ||
| } | ||
|
|
||
| display.push_str(CLEAR_BELOW); | ||
| display | ||
| } else { | ||
| String::new() | ||
| } | ||
| } | ||
|
|
||
| fn format_tasks_complete_from_event(event: &TaskExecutionNotificationEvent) -> String { | ||
| if let TaskExecutionNotificationEvent::TasksComplete { | ||
| stats, | ||
| failed_tasks, | ||
| } = event | ||
| { | ||
| let mut summary = String::new(); | ||
| summary.push_str("Execution Complete!\n"); | ||
| summary.push_str("═══════════════════════\n"); | ||
|
|
||
| summary.push_str(&format!("Total Tasks: {}\n", stats.total)); | ||
| summary.push_str(&format!("✅ Completed: {}\n", stats.completed)); | ||
| summary.push_str(&format!("❌ Failed: {}\n", stats.failed)); | ||
| summary.push_str(&format!("📈 Success Rate: {:.1}%\n", stats.success_rate)); | ||
|
|
||
| if !failed_tasks.is_empty() { | ||
| summary.push_str("\n❌ Failed Tasks:\n"); | ||
| for task in failed_tasks { | ||
| summary.push_str(&format!(" • {}\n", task.name)); | ||
| if let Some(error) = &task.error { | ||
| summary.push_str(&format!(" Error: {}\n", error)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| summary.push_str("\n📝 Generating summary...\n"); | ||
| summary | ||
| } else { | ||
| String::new() | ||
| } | ||
| } | ||
|
|
||
| fn format_task_display(task: &TaskInfo) -> String { | ||
| let mut task_display = String::new(); | ||
|
|
||
| let status_icon = match task.status { | ||
| TaskStatus::Pending => "⏳", | ||
| TaskStatus::Running => "🏃", | ||
| TaskStatus::Completed => "✅", | ||
| TaskStatus::Failed => "❌", | ||
| }; | ||
|
|
||
| task_display.push_str(&format!( | ||
| "{} {} ({}){}\n", | ||
| status_icon, task.task_name, task.task_type, CLEAR_TO_EOL | ||
| )); | ||
|
|
||
| if !task.task_metadata.is_empty() { | ||
| task_display.push_str(&format!( | ||
| " 📋 Parameters: {}{}\n", | ||
| task.task_metadata, CLEAR_TO_EOL | ||
| )); | ||
| } | ||
|
|
||
| if let Some(duration_secs) = task.duration_secs { | ||
| task_display.push_str(&format!(" ⏱️ {:.1}s{}\n", duration_secs, CLEAR_TO_EOL)); | ||
| } | ||
|
|
||
| if matches!(task.status, TaskStatus::Running) && !task.current_output.trim().is_empty() { | ||
| let processed_output = process_output_for_display(&task.current_output); | ||
| if !processed_output.is_empty() { | ||
| task_display.push_str(&format!(" 💬 {}{}\n", processed_output, CLEAR_TO_EOL)); | ||
| } | ||
| } | ||
|
|
||
| if matches!(task.status, TaskStatus::Completed) { | ||
| if let Some(result_data) = &task.result_data { | ||
| let result_preview = format_result_data_for_display(result_data); | ||
| if !result_preview.is_empty() { | ||
| task_display.push_str(&format!(" 📄 {}{}\n", result_preview, CLEAR_TO_EOL)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if matches!(task.status, TaskStatus::Failed) { | ||
| if let Some(error) = &task.error { | ||
| let error_preview = truncate_with_ellipsis(error, 80); | ||
| task_display.push_str(&format!( | ||
| " ⚠️ {}{}\n", | ||
| error_preview.replace('\n', " "), | ||
| CLEAR_TO_EOL | ||
| )); | ||
| } | ||
| } | ||
|
|
||
| task_display.push_str(&format!("{}\n", CLEAR_TO_EOL)); | ||
| task_display | ||
| } | ||
|
Contributor
There was a problem hiding this comment. The rendering of subagents is something we should do an overall think about at some later stage as its a complicated experience to design. |
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This property name is a little confusing IMO, maybe use something like
allow_concurrent?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hi @jamadeo
Thank you for the review and your time!
I thought of
allow_concurrent, and i feel that it might have 2 concerns:allow_concurrent: falseit might confuse people that the sub-recipe cannot be running concurrently with other sub-recipes. The...when_repeatedmakes it clear that it cannot be run parallel when running repeatedly.By default the sub-recipe can be run in parallel. Usually if the boolean attribute is not specified, the default value is false. If
allow_concurrentis specified, people might be think theallow_concurrent: falseNaming is usually hard. Happy to change to a better name.