From a309006458e40e20ad2f104fdc1a880efe101873 Mon Sep 17 00:00:00 2001 From: Ryan Ward Date: Wed, 3 Jun 2026 13:29:09 -0500 Subject: [PATCH] Implement local sub-agent execution for OSS/Bedrock mode - Fix start_agent tool mapping: route to Tool::StartAgent instead of dead-end Tool::Subagent so tool calls become executable actions - Block start_agent until child finishes: parent waits for child conversation to complete and receives full output as tool result - Fix tool result delivery: add StartAgent/StartAgentV2 cases to extract_tool_result_content so the model actually sees agent output - Support parallel agent spawning: change StartAgent action phase from Serial to Parallel, and track multiple pending agents via Vec - Mark child conversations as Success on EndTurn: emit ConversationStatus::Success when a child stream ends with no actions - Skip orchestration SSE in local mode: prevent app freeze from trying to connect to non-existent server - Remove send_message_to_agent and suggest_next_prompt from tool list: these require server infrastructure that doesn't exist in OSS mode - Add "Waiting for sub-agents..." status message while agents process - Fix child agent pane close: actually dismiss instead of re-hiding, track dismissed IDs to prevent re-creation on restart - Fix cache_miss_tokens calculation and show per-block cache stats - Add [tool-debug] logging throughout tool invocation pipeline Bump version to 1.6.0. Co-Authored-By: Claude Opus 4.6 (1M context) --- Cargo.lock | 2 +- app/Cargo.toml | 2 +- app/src/ai/agent/conversation.rs | 23 +- app/src/ai/bedrock/request_translator.rs | 55 ++-- app/src/ai/bedrock/response_translator.rs | 34 ++- app/src/ai/blocklist/action_model.rs | 47 ++++ app/src/ai/blocklist/action_model/execute.rs | 54 +++- .../action_model/execute/call_mcp_tool.rs | 16 ++ .../action_model/execute/file_glob.rs | 1 + .../ai/blocklist/action_model/execute/grep.rs | 1 + .../action_model/execute/read_files.rs | 4 + .../execute/request_file_edits.rs | 3 +- .../action_model/execute/send_message.rs | 21 ++ .../action_model/execute/shell_command.rs | 4 + .../action_model/execute/start_agent.rs | 248 +++++++++++++----- app/src/ai/blocklist/controller.rs | 24 ++ app/src/pane_group/mod.rs | 27 +- app/src/terminal/input/agent.rs | 6 +- 18 files changed, 458 insertions(+), 114 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7ab1e2a4..2e1163fa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5313,7 +5313,7 @@ dependencies = [ [[package]] name = "galaxy" -version = "1.5.6" +version = "1.6.0" dependencies = [ "addr", "aho-corasick", diff --git a/app/Cargo.toml b/app/Cargo.toml index f30b37b2..81de7b5f 100644 --- a/app/Cargo.toml +++ b/app/Cargo.toml @@ -5,7 +5,7 @@ description = "Galaxy - AI-powered terminal" edition = "2021" autobins = false name = "galaxy" -version = "1.5.6" +version = "1.6.0" publish.workspace = true license.workspace = true diff --git a/app/src/ai/agent/conversation.rs b/app/src/ai/agent/conversation.rs index 84774455..e442a18b 100644 --- a/app/src/ai/agent/conversation.rs +++ b/app/src/ai/agent/conversation.rs @@ -3320,7 +3320,28 @@ impl AIConversation { pub fn cache_miss_tokens(&self) -> u32 { self.total_token_usage_by_model .values() - .map(|u| u.total_input) + .map(|u| u.total_input.saturating_sub(u.input_cache_read + u.input_cache_write)) + .sum() + } + + pub fn last_block_cache_read_tokens(&self) -> u32 { + self.last_block_token_usage_by_model + .values() + .map(|u| u.input_cache_read) + .sum() + } + + pub fn last_block_cache_write_tokens(&self) -> u32 { + self.last_block_token_usage_by_model + .values() + .map(|u| u.input_cache_write) + .sum() + } + + pub fn last_block_cache_miss_tokens(&self) -> u32 { + self.last_block_token_usage_by_model + .values() + .map(|u| u.total_input.saturating_sub(u.input_cache_read + u.input_cache_write)) .sum() } diff --git a/app/src/ai/bedrock/request_translator.rs b/app/src/ai/bedrock/request_translator.rs index 29f79a85..3c681c7f 100644 --- a/app/src/ai/bedrock/request_translator.rs +++ b/app/src/ai/bedrock/request_translator.rs @@ -1109,7 +1109,7 @@ pub fn default_tool_definitions() -> Vec { }, ToolDefinition { name: "start_agent".to_string(), - description: "Start a sub-agent to handle a specific task autonomously. Use for delegating independent work that can run in parallel. The agent gets its own conversation context and tool access.".to_string(), + description: "Start a sub-agent to handle a specific task autonomously. Use for delegating independent work that can run in parallel. The agent gets its own conversation context and tool access. IMPORTANT: Only use this for the initial investigation or when genuinely new research is needed. Do NOT re-spawn agents for follow-up questions if you already have their output in context — just answer from the information you already have.".to_string(), input_schema: serde_json::json!({ "type": "object", "properties": { @@ -1119,18 +1119,9 @@ pub fn default_tool_definitions() -> Vec { "required": ["name", "prompt"] }), }, - ToolDefinition { - name: "send_message_to_agent".to_string(), - description: "Send a message to a running sub-agent. Use to provide additional context, ask for updates, or redirect the agent's work.".to_string(), - input_schema: serde_json::json!({ - "type": "object", - "properties": { - "agent_id": { "type": "string", "description": "ID of the target sub-agent" }, - "message": { "type": "string", "description": "Message to send to the agent" } - }, - "required": ["agent_id", "message"] - }), - }, + // send_message_to_agent removed — child agents in local/Bedrock mode + // run autonomously and cannot receive messages. Exposing this tool + // causes the model to poll in a loop. ToolDefinition { name: "ask_user_question".to_string(), description: "Ask the user a question when you need clarification or a decision. Present clear options when possible. Use sparingly — prefer making reasonable assumptions.".to_string(), @@ -1143,18 +1134,10 @@ pub fn default_tool_definitions() -> Vec { "required": ["question"] }), }, - ToolDefinition { - name: "suggest_next_prompt".to_string(), - description: "After completing a task, suggest a relevant follow-up action. Only call once at the end of your response. Keep labels concise and action-oriented.".to_string(), - input_schema: serde_json::json!({ - "type": "object", - "properties": { - "prompt": { "type": "string", "description": "The full prompt text sent to the agent if the user clicks" }, - "label": { "type": "string", "description": "Short display label (under 40 chars)" } - }, - "required": ["prompt", "label"] - }), - }, + // suggest_next_prompt removed — its executor hangs waiting for UI + // interaction that doesn't exist in the Bedrock path. The stream-level + // skip (response_translator) prevents deadlocks, but removing the tool + // definition avoids wasting output tokens on calls that will be discarded. ToolDefinition { name: "read_skill".to_string(), description: "Read a skill definition to understand available capabilities and how to use them.".to_string(), @@ -1398,6 +1381,28 @@ fn extract_tool_result_content(result: &api::request::input::ToolCallResult) -> } None => ("Write to shell command completed.".to_string(), false), }, + api::request::input::tool_call_result::Result::StartAgent(start_agent_result) => { + match &start_agent_result.result { + Some(api::start_agent_result::Result::Success(success)) => { + (success.agent_id.clone(), false) + } + Some(api::start_agent_result::Result::Error(error)) => { + (format!("Agent error: {}", error.error), true) + } + None => ("Agent completed.".to_string(), false), + } + } + api::request::input::tool_call_result::Result::StartAgentV2(start_agent_result) => { + match &start_agent_result.result { + Some(api::start_agent_v2_result::Result::Success(success)) => { + (success.agent_id.clone(), false) + } + Some(api::start_agent_v2_result::Result::Error(error)) => { + (format!("Agent error: {}", error.error), true) + } + None => ("Agent completed.".to_string(), false), + } + } _ => ("Tool completed successfully.".to_string(), false), } } else { diff --git a/app/src/ai/bedrock/response_translator.rs b/app/src/ai/bedrock/response_translator.rs index 0a2903d8..85423ceb 100644 --- a/app/src/ai/bedrock/response_translator.rs +++ b/app/src/ai/bedrock/response_translator.rs @@ -104,6 +104,7 @@ pub fn bedrock_stream_to_response_events( let mut current_tool_name = String::new(); let mut current_tool_input_json = String::new(); let mut _has_tool_calls = false; + let mut has_start_agent_calls = false; let mut input_tokens: i32 = 0; let mut output_tokens: i32 = 0; let mut cache_read_input_tokens: i32 = 0; @@ -275,6 +276,9 @@ pub fn bedrock_stream_to_response_events( current_tool_name, current_tool_use_id, current_tool_input_json )); } + if current_tool_name == "start_agent" { + has_start_agent_calls = true; + } let tool_msg = build_tool_call_message( &task_id, ¤t_tool_use_id, @@ -372,12 +376,29 @@ pub fn bedrock_stream_to_response_events( if !text_flushed { let add_msg = build_add_agent_output_message(&task_id, &msg_id, &buffered_text); yield Ok(add_msg); + text_flushed = true; + current_text_message_id = Some(msg_id); } else { let append = build_append_text(&task_id, &msg_id, &buffered_text); yield Ok(append); } } + // Show a status message while sub-agents are working so the user + // knows we're waiting on them. + if has_start_agent_calls { + let status_text = "\n\n*Waiting for sub-agents to finish, then I'll review and continue...*"; + if text_flushed { + let msg_id = current_text_message_id.as_ref().unwrap(); + let append = build_append_text(&task_id, msg_id, status_text); + yield Ok(append); + } else { + let msg_id = Uuid::new_v4().to_string(); + let add_msg = build_add_agent_output_message(&task_id, &msg_id, status_text); + yield Ok(add_msg); + } + } + let cost = estimate_cost_cents( input_tokens as u32, output_tokens as u32, @@ -985,13 +1006,14 @@ fn build_tool_call_message( )) } "start_agent" => { - let _name = input.get("name").and_then(|v| v.as_str()).unwrap_or("").to_string(); + let name = input.get("name").and_then(|v| v.as_str()).unwrap_or("").to_string(); let prompt = input.get("prompt").and_then(|v| v.as_str()).unwrap_or("").to_string(); - Some(api::message::tool_call::Tool::Subagent( - api::message::tool_call::Subagent { - task_id: String::new(), - payload: prompt, - metadata: None, + Some(api::message::tool_call::Tool::StartAgent( + api::StartAgent { + name, + prompt, + execution_mode: None, + lifecycle_subscription: None, }, )) } diff --git a/app/src/ai/blocklist/action_model.rs b/app/src/ai/blocklist/action_model.rs index b1fa897b..c90fbb49 100644 --- a/app/src/ai/blocklist/action_model.rs +++ b/app/src/ai/blocklist/action_model.rs @@ -465,6 +465,16 @@ impl BlocklistAIActionModel { conversation_id: AIConversationId, ctx: &mut ModelContext, ) { + let pending_count = self + .pending_actions + .get(&conversation_id) + .map(|q| q.len()) + .unwrap_or(0); + log::info!( + "[tool-debug] try_to_execute_available_actions: conversation={:?}, pending_count={}", + conversation_id, + pending_count + ); loop { let Some(front_action) = self .pending_actions @@ -472,9 +482,16 @@ impl BlocklistAIActionModel { .and_then(|queue| queue.front()) .cloned() else { + log::info!("[tool-debug] try_to_execute_available_actions: no more pending actions"); return; }; + log::info!( + "[tool-debug] try_to_execute_available_actions: trying action id={:?}, type={:?}", + front_action.id, + std::mem::discriminant(&front_action.action) + ); + if let Some(current_phase) = self.action_execution_phase(conversation_id) { if !self.can_start_action_in_current_phase( &front_action, @@ -482,6 +499,10 @@ impl BlocklistAIActionModel { current_phase, ctx, ) { + log::info!( + "[tool-debug] try_to_execute_available_actions: cannot start in current phase {:?}", + current_phase + ); return; } } @@ -489,15 +510,22 @@ impl BlocklistAIActionModel { let Some(result) = self.start_pending_action_by_id(&front_action.id, conversation_id, false, ctx) else { + log::info!("[tool-debug] try_to_execute_available_actions: start_pending_action_by_id returned None (blocked)"); return; }; + log::info!( + "[tool-debug] try_to_execute_available_actions: action started, result={:?}", + std::mem::discriminant(&result) + ); + if matches!( result, StartedAction::Async { phase: RunningActionPhase::Serial } ) { + log::info!("[tool-debug] try_to_execute_available_actions: serial async action, stopping loop"); return; } } @@ -853,6 +881,19 @@ impl BlocklistAIActionModel { conversation_id: AIConversationId, ctx: &mut ModelContext, ) { + log::info!( + "[tool-debug] queue_actions: queuing {} actions for conversation {:?}", + actions.len(), + conversation_id + ); + for (i, action) in actions.iter().enumerate() { + log::info!( + "[tool-debug] queue_actions: [{}] id={:?}, type={:?}", + i, + action.id, + std::mem::discriminant(&action.action) + ); + } self.action_order.insert( conversation_id, actions @@ -1135,6 +1176,12 @@ impl BlocklistAIActionModel { cancellation_reason: Option, ctx: &mut ModelContext, ) { + log::info!( + "[tool-debug] handle_action_result: action_id={:?}, result_type={:?}, cancellation={:?}", + action_result.id, + std::mem::discriminant(&action_result.result), + cancellation_reason + ); let should_remove_entry = self.running_actions .get_mut(&conversation_id) diff --git a/app/src/ai/blocklist/action_model/execute.rs b/app/src/ai/blocklist/action_model/execute.rs index 1929c428..d98ee397 100644 --- a/app/src/ai/blocklist/action_model/execute.rs +++ b/app/src/ai/blocklist/action_model/execute.rs @@ -405,6 +405,9 @@ impl BlocklistAIActionExecutor { { RunningActionPhase::Parallel(ParallelExecutionPolicy::ReadOnlyLocalContext) } + AIAgentActionType::StartAgent { .. } => { + RunningActionPhase::Parallel(ParallelExecutionPolicy::ReadOnlyLocalContext) + } _ => RunningActionPhase::Serial, } } @@ -527,8 +530,16 @@ impl BlocklistAIActionExecutor { is_user_initiated: bool, ctx: &mut ModelContext, ) -> TryExecuteResult { + log::info!( + "[tool-debug] try_to_execute_action: action_id={:?}, type={:?}, is_user_initiated={}", + action.id, + std::mem::discriminant(&action.action), + is_user_initiated + ); + // We should never actually execute actions in view-only mode. if self.is_shared_session_viewer() { + log::info!("[tool-debug] try_to_execute_action: BLOCKED - shared session viewer mode"); return TryExecuteResult::NotExecuted { reason: NotExecutedReason::WaitingOnSharer, action: Box::new(action), @@ -541,6 +552,11 @@ impl BlocklistAIActionExecutor { }; let can_auto_execute = self.should_autoexecute(input, ctx); let is_agent_autonomous = AppExecutionMode::as_ref(ctx).is_autonomous(); + log::info!( + "[tool-debug] try_to_execute_action: can_auto_execute={}, is_agent_autonomous={}", + can_auto_execute, + is_agent_autonomous + ); // The agent cannot auto execute and either: // - the agent is interactive, OR @@ -549,6 +565,10 @@ impl BlocklistAIActionExecutor { || can_auto_execute || (is_agent_autonomous && action.action.is_request_command_output())); if needs_confirmation { + log::info!( + "[tool-debug] try_to_execute_action: NEEDS CONFIRMATION - action_id={:?}", + action.id + ); return TryExecuteResult::NotExecuted { action: Box::new(action), reason: NotExecutedReason::NeedsConfirmation, @@ -580,6 +600,11 @@ impl BlocklistAIActionExecutor { } } + log::info!( + "[tool-debug] try_to_execute_action: EXECUTING action_id={:?}, type={:?}", + action.id, + std::mem::discriminant(&action.action) + ); let action_clone = action.clone(); let execution = match &action.action { AIAgentActionType::RequestCommandOutput { .. } @@ -702,12 +727,26 @@ impl BlocklistAIActionExecutor { }; let action_id = action_clone.id.clone(); - match execution { - AnyActionExecution::NotReady => TryExecuteResult::NotExecuted { - reason: NotExecutedReason::NotReady, - action: Box::new(action_clone), + log::info!( + "[tool-debug] try_to_execute_action: execution result type={:?} for action_id={:?}", + match &execution { + AnyActionExecution::NotReady => "NotReady", + AnyActionExecution::InvalidAction => "InvalidAction", + AnyActionExecution::Async { .. } => "Async", + AnyActionExecution::Sync(_) => "Sync", }, + action_id + ); + match execution { + AnyActionExecution::NotReady => { + log::info!("[tool-debug] try_to_execute_action: NOT READY - action_id={:?}", action_id); + TryExecuteResult::NotExecuted { + reason: NotExecutedReason::NotReady, + action: Box::new(action_clone), + } + } AnyActionExecution::InvalidAction => { + log::error!("[tool-debug] try_to_execute_action: INVALID ACTION - action_id={:?}", action_id); debug_assert!(false, "Tried to execute AIAgentAction with wrong executor."); TryExecuteResult::NotExecuted { reason: NotExecutedReason::NotReady, @@ -728,11 +767,18 @@ impl BlocklistAIActionExecutor { ctx.emit(BlocklistAIActionExecutorEvent::ExecutingAction { action_id: action_id.clone(), }); + log::info!("[tool-debug] try_to_execute_action: spawning ASYNC execution for action_id={:?}", action_id); ctx.spawn(execute_future, move |me, result, ctx| { let Some(running) = me.async_executing_actions.remove(&action_id) else { + log::warn!("[tool-debug] try_to_execute_action: async action completed but not found in executing map, action_id={:?}", action_id); return; }; let result = on_complete(result, ctx); + log::info!( + "[tool-debug] try_to_execute_action: ASYNC action COMPLETED action_id={:?}, result_type={:?}", + action_id, + std::mem::discriminant(&result) + ); ctx.emit(BlocklistAIActionExecutorEvent::FinishedAction { result: Arc::new(AIAgentActionResult { id: action_id, diff --git a/app/src/ai/blocklist/action_model/execute/call_mcp_tool.rs b/app/src/ai/blocklist/action_model/execute/call_mcp_tool.rs index 87668efa..e620f47d 100644 --- a/app/src/ai/blocklist/action_model/execute/call_mcp_tool.rs +++ b/app/src/ai/blocklist/action_model/execute/call_mcp_tool.rs @@ -84,6 +84,7 @@ impl CallMCPToolExecutor { #[cfg(not(target_family = "wasm"))] { + log::info!("[tool-debug] CallMCPToolExecutor::execute called"); let server_output_id = get_server_output_id(input.conversation_id, ctx); let AIAgentAction { action: @@ -95,13 +96,21 @@ impl CallMCPToolExecutor { .. } = input.action else { + log::error!("[tool-debug] CallMCPToolExecutor::execute: action type mismatch!"); return ActionExecution::InvalidAction; }; let name_owned = name.to_owned(); let name_clone = name_owned.clone(); + log::info!( + "[tool-debug] CallMCPToolExecutor: tool_name={}, server_id={:?}, input={}", + name, + server_id, + serde_json::to_string(input).unwrap_or_else(|_| "".to_string()) + ); let serde_json::Value::Object(mut arguments) = input.clone() else { + log::error!("[tool-debug] CallMCPToolExecutor: input is not an object!"); return ActionExecution::Sync(AIAgentActionResultType::CallMCPTool( CallMCPToolResult::Error("MCP server tool input not an object".to_owned()), )); @@ -132,10 +141,12 @@ impl CallMCPToolExecutor { }; let Some(reconnecting_peer) = templatable_peer else { + log::error!("[tool-debug] CallMCPToolExecutor: MCP server for tool '{}' NOT FOUND", name_owned); return ActionExecution::Sync(AIAgentActionResultType::CallMCPTool( CallMCPToolResult::Error("MCP server for tool not found".to_owned()), )); }; + log::info!("[tool-debug] CallMCPToolExecutor: found MCP server peer for tool '{}'", name_owned); let name_owned_inner = name_owned.clone(); ActionExecution::new_async( @@ -210,6 +221,11 @@ fn handle_call_tool_result( tool_name: String, ctx: &galaxyui::AppContext, ) -> AIAgentActionResultType { + log::info!( + "[tool-debug] handle_call_tool_result: tool_name={}, is_ok={}", + tool_name, + res.is_ok() + ); let action_result = match res { Ok(result) => { // Even if the call was successful, the response could still be an error so we need to check. diff --git a/app/src/ai/blocklist/action_model/execute/file_glob.rs b/app/src/ai/blocklist/action_model/execute/file_glob.rs index 37394902..7fa0305c 100644 --- a/app/src/ai/blocklist/action_model/execute/file_glob.rs +++ b/app/src/ai/blocklist/action_model/execute/file_glob.rs @@ -111,6 +111,7 @@ impl FileGlobExecutor { else { return ActionExecution::InvalidAction; }; + log::info!("[tool-debug] FileGlobExecutor::execute: patterns={:?}, path={:?}", patterns, path); // If the path is not provided, use the current working directory. let path = path.clone().unwrap_or_else(|| ".".to_string()); diff --git a/app/src/ai/blocklist/action_model/execute/grep.rs b/app/src/ai/blocklist/action_model/execute/grep.rs index 059e09a6..39896c01 100644 --- a/app/src/ai/blocklist/action_model/execute/grep.rs +++ b/app/src/ai/blocklist/action_model/execute/grep.rs @@ -252,6 +252,7 @@ impl GrepExecutor { else { return ActionExecution::InvalidAction; }; + log::info!("[tool-debug] GrepExecutor::execute: queries={:?}, path={:?}", queries, path); let shell_launch_data = self.active_session.as_ref(ctx).shell_launch_data(ctx); let shell_type = self.active_session.as_ref(ctx).shell_type(ctx); diff --git a/app/src/ai/blocklist/action_model/execute/read_files.rs b/app/src/ai/blocklist/action_model/execute/read_files.rs index 2c395295..2bba0e66 100644 --- a/app/src/ai/blocklist/action_model/execute/read_files.rs +++ b/app/src/ai/blocklist/action_model/execute/read_files.rs @@ -95,6 +95,10 @@ impl ReadFilesExecutor { else { return ActionExecution::InvalidAction; }; + log::info!( + "[tool-debug] ReadFilesExecutor::execute: {} files requested", + locations.len() + ); BlocklistAIPermissions::handle(ctx).update(ctx, |model, _ctx| { model.add_temporary_file_read_permissions( diff --git a/app/src/ai/blocklist/action_model/execute/request_file_edits.rs b/app/src/ai/blocklist/action_model/execute/request_file_edits.rs index c1425b1c..de49ee97 100644 --- a/app/src/ai/blocklist/action_model/execute/request_file_edits.rs +++ b/app/src/ai/blocklist/action_model/execute/request_file_edits.rs @@ -151,9 +151,10 @@ impl RequestFileEditsExecutor { else { return ActionExecution::InvalidAction; }; + log::info!("[tool-debug] RequestFileEditsExecutor::execute: action_id={:?}", id); let Some(diff_view) = self.diff_views.get(id) else { - log::warn!("Tried to execute a RequestFileEdits action without a diff view"); + log::warn!("[tool-debug] RequestFileEditsExecutor: no diff view found for action_id={:?}", id); return ActionExecution::NotReady; }; diff --git a/app/src/ai/blocklist/action_model/execute/send_message.rs b/app/src/ai/blocklist/action_model/execute/send_message.rs index f7160be3..a1518b87 100644 --- a/app/src/ai/blocklist/action_model/execute/send_message.rs +++ b/app/src/ai/blocklist/action_model/execute/send_message.rs @@ -62,6 +62,27 @@ impl SendMessageToAgentExecutor { .and_then(|c| c.run_id()) .map(|s| s.to_string()) .unwrap_or_default(); + + // In local Bedrock mode, child agents run autonomously in their + // own conversation loop — there's no server API to relay messages. + // Return an error explaining the limitation so the model doesn't + // keep polling in a loop. + if sender_run_id.is_empty() { + log::info!( + "[send_message] No run_id for conversation {:?}, assuming local mode — skipping server API call", + conversation_id + ); + return ActionExecution::<()>::Sync(AIAgentActionResultType::SendMessageToAgent( + SendMessageToAgentResult::Error( + "Child agents are running autonomously in local mode. \ + You cannot send messages to them. They will complete their \ + tasks independently. Continue with your own work or wait \ + for the user to share the results." + .to_string(), + ), + )) + .into(); + } let log_addresses = addresses.clone(); let log_subject = subject.clone(); let log_sender_run_id = sender_run_id.clone(); diff --git a/app/src/ai/blocklist/action_model/execute/shell_command.rs b/app/src/ai/blocklist/action_model/execute/shell_command.rs index 64931d75..2866df7f 100644 --- a/app/src/ai/blocklist/action_model/execute/shell_command.rs +++ b/app/src/ai/blocklist/action_model/execute/shell_command.rs @@ -212,6 +212,10 @@ impl ShellCommandExecutor { input: ExecuteActionInput, ctx: &mut ModelContext, ) -> impl Into { + log::info!( + "[tool-debug] ShellCommandExecutor::execute: action_type={:?}", + std::mem::discriminant(&input.action.action) + ); let model = self.terminal_model.lock(); // Determine the action we want to take based on the input. diff --git a/app/src/ai/blocklist/action_model/execute/start_agent.rs b/app/src/ai/blocklist/action_model/execute/start_agent.rs index 32b60cee..75126f31 100644 --- a/app/src/ai/blocklist/action_model/execute/start_agent.rs +++ b/app/src/ai/blocklist/action_model/execute/start_agent.rs @@ -1,7 +1,7 @@ use futures::{future::BoxFuture, FutureExt}; use galaxyui::{Entity, ModelContext, SingletonEntity}; -use crate::ai::agent::conversation::{AIConversationId, ConversationStatus}; +use crate::ai::agent::conversation::{AIConversation, AIConversationId, ConversationStatus}; use crate::ai::agent::{ AIAgentAction, AIAgentActionResultType, AIAgentActionType, LifecycleEventType, StartAgentExecutionMode, StartAgentResult, @@ -18,6 +18,8 @@ use super::{ActionExecution, AnyActionExecution, ExecuteActionInput, PreprocessA enum StartAgentDecision { /// The child conversation was created successfully. Started { agent_id: String }, + /// The child agent completed and here is its output (local/blocking mode). + Completed { agent_id: String, output: String }, /// An error occurred while starting the agent. Error(String), } @@ -43,19 +45,22 @@ pub struct StartAgentRequest { pub parent_run_id: Option, } -/// Tracks a single in-flight StartAgent action. At most one can be pending at -/// a time because StartAgent actions execute serially (RunningActionPhase::Serial). +/// Tracks a single in-flight StartAgent action. struct PendingStartAgent { parent_conversation_id: AIConversationId, /// Set when `StartedNewConversation` fires for a conversation whose /// `parent_conversation_id` matches. child_conversation_id: Option, sender: async_channel::Sender, + /// When true, the executor blocks until the child agent finishes and + /// returns its output as the tool result (local Bedrock mode). + wait_for_completion: bool, } pub struct StartAgentExecutor { - /// The currently pending StartAgent action, if any. - pending: Option, + /// All in-flight StartAgent actions. Multiple agents can be spawned + /// concurrently from the same parent. + pending: Vec, } impl StartAgentExecutor { @@ -63,7 +68,7 @@ impl StartAgentExecutor { let history_model = BlocklistAIHistoryModel::handle(ctx); ctx.subscribe_to_model(&history_model, Self::handle_history_event); - Self { pending: None } + Self { pending: Vec::new() } } fn handle_history_event( @@ -76,44 +81,106 @@ impl StartAgentExecutor { new_conversation_id, .. } => { - let Some(pending) = self.pending.as_mut() else { - return; - }; - if pending.child_conversation_id.is_some() { + if self.pending.is_empty() { return; } let history = BlocklistAIHistoryModel::as_ref(ctx); let Some(conversation) = history.conversation(new_conversation_id) else { return; }; - if conversation.parent_conversation_id() == Some(pending.parent_conversation_id) { + let parent_id = conversation.parent_conversation_id(); + // Find the first pending entry that matches this parent and hasn't been assigned a child yet. + if let Some(pending) = self.pending.iter_mut().find(|p| { + p.child_conversation_id.is_none() + && parent_id == Some(p.parent_conversation_id) + }) { pending.child_conversation_id = Some(*new_conversation_id); } } BlocklistAIHistoryEvent::ConversationServerTokenAssigned { conversation_id, .. } => { - let matches = self - .pending - .as_ref() - .is_some_and(|p| p.child_conversation_id.as_ref() == Some(conversation_id)); - if !matches { + let Some(idx) = self.pending.iter().position(|p| { + p.child_conversation_id.as_ref() == Some(conversation_id) + }) else { + return; + }; + // Don't remove yet if we're waiting for completion — we need + // the entry to stay so UpdatedConversationStatus can find it. + if self.pending[idx].wait_for_completion { + // Just log and continue — we'll resolve on Success status. + let conversation = BlocklistAIHistoryModel::as_ref(ctx) + .conversation(conversation_id); + let agent_id = conversation + .and_then(|c| c.orchestration_agent_id()) + .or_else(|| { + conversation.and_then(|c| { + c.server_conversation_token() + .map(|t| t.as_str().to_string()) + }) + }) + .unwrap_or_else(|| conversation_id.to_string()); + log::info!( + "[start_agent] Child agent started: conversation_id={:?}, agent_id={}", + conversation_id, + agent_id + ); + log::info!( + "[start_agent] Local mode: waiting for child {:?} to complete", + conversation_id + ); return; } - let pending = self.pending.take().unwrap(); - let agent_id = BlocklistAIHistoryModel::as_ref(ctx) - .conversation(conversation_id) - .and_then(|c| c.orchestration_agent_id()); - match agent_id { - Some(id) => { - let _ = pending.sender.try_send(StartAgentDecision::Started { - agent_id: id.clone(), - }); + let pending = self.pending.remove(idx); + let conversation = BlocklistAIHistoryModel::as_ref(ctx) + .conversation(conversation_id); + // orchestration_agent_id() uses run_id in v2 mode, which won't + // exist for locally-spawned Bedrock child agents. Fall back to + // the server conversation token (set by the stream Init event) + // or the conversation ID itself as the agent identifier. + let agent_id = conversation + .and_then(|c| c.orchestration_agent_id()) + .or_else(|| { + conversation.and_then(|c| { + c.server_conversation_token() + .map(|t| t.as_str().to_string()) + }) + }) + .unwrap_or_else(|| conversation_id.to_string()); + + log::info!( + "[start_agent] Child agent started: conversation_id={:?}, agent_id={}", + conversation_id, + agent_id + ); + + // Only register with the orchestration streamer if the parent + // has a run_id (server-assigned). Local Bedrock child agents + // don't have server-side orchestration — attempting to open an + // SSE stream to a non-existent server freezes the app. + let parent_has_run_id = BlocklistAIHistoryModel::as_ref(ctx) + .conversation(&pending.parent_conversation_id) + .and_then(|c| c.run_id()) + .is_some(); + + if pending.wait_for_completion { + // Local mode: don't resolve yet — wait for the child to + // finish (UpdatedConversationStatus → Success) so we can + // return its output as the tool result. + log::info!( + "[start_agent] Local mode: waiting for child {:?} to complete", + conversation_id + ); + } else { + let _ = pending.sender.try_send(StartAgentDecision::Started { + agent_id: agent_id.clone(), + }); + if parent_has_run_id { if FeatureFlag::OrchestrationV2.is_enabled() { OrchestrationEventStreamer::handle(ctx).update(ctx, |streamer, ctx| { streamer.register_watched_run_id( pending.parent_conversation_id, - id, + agent_id, ctx, ); }); @@ -123,60 +190,70 @@ impl StartAgentExecutor { }); } } - None => { - log::error!( - "ConversationServerTokenAssigned fired but no agent identifier for \ - {conversation_id:?}" - ); - let _ = pending.sender.try_send(StartAgentDecision::Error( - "Server did not assign an agent identifier".to_string(), - )); - if !FeatureFlag::OrchestrationV2.is_enabled() { - OrchestrationEventService::handle(ctx).update(ctx, |svc, ctx| { - svc.emit_child_startup_errored( - *conversation_id, - "missing_agent_id".to_string(), - "Server did not assign an agent identifier".to_string(), - ctx, - ); - }); - } - } } } BlocklistAIHistoryEvent::UpdatedConversationStatus { conversation_id, .. } => { - let matches = self - .pending - .as_ref() - .is_some_and(|p| p.child_conversation_id.as_ref() == Some(conversation_id)); - if !matches { + let Some(idx) = self.pending.iter().position(|p| { + p.child_conversation_id.as_ref() == Some(conversation_id) + }) else { return; - } + }; let history = BlocklistAIHistoryModel::as_ref(ctx); let Some(conversation) = history.conversation(conversation_id) else { return; }; - let error_msg = start_agent_error_message_for_status( - conversation.status(), - conversation.status_error_message(), - ); - if let Some(error_msg) = error_msg { - let pending = self.pending.take().unwrap(); - let _ = pending - .sender - .try_send(StartAgentDecision::Error(error_msg.clone())); - if !FeatureFlag::OrchestrationV2.is_enabled() { - OrchestrationEventService::handle(ctx).update(ctx, |svc, ctx| { - svc.emit_child_startup_errored( - *conversation_id, - "conversation_status".to_string(), - error_msg, - ctx, - ); + + match conversation.status() { + ConversationStatus::Success => { + if !self.pending[idx].wait_for_completion { + // Non-blocking mode — already resolved on token assignment. + return; + } + let pending = self.pending.remove(idx); + // Extract the child's text output from its last exchange. + let output = extract_child_output(conversation); + log::info!( + "[start_agent] Child agent {:?} completed with {} chars of output", + conversation_id, + output.len() + ); + let agent_id = conversation + .orchestration_agent_id() + .or_else(|| { + conversation + .server_conversation_token() + .map(|t| t.as_str().to_string()) + }) + .unwrap_or_else(|| conversation_id.to_string()); + let _ = pending.sender.try_send(StartAgentDecision::Completed { + agent_id, + output, }); } + status => { + let error_msg = start_agent_error_message_for_status( + status, + conversation.status_error_message(), + ); + if let Some(error_msg) = error_msg { + let pending = self.pending.remove(idx); + let _ = pending + .sender + .try_send(StartAgentDecision::Error(error_msg.clone())); + if !FeatureFlag::OrchestrationV2.is_enabled() { + OrchestrationEventService::handle(ctx).update(ctx, |svc, ctx| { + svc.emit_child_startup_errored( + *conversation_id, + "conversation_status".to_string(), + error_msg, + ctx, + ); + }); + } + } + } } } BlocklistAIHistoryEvent::CreatedSubtask { .. } @@ -356,11 +433,16 @@ impl StartAgentExecutor { } }; + // In local mode (no parent_run_id), block until the child finishes + // so the parent model receives the child's output as the tool result. + let wait_for_completion = parent_run_id.is_none(); + let (sender, receiver) = async_channel::bounded(1); - self.pending = Some(PendingStartAgent { + self.pending.push(PendingStartAgent { parent_conversation_id, child_conversation_id: None, sender, + wait_for_completion, }); ctx.emit(StartAgentExecutorEvent::CreateAgent(StartAgentRequest { @@ -380,6 +462,15 @@ impl StartAgentExecutor { version, }) } + Ok(StartAgentDecision::Completed { agent_id, output }) => { + // Return the child's output as a "success" with the output + // embedded in the agent_id field. The Display impl on + // StartAgentResult will show this to the model. + AIAgentActionResultType::StartAgent(StartAgentResult::Success { + agent_id: format!("{agent_id}\n\nAgent output:\n{output}"), + version, + }) + } Ok(StartAgentDecision::Error(error)) => { AIAgentActionResultType::StartAgent(StartAgentResult::Error { error, version }) } @@ -399,6 +490,25 @@ impl StartAgentExecutor { } } +/// Extracts the text output from a child agent's conversation. +/// Collects text from all exchanges in the conversation. +fn extract_child_output(conversation: &AIConversation) -> String { + let mut output_parts = Vec::new(); + for exchange in conversation.all_exchanges() { + if let Some(output) = exchange.output_status.output() { + let text = output.get().format_for_copy(None); + if !text.is_empty() { + output_parts.push(text); + } + } + } + if output_parts.is_empty() { + "Agent completed but produced no text output.".to_string() + } else { + output_parts.join("\n\n") + } +} + fn start_agent_error_message_for_status( status: &ConversationStatus, error_message: Option<&str>, diff --git a/app/src/ai/blocklist/controller.rs b/app/src/ai/blocklist/controller.rs index 13ec52d5..5449191c 100644 --- a/app/src/ai/blocklist/controller.rs +++ b/app/src/ai/blocklist/controller.rs @@ -2656,6 +2656,30 @@ impl BlocklistAIController { "[bedrock-debug] AfterStreamFinished: NO actions to queue, was_passive={}, is_any_unfinished={}", was_passive_request, is_any_exchange_unfinished ); + + // If this is a child conversation (has a parent) and the + // stream ended with EndTurn and no actions, the child agent + // is done. Mark it as Success so the StartAgentExecutor + // can resolve and return the output to the parent. + let is_child = history_model + .as_ref(ctx) + .conversation(&conversation_id) + .and_then(|c| c.parent_conversation_id()) + .is_some(); + if is_child && !was_passive_request { + log::info!( + "[bedrock-debug] AfterStreamFinished: child conversation {:?} completed, setting status to Success", + conversation_id + ); + history_model.update(ctx, |history_model, ctx| { + history_model.update_conversation_status( + self.terminal_view_id, + conversation_id, + crate::ai::agent::conversation::ConversationStatus::Success, + ctx, + ); + }); + } } // Cancelled streams will handle pending_response_stream updates synchronously. diff --git a/app/src/pane_group/mod.rs b/app/src/pane_group/mod.rs index 1ccc872c..dd03be37 100644 --- a/app/src/pane_group/mod.rs +++ b/app/src/pane_group/mod.rs @@ -888,6 +888,10 @@ pub struct PaneGroup { /// be revealed from the parent's status card. child_agent_panes: HashMap, + /// Child agent conversations that the user has explicitly closed. + /// Prevents them from being re-created on app restart. + dismissed_child_agents: HashSet, + /// Tab-level custom title set via the rename-tab flow. custom_title: Option, } @@ -3001,6 +3005,7 @@ impl PaneGroup { is_right_panel_maximized: false, pending_ambient_agent_conversation_restorations: HashMap::new(), child_agent_panes: HashMap::new(), + dismissed_child_agents: HashSet::new(), custom_title: None, }; @@ -3052,14 +3057,18 @@ impl PaneGroup { // Check in-memory children (live conversations). for child in history_model.child_conversations_of(parent_id) { let child_id = child.id(); - if !self.child_agent_panes.contains_key(&child_id) { + if !self.child_agent_panes.contains_key(&child_id) + && !self.dismissed_child_agents.contains(&child_id) + { children_to_create.insert(child_id); } } // Check the startup index for children not yet in memory. for &child_id in history_model.child_conversation_ids_of(&parent_id) { - if !self.child_agent_panes.contains_key(&child_id) { + if !self.child_agent_panes.contains_key(&child_id) + && !self.dismissed_child_agents.contains(&child_id) + { children_to_create.insert(child_id); } } @@ -4448,16 +4457,28 @@ impl PaneGroup { return; } - // If this pane is a child agent, re-hide it instead of closing it. + // If this pane is a child agent, remove it from tracking and + // discard it fully so it doesn't reappear on re-launch. if self.is_child_agent_pane(pane_id) { + let dismissed_id = self + .child_agent_panes + .iter() + .find(|(_, p)| **p == pane_id) + .map(|(id, _)| *id); + if let Some(id) = dismissed_id { + self.dismissed_child_agents.insert(id); + } + self.child_agent_panes.retain(|_, &mut p| p != pane_id); if !self.panes.is_pane_hidden(&pane_id) { self.panes.hide_pane_for_child_agent(pane_id); } + self.panes.remove_hidden_pane(pane_id); self.focus_next_terminal_pane_and_activate_session( pane_id, PaneRemovalReason::Close, ctx, ); + self.pane_contents.remove(&pane_id); self.handle_pane_count_change(ctx); ctx.emit(Event::TerminalViewStateChanged); ctx.emit(Event::AppStateChanged); diff --git a/app/src/terminal/input/agent.rs b/app/src/terminal/input/agent.rs index b7748fdc..ad8b6971 100644 --- a/app/src/terminal/input/agent.rs +++ b/app/src/terminal/input/agent.rs @@ -634,9 +634,9 @@ fn render_session_status_bar(appearance: &Appearance, app: &AppContext, conversa let (cache_read, cache_write, cache_miss, cost_cents, context_usage, current_context) = if let Some(conversation) = BlocklistAIHistoryModel::as_ref(app).conversation(&conversation_id) { ( - conversation.total_cache_read_tokens(), - conversation.total_cache_write_tokens(), - conversation.cache_miss_tokens(), + conversation.last_block_cache_read_tokens(), + conversation.last_block_cache_write_tokens(), + conversation.last_block_cache_miss_tokens(), conversation.total_cost_cents(), conversation.context_window_usage(), conversation.current_context_tokens(),