feat: run sub recipe multiple times in parallel (Experimental feature) (#3274)

Co-authored-by: Wendy Tang <wendytang@squareup.com>

crates/goose-cli/src/recipes/extract_from_cli.rs

@@ -32,6 +32,7 @@ pub fn extract_recipe_info_from_cli(
                         path: recipe_file_path.to_string_lossy().to_string(),
                         name,
                         values: None,
+                        sequential_when_repeated: true,
                     };
                     all_sub_recipes.push(additional_sub_recipe);
                 }

crates/goose-cli/src/session/mod.rs

@@ -4,8 +4,11 @@ mod export;
 mod input;
 mod output;
 mod prompt;
+mod task_execution_display;
 mod thinking;
 
+use crate::session::task_execution_display::TASK_EXECUTION_NOTIFICATION_TYPE;
+
 pub use self::export::message_to_markdown;
 pub use builder::{build_session, SessionBuilderConfig, SessionSettings};
 use console::Color;
@@ -17,6 +20,8 @@ use goose::permission::PermissionConfirmation;
 use goose::providers::base::Provider;
 pub use goose::session::Identifier;
 use goose::utils::safe_truncate;
+use std::io::Write;
+use task_execution_display::format_task_execution_notification;
 
 use anyhow::{Context, Result};
 use completion::GooseCompleter;
@@ -1008,7 +1013,7 @@ impl Session {
                                 match method.as_str() {
                                     "notifications/message" => {
                                         let data = o.get("data").unwrap_or(&Value::Null);
-                                        let (formatted_message, subagent_id, _notification_type) = match data {
+                                        let (formatted_message, subagent_id, message_notification_type) = match data {
                                             Value::String(s) => (s.clone(), None, None),
                                             Value::Object(o) => {
                                                 // Check for subagent notification structure first
@@ -1059,6 +1064,8 @@ impl Session {
                                                 } else if let Some(Value::String(output)) = o.get("output") {
                                                     // Fallback for other MCP notification types
                                                     (output.to_owned(), None, None)
+                                                } else if let Some(result) = format_task_execution_notification(data) {
+                                                    result
                                                 } else {
                                                     (data.to_string(), None, None)
                                                 }
@@ -1077,7 +1084,19 @@ impl Session {
                                             } else {
                                                 progress_bars.log(&formatted_message);
                                             }
-                                        } else {
+                                        } else if let Some(ref notification_type) = message_notification_type {
+                                            if notification_type == TASK_EXECUTION_NOTIFICATION_TYPE {
+                                                if interactive {
+                                                    let _ = progress_bars.hide();
+                                                    print!("{}", formatted_message);
+                                                    std::io::stdout().flush().unwrap();
+                                                } else {
+                                                    print!("{}", formatted_message);
+                                                    std::io::stdout().flush().unwrap();
+                                                }
+                                            }
+                                        }
+                                        else {
                                             // Non-subagent notification, display immediately with compact spacing
                                             if interactive {
                                                 let _ = progress_bars.hide();

crates/goose-cli/src/session/task_execution_display/mod.rs

@@ -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
+}

crates/goose-cli/src/session/task_execution_display/tests.rs

@@ -0,0 +1,337 @@
+use super::*;
+use goose::agents::sub_recipe_execution_tool::notification_events::{
+    FailedTaskInfo, TaskCompletionStats, TaskExecutionStats,
+};
+use serde_json::json;
+
+#[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(""), "");
+}
+
+#[test]
+fn test_truncate_with_ellipsis() {
+    assert_eq!(truncate_with_ellipsis("hello", 10), "hello");
+    assert_eq!(truncate_with_ellipsis("hello", 5), "hello");
+    assert_eq!(truncate_with_ellipsis("hello world", 8), "hello...");
+    assert_eq!(truncate_with_ellipsis("hello", 3), "...");
+    assert_eq!(truncate_with_ellipsis("hello", 2), "...");
+    assert_eq!(truncate_with_ellipsis("hello", 1), "...");
+    assert_eq!(truncate_with_ellipsis("", 5), "");
+}
+
+#[test]
+fn test_process_output_for_display() {
+    assert_eq!(process_output_for_display("hello world"), "hello world");
+    assert_eq!(
+        process_output_for_display("line1\nline2"),
+        "line1 ... line2"
+    );
+
+    let input = "line1\nline2\nline3\nline4";
+    let result = process_output_for_display(input);
+    assert_eq!(result, "line3 ... line4");
+
+    let long_line = "a".repeat(150);
+    let result = process_output_for_display(&long_line);
+    assert!(result.len() <= 100);
+    assert!(result.ends_with("..."));
+
+    let ansi_output = "\x1b[31mred line 1\x1b[0m\n\x1b[32mgreen line 2\x1b[0m";
+    let result = process_output_for_display(ansi_output);
+    assert_eq!(result, "red line 1 ... green line 2");
+
+    assert_eq!(process_output_for_display(""), "");
+}
+
+#[test]
+fn test_format_result_data_for_display() {
+    let string_val = json!("hello world");
+    assert_eq!(format_result_data_for_display(&string_val), "hello world");
+
+    let ansi_string = json!("\x1b[31mred text\x1b[0m");
+    assert_eq!(format_result_data_for_display(&ansi_string), "red text");
+
+    assert_eq!(format_result_data_for_display(&json!(true)), "true");
+    assert_eq!(format_result_data_for_display(&json!(false)), "false");
+    assert_eq!(format_result_data_for_display(&json!(42)), "42");
+    assert_eq!(format_result_data_for_display(&json!(3.14)), "3.14");
+    assert_eq!(format_result_data_for_display(&json!(null)), "null");
+
+    let partial_obj = json!({
+        "partial_output": "some output",
+        "other_field": "ignored"
+    });
+    assert_eq!(
+        format_result_data_for_display(&partial_obj),
+        "Partial output: some output"
+    );
+
+    let obj = json!({"key": "value", "num": 42});
+    let result = format_result_data_for_display(&obj);
+    assert!(result.contains("key"));
+    assert!(result.contains("value"));
+
+    let arr = json!([1, 2, 3]);
+    let result = format_result_data_for_display(&arr);
+    assert!(result.contains("1"));
+    assert!(result.contains("2"));
+    assert!(result.contains("3"));
+}
+
+#[test]
+fn test_format_task_execution_notification_line_output() {
+    let _event = TaskExecutionNotificationEvent::LineOutput {
+        task_id: "task-1".to_string(),
+        output: "Hello World".to_string(),
+    };
+
+    let data = json!({
+        "subtype": "line_output",
+        "task_id": "task-1",
+        "output": "Hello World"
+    });
+
+    let result = format_task_execution_notification(&data);
+    assert!(result.is_some());
+
+    let (formatted, second, third) = result.unwrap();
+    assert_eq!(formatted, "Hello World\n");
+    assert_eq!(second, None);
+    assert_eq!(third, Some("task_execution".to_string()));
+}
+
+#[test]
+fn test_format_task_execution_notification_invalid_data() {
+    let invalid_data = json!({
+        "invalid": "structure"
+    });
+
+    let result = format_task_execution_notification(&invalid_data);
+    assert_eq!(result, None);
+
+    let incomplete_data = json!({
+        "subtype": "line_output"
+    });
+
+    let result = format_task_execution_notification(&incomplete_data);
+    assert_eq!(result, None);
+}
+
+#[test]
+fn test_format_tasks_update_from_event() {
+    INITIAL_SHOWN.store(false, Ordering::SeqCst);
+
+    let stats = TaskExecutionStats::new(3, 1, 1, 1, 0);
+    let tasks = vec![
+        TaskInfo {
+            id: "task-1".to_string(),
+            status: TaskStatus::Running,
+            duration_secs: Some(1.5),
+            current_output: "Processing...".to_string(),
+            task_type: "sub_recipe".to_string(),
+            task_name: "test-task".to_string(),
+            task_metadata: "param=value".to_string(),
+            error: None,
+            result_data: None,
+        },
+        TaskInfo {
+            id: "task-2".to_string(),
+            status: TaskStatus::Completed,
+            duration_secs: Some(2.3),
+            current_output: "".to_string(),
+            task_type: "text_instruction".to_string(),
+            task_name: "another-task".to_string(),
+            task_metadata: "".to_string(),
+            error: None,
+            result_data: Some(json!({"result": "success"})),
+        },
+    ];
+
+    let event = TaskExecutionNotificationEvent::TasksUpdate { stats, tasks };
+    let result = format_tasks_update_from_event(&event);
+
+    assert!(result.contains("🎯 Task Execution Dashboard"));
+    assert!(result.contains("═══════════════════════════"));
+    assert!(result.contains("📊 Progress: 3 total"));
+    assert!(result.contains("⏳ 1 pending"));
+    assert!(result.contains("🏃 1 running"));
+    assert!(result.contains("✅ 1 completed"));
+    assert!(result.contains("❌ 0 failed"));
+    assert!(result.contains("🏃 test-task"));
+    assert!(result.contains("✅ another-task"));
+    assert!(result.contains("📋 Parameters: param=value"));
+    assert!(result.contains("⏱️  1.5s"));
+    assert!(result.contains("💬 Processing..."));
+
+    let result2 = format_tasks_update_from_event(&event);
+    assert!(!result2.contains("🎯 Task Execution Dashboard"));
+    assert!(result2.contains(MOVE_TO_PROGRESS_LINE));
+}
+
+#[test]
+fn test_format_tasks_complete_from_event() {
+    let stats = TaskCompletionStats::new(5, 4, 1);
+    let failed_tasks = vec![FailedTaskInfo {
+        id: "task-3".to_string(),
+        name: "failed-task".to_string(),
+        error: Some("Connection timeout".to_string()),
+    }];
+
+    let event = TaskExecutionNotificationEvent::TasksComplete {
+        stats,
+        failed_tasks,
+    };
+    let result = format_tasks_complete_from_event(&event);
+
+    assert!(result.contains("Execution Complete!"));
+    assert!(result.contains("═══════════════════════"));
+    assert!(result.contains("Total Tasks: 5"));
+    assert!(result.contains("✅ Completed: 4"));
+    assert!(result.contains("❌ Failed: 1"));
+    assert!(result.contains("📈 Success Rate: 80.0%"));
+    assert!(result.contains("❌ Failed Tasks:"));
+    assert!(result.contains("• failed-task"));
+    assert!(result.contains("Error: Connection timeout"));
+    assert!(result.contains("📝 Generating summary..."));
+}
+
+#[test]
+fn test_format_tasks_complete_from_event_no_failures() {
+    let stats = TaskCompletionStats::new(3, 3, 0);
+    let failed_tasks = vec![];
+
+    let event = TaskExecutionNotificationEvent::TasksComplete {
+        stats,
+        failed_tasks,
+    };
+    let result = format_tasks_complete_from_event(&event);
+
+    assert!(!result.contains("❌ Failed Tasks:"));
+    assert!(result.contains("📈 Success Rate: 100.0%"));
+    assert!(result.contains("❌ Failed: 0"));
+}
+
+#[test]
+fn test_format_task_display_running() {
+    let task = TaskInfo {
+        id: "task-1".to_string(),
+        status: TaskStatus::Running,
+        duration_secs: Some(1.5),
+        current_output: "Processing data...\nAlmost done...".to_string(),
+        task_type: "sub_recipe".to_string(),
+        task_name: "data-processor".to_string(),
+        task_metadata: "input=file.txt,output=result.json".to_string(),
+        error: None,
+        result_data: None,
+    };
+
+    let result = format_task_display(&task);
+
+    assert!(result.contains("🏃 data-processor (sub_recipe)"));
+    assert!(result.contains("📋 Parameters: input=file.txt,output=result.json"));
+    assert!(result.contains("⏱️  1.5s"));
+    assert!(result.contains("💬 Processing data... ... Almost done..."));
+}
+
+#[test]
+fn test_format_task_display_completed() {
+    let task = TaskInfo {
+        id: "task-2".to_string(),
+        status: TaskStatus::Completed,
+        duration_secs: Some(3.2),
+        current_output: "".to_string(),
+        task_type: "text_instruction".to_string(),
+        task_name: "analyzer".to_string(),
+        task_metadata: "".to_string(),
+        error: None,
+        result_data: Some(json!({"status": "success", "count": 42})),
+    };
+
+    let result = format_task_display(&task);
+
+    assert!(result.contains("✅ analyzer (text_instruction)"));
+    assert!(result.contains("⏱️  3.2s"));
+    assert!(!result.contains("📋 Parameters"));
+    assert!(result.contains("📄"));
+}
+
+#[test]
+fn test_format_task_display_failed() {
+    let task = TaskInfo {
+        id: "task-3".to_string(),
+        status: TaskStatus::Failed,
+        duration_secs: None,
+        current_output: "".to_string(),
+        task_type: "sub_recipe".to_string(),
+        task_name: "failing-task".to_string(),
+        task_metadata: "".to_string(),
+        error: Some(
+            "Network connection failed after multiple retries. The server is unreachable."
+                .to_string(),
+        ),
+        result_data: None,
+    };
+
+    let result = format_task_display(&task);
+
+    assert!(result.contains("❌ failing-task (sub_recipe)"));
+    assert!(!result.contains("⏱️"));
+    assert!(result.contains("⚠️"));
+    assert!(result.contains("Network connection failed after multiple retries"));
+}
+
+#[test]
+fn test_format_task_display_pending() {
+    let task = TaskInfo {
+        id: "task-4".to_string(),
+        status: TaskStatus::Pending,
+        duration_secs: None,
+        current_output: "".to_string(),
+        task_type: "sub_recipe".to_string(),
+        task_name: "waiting-task".to_string(),
+        task_metadata: "priority=high".to_string(),
+        error: None,
+        result_data: None,
+    };
+
+    let result = format_task_display(&task);
+
+    assert!(result.contains("⏳ waiting-task (sub_recipe)"));
+    assert!(result.contains("📋 Parameters: priority=high"));
+    assert!(!result.contains("⏱️"));
+    assert!(!result.contains("💬"));
+    assert!(!result.contains("📄"));
+    assert!(!result.contains("⚠️"));
+}
+
+#[test]
+fn test_format_task_display_empty_current_output() {
+    let task = TaskInfo {
+        id: "task-5".to_string(),
+        status: TaskStatus::Running,
+        duration_secs: Some(0.5),
+        current_output: "   \n\t  \n   ".to_string(),
+        task_type: "sub_recipe".to_string(),
+        task_name: "quiet-task".to_string(),
+        task_metadata: "".to_string(),
+        error: None,
+        result_data: None,
+    };
+
+    let result = format_task_display(&task);
+
+    assert!(!result.contains("💬"));
+}