public/galaxy
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

Implement local sub-agent execution for OSS/Bedrock mode

Browse Source 
- 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) <noreply@anthropic.com>
This commit is contained in:
Ryan Ward
2026-06-03 13:29:09 -05:00
co-authored by Claude Opus 4.6
parent 17ecf67970
commit a309006458
18 changed files with 458 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/ai/blocklist/action_model/execute.rs
+50 -4
View File
@@ -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<Self>,
) -> 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,
app/src/ai/blocklist/action_model/execute/call_mcp_tool.rs
+16
View File
@@ -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(|_| "<serialize error>".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.
app/src/ai/blocklist/action_model/execute/file_glob.rs
+1
View File
@@ -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());
app/src/ai/blocklist/action_model/execute/grep.rs
+1
View File
@@ -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);
app/src/ai/blocklist/action_model/execute/read_files.rs
+4
View File
@@ -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(
app/src/ai/blocklist/action_model/execute/request_file_edits.rs
+2 -1
View File
@@ -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;
};
app/src/ai/blocklist/action_model/execute/send_message.rs
+21
View File
@@ -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();
app/src/ai/blocklist/action_model/execute/shell_command.rs
+4
View File
@@ -212,6 +212,10 @@ impl ShellCommandExecutor {
input: ExecuteActionInput,
ctx: &mut ModelContext<Self>,
) -> impl Into<AnyActionExecution> {
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.
app/src/ai/blocklist/action_model/execute/start_agent.rs
+179 -69
View File
@@ -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<String>,
}
/// 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<AIConversationId>,
sender: async_channel::Sender<StartAgentDecision>,
/// 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<PendingStartAgent>,
/// All in-flight StartAgent actions. Multiple agents can be spawned
/// concurrently from the same parent.
pending: Vec<PendingStartAgent>,
}
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>,