Skip to content

fix: clean up subagent #3565

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wendytang merged 17 commits into from
Jul 24, 2025
Merged

fix: clean up subagent #3565

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
408afb9
clean up
wendytang Jul 21, 2025
a7b1c23
fn
wendytang Jul 21, 2025
7619c40
unwrap
wendytang Jul 21, 2025
a633062
safe truncate, fmt
wendytang Jul 22, 2025
8b232a1
rm subagent simple
wendytang Jul 22, 2025
5fae8b4
Merge branch 'main' into wtang/subagent_cleanup
wendytang Jul 22, 2025
6c7c762
rmcp JsonRpcMessage
wendytang Jul 22, 2025
e1e4955
fix tests
wendytang Jul 23, 2025
7b2416a
address comments
wendytang Jul 23, 2025
e1bac06
remove truncate
wendytang Jul 23, 2025
a14fd7e
Merge branch 'main' into wtang/subagent_cleanup
wendytang Jul 23, 2025
6735bb0
feat: subagent remove mcp notifications (#3621)
wendytang Jul 23, 2025
e4b28b7
Merge branch 'main' into wtang/subagent_cleanup
wendytang Jul 23, 2025
9ed7cc9
rm tool error
wendytang Jul 23, 2025
a2be6d2
rm content parsing jibberish, fmt
wendytang Jul 24, 2025
da48aee
keep it simple
wendytang Jul 24, 2025
2e6d677
rm debugs
wendytang Jul 24, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 34 additions & 20 deletions crates/goose-cli/src/session/output.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,7 @@ fn render_tool_request(req: &ToolRequest, theme: Theme, debug: bool) {
Ok(call) => match call.name.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),
_ => render_default_request(call, debug),
},
Err(e) => print_markdown(&e.to_string(), theme),
Expand Down Expand Up @@ -392,6 +393,37 @@ fn render_shell_request(call: &ToolCall, debug: bool) {
}
}

fn render_dynamic_task_request(call: &ToolCall, debug: bool) {
print_tool_header(call);

// Print task_parameters array
if let Some(Value::Array(task_parameters)) = call.arguments.get("task_parameters") {
println!("{}:", style("task_parameters").dim());

for task_param in task_parameters.iter() {
println!(" -");

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());
}
_ => {
// For everything else, use print_params
print!(" ");
print_params(value, 0, debug);
}
}
}
}
}
}

println!();
}

fn render_default_request(call: &ToolCall, debug: bool) {
print_tool_header(call);
print_params(&call.arguments, 0, debug);
Expand Down Expand Up @@ -463,26 +495,8 @@ fn print_params(value: &Value, depth: usize, debug: bool) {
}
}
Value::String(s) => {
// Special handling for text_instruction to show more content
let max_length = if key == "text_instruction" {
200 // Allow longer display for text instructions
} else {
get_tool_params_max_length()
};

if !debug && s.len() > max_length {
// For text instructions, show a preview instead of just "..."
if key == "text_instruction" {
let preview = &s[..max_length.saturating_sub(3)];
println!(
"{}{}: {}",
indent,
style(key).dim(),
style(format!("{}...", preview)).green()
);
} else {
println!("{}{}: {}", indent, style(key).dim(), style("...").dim());
}
if !debug && s.len() > get_tool_params_max_length() {
println!("{}{}: {}", indent, style(key).dim(), style("...").dim());
} else {
println!("{}{}: {}", indent, style(key).dim(), style(s).green());
}
Expand Down
40 changes: 1 addition & 39 deletions crates/goose/src/agents/agent.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,8 +75,6 @@ pub struct Agent {
pub(super) tool_monitor: Arc<Mutex<Option<ToolMonitor>>>,
pub(super) router_tool_selector: Mutex<Option<Arc<Box<dyn RouterToolSelector>>>>,
pub(super) scheduler_service: Mutex<Option<Arc<dyn SchedulerTrait>>>,
pub(super) mcp_tx: Mutex<mpsc::Sender<JsonRpcMessage>>,
pub(super) mcp_notification_rx: Arc<Mutex<mpsc::Receiver<JsonRpcMessage>>>,
pub(super) retry_manager: RetryManager,
}

Expand Down Expand Up @@ -132,8 +130,6 @@ impl Agent {
// Create channels with buffer size 32 (adjust if needed)
let (confirm_tx, confirm_rx) = mpsc::channel(32);
let (tool_tx, tool_rx) = mpsc::channel(32);
// Add MCP notification channel
let (mcp_tx, mcp_rx) = mpsc::channel(100);

let tool_monitor = Arc::new(Mutex::new(None));
let retry_manager = RetryManager::with_tool_monitor(tool_monitor.clone());
Expand All @@ -154,9 +150,6 @@ impl Agent {
tool_monitor,
router_tool_selector: Mutex::new(None),
scheduler_service: Mutex::new(None),
// Initialize with MCP notification support
mcp_tx: Mutex::new(mcp_tx),
mcp_notification_rx: Arc::new(Mutex::new(mcp_rx)),
retry_manager,
}
}
Expand Down Expand Up @@ -342,9 +335,8 @@ impl Agent {
.await
} else if tool_call.name == SUBAGENT_EXECUTE_TASK_TOOL_NAME {
let provider = self.provider().await.ok();
let mcp_tx = self.mcp_tx.lock().await.clone();

let task_config = TaskConfig::new(provider, mcp_tx);
let task_config = TaskConfig::new(provider);
subagent_execute_task_tool::run_tasks(
tool_call.arguments.clone(),
task_config,
Expand Down Expand Up @@ -771,24 +763,6 @@ impl Agent {
break;
}

// Handle MCP notifications from subagents
let mcp_notifications = self.get_mcp_notifications().await;
for notification in mcp_notifications {
if let JsonRpcMessage::Notification(notif) = &notification {
if let Some(data) = notif.notification.params.get("data") {
if let (Some(subagent_id), Some(_message)) = (
data.get("subagent_id").and_then(|v| v.as_str()),
data.get("message").and_then(|v| v.as_str()),
) {
yield AgentEvent::McpNotification((
subagent_id.to_string(),
notification.clone(),
));
}
}
}
}

let mut stream = Self::stream_response_from_provider(
self.provider().await?,
&system_prompt,
Expand Down Expand Up @@ -1085,18 +1059,6 @@ impl Agent {
prompt_manager.add_system_prompt_extra(instruction);
}

/// Get MCP notifications from subagents
pub async fn get_mcp_notifications(&self) -> Vec<JsonRpcMessage> {
let mut notifications = Vec::new();
let mut rx = self.mcp_notification_rx.lock().await;

while let Ok(notification) = rx.try_recv() {
notifications.push(notification);
}

notifications
}

pub async fn update_provider(&self, provider: Arc<dyn Provider>) -> Result<()> {
let mut current_provider = self.provider.lock().await;
*current_provider = Some(provider.clone());
Expand Down
8 changes: 0 additions & 8 deletions crates/goose/src/agents/recipe_tools/dynamic_task_tools.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,8 @@ pub fn create_dynamic_task_tool() -> Tool {
text_instruction: Search for the config file in the root directory.
Examples of 'task_parameters' for multiple tasks:
text_instruction: Get weather for Melbourne.
timeout_seconds: 300
text_instruction: Get weather for Los Angeles.
timeout_seconds: 300
text_instruction: Get weather for San Francisco.
timeout_seconds: 300
".to_string(),
json!({
"type": "object",
Expand All @@ -54,11 +51,6 @@ pub fn create_dynamic_task_tool() -> Tool {
"type": "string",
"description": "The text instruction to execute"
},
"timeout_seconds": {
"type": "integer",
"description": "Optional timeout for the task in seconds (default: 300)",
"minimum": 1
}
},
"required": ["text_instruction"]
}
Expand Down
123 changes: 20 additions & 103 deletions crates/goose/src/agents/subagent.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,6 @@ use crate::{
use anyhow::anyhow;
use chrono::{DateTime, Utc};
use mcp_core::{handler::ToolError, tool::Tool};
use rmcp::model::{JsonRpcMessage, JsonRpcNotification, JsonRpcVersion2_0, Notification};
use rmcp::object;
use serde::{Deserialize, Serialize};
// use serde_json::{self};
use std::{collections::HashMap, sync::Arc};
Expand Down Expand Up @@ -52,9 +50,7 @@ pub struct SubAgent {
impl SubAgent {
/// Create a new subagent with the given configuration and provider
#[instrument(skip(task_config))]
pub async fn new(
task_config: TaskConfig,
) -> Result<(Arc<Self>, tokio::task::JoinHandle<()>), anyhow::Error> {
pub async fn new(task_config: TaskConfig) -> Result<Arc<Self>, anyhow::Error> {
debug!("Creating new subagent with id: {}", task_config.id);

// Create a new extension manager for this subagent
Expand Down Expand Up @@ -90,21 +86,8 @@ impl SubAgent {
extension_manager: Arc::new(RwLock::new(extension_manager)),
});

// Send initial MCP notification
let subagent_clone = Arc::clone(&subagent);
subagent_clone
.send_mcp_notification("subagent_created", "Subagent created and ready")
.await;

// Create a background task handle (for future use with streaming/monitoring)
let subagent_clone = Arc::clone(&subagent);
let handle = tokio::spawn(async move {
// This could be used for background monitoring, cleanup, etc.
debug!("Subagent {} background task started", subagent_clone.id);
});

debug!("Subagent {} created successfully", subagent.id);
Ok((subagent, handle))
Ok(subagent)
}

/// Get the current status of the subagent
Expand All @@ -119,51 +102,6 @@ impl SubAgent {
let mut current_status = self.status.write().await;
*current_status = status.clone();
} // Write lock is released here!

// Send MCP notifications based on status
match &status {
SubAgentStatus::Processing => {
self.send_mcp_notification("status_changed", "Processing request")
.await;
}
SubAgentStatus::Completed(msg) => {
self.send_mcp_notification("completed", &format!("Completed: {}", msg))
.await;
}
SubAgentStatus::Terminated => {
self.send_mcp_notification("terminated", "Subagent terminated")
.await;
}
_ => {}
}
}

/// Send an MCP notification about the subagent's activity
pub async fn send_mcp_notification(&self, notification_type: &str, message: &str) {
let notification = JsonRpcMessage::Notification(JsonRpcNotification {
jsonrpc: JsonRpcVersion2_0,
notification: Notification {
method: "notifications/message".to_string(),
params: object!({
"level": "info",
"logger": format!("subagent_{}", self.id),
"data": {
"subagent_id": self.id,
"type": notification_type,
"message": message,
"timestamp": Utc::now().to_rfc3339()
}
}),
extensions: Default::default(),
},
});

if let Err(e) = self.config.mcp_tx.send(notification).await {
error!(
"Failed to send MCP notification from subagent {}: {}",
self.id, e
);
}
}

/// Get current progress information
Expand Down Expand Up @@ -192,10 +130,8 @@ impl SubAgent {
&self,
message: String,
task_config: TaskConfig,
) -> Result<Message, anyhow::Error> {
) -> Result<Vec<Message>, anyhow::Error> {
debug!("Processing message for subagent {}", self.id);
self.send_mcp_notification("message_processing", &format!("Processing: {}", message))
.await;

// Get provider from task config
let provider = self
Expand Down Expand Up @@ -234,6 +170,7 @@ impl SubAgent {
// Generate response from provider with loop for tool processing (max_turns iterations)
let mut loop_count = 0;
let max_turns = self.config.max_turns.unwrap_or(DEFAULT_SUBAGENT_MAX_TURNS);
let mut last_error: Option<anyhow::Error> = None;

// Generate response from provider
loop {
Expand Down Expand Up @@ -265,18 +202,12 @@ impl SubAgent {
// If there are no tool requests, we're done
if tool_requests.is_empty() || loop_count >= max_turns {
self.add_message(response.clone()).await;
messages.push(response.clone());

// Send notification about response
self.send_mcp_notification(
"response_generated",
&format!("Responded: {}", response.as_concat_text()),
)
.await;

// Set status back to ready and return the final response
// Set status back to ready
self.set_status(SubAgentStatus::Completed("Completed!".to_string()))
.await;
break Ok(response);
break;
}

// Add the assistant message with tool calls to the conversation
Expand All @@ -285,13 +216,6 @@ impl SubAgent {
// Process each tool request and create user response messages
for request in &tool_requests {
if let Ok(tool_call) = &request.tool_call {
// Send notification about tool usage
self.send_mcp_notification(
"tool_usage",
&format!("Using tool: {}", tool_call.name),
)
.await;

// Handle platform tools or dispatch to extension manager
let tool_result = match self
.extension_manager
Expand All @@ -310,13 +234,6 @@ impl SubAgent {
let tool_response_message = Message::user()
.with_tool_response(request.id.clone(), Ok(result.clone()));
messages.push(tool_response_message);

// Send notification about tool completion
self.send_mcp_notification(
"tool_completed",
&format!("Tool {} completed successfully", tool_call.name),
)
.await;
}
Err(e) => {
// Create a user message with the tool error
Expand All @@ -325,13 +242,6 @@ impl SubAgent {
Err(ToolError::ExecutionError(e.to_string())),
);
messages.push(tool_error_message);

// Send notification about tool error
self.send_mcp_notification(
"tool_error",
&format!("Tool {} error: {}", tool_call.name, e),
)
.await;
}
}
}
Expand All @@ -344,24 +254,31 @@ impl SubAgent {
"Context length exceeded".to_string(),
))
.await;
break Ok(Message::assistant().with_context_length_exceeded(
"The context length of the model has been exceeded. Please start a new session and try again.",
));
last_error = Some(anyhow::anyhow!("Context length exceeded"));
break;
}
Err(ProviderError::RateLimitExceeded(_)) => {
self.set_status(SubAgentStatus::Completed("Rate limit exceeded".to_string()))
.await;
break Ok(Message::assistant()
.with_text("Rate limit exceeded. Please try again later."));
last_error = Some(anyhow::anyhow!("Rate limit exceeded"));
break;
}
Err(e) => {
self.set_status(SubAgentStatus::Completed(format!("Error: {}", e)))
.await;
error!("Error: {}", e);
break Ok(Message::assistant().with_text(format!("Ran into this error: {e}.\n\nPlease retry if you think this is a transient or recoverable error.")));
last_error = Some(anyhow::anyhow!("Provider error: {}", e));
break;
}
}
}

// Handle error cases or return the last message
if let Some(error) = last_error {
Err(error)
} else {
Ok(messages)
}
}

/// Add a message to the conversation (for tracking agent responses)
Expand Down
Loading
Loading