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

Fix Local Agent Execution And Auth Checks

Browse Source 
- Gate server requests on available credentials
- Run local child agents directly without a parent run ID
- Include command IDs in Bedrock context and recognize transfer tools
This commit is contained in:
rkw6086
2026-07-28 10:43:59 -05:00
parent 87e0c83e9e
commit a078287f4b
35 changed files with 1078 additions and 237 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/ai/blocklist/action_model/execute/run_agents.rs
+4
View File
@@ -299,6 +299,10 @@ impl RunAgentsExecutor {
Ok(StartAgentOutcome::Started { agent_id }),
_,
)) => RunAgentsAgentOutcomeKind::Launched { agent_id },
futures::future::Either::Left((
Ok(StartAgentOutcome::Completed { agent_id, .. }),
_,
)) => RunAgentsAgentOutcomeKind::Launched { agent_id },
futures::future::Either::Left((
Ok(StartAgentOutcome::Error(error)),
_,
app/src/ai/blocklist/action_model/execute/start_agent.rs
+79 -36
View File
@@ -7,7 +7,7 @@ use galaxyui::{Entity, ModelContext, ModelHandle, SingletonEntity};
use shell_words::split as split_shell_words;
use super::{ActionExecution, AnyActionExecution, ExecuteActionInput, PreprocessActionInput};
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,
@@ -22,6 +22,11 @@ pub enum StartAgentOutcome {
Started {
agent_id: String,
},
/// A direct-provider child completed and returned its output inline.
Completed {
agent_id: String,
output: String,
},
/// An error occurred while starting the agent.
Error(String),
}
@@ -114,6 +119,10 @@ struct PendingStartAgent {
/// Set once the child conversation is synchronously created.
child_conversation_id: Option<AIConversationId>,
sender: async_channel::Sender<StartAgentOutcome>,
/// Direct Bedrock/OpenAI parents do not have a server run id or an
/// orchestration event stream. Keep the tool call open until their local
/// child finishes, then return the child's output inline.
wait_for_completion: bool,
}
pub struct StartAgentExecutor {
@@ -194,6 +203,35 @@ impl StartAgentExecutor {
}
}
fn complete_pending_as_completed(
&mut self,
request_id: StartAgentRequestId,
child_conversation_id: AIConversationId,
ctx: &mut ModelContext<Self>,
) {
let Some(conversation) =
BlocklistAIHistoryModel::as_ref(ctx).conversation(&child_conversation_id)
else {
return;
};
let agent_id = conversation
.orchestration_agent_id()
.or_else(|| {
conversation
.server_conversation_token()
.map(|token| token.as_str().to_string())
})
.unwrap_or_else(|| child_conversation_id.to_string());
let output = extract_child_output(conversation);
let Some(pending) = self.pending.remove(&request_id) else {
return;
};
let _ = pending
.sender
.try_send(StartAgentOutcome::Completed { agent_id, output });
}
fn complete_pending_as_error(
&mut self,
request_id: StartAgentRequestId,
@@ -238,6 +276,14 @@ impl StartAgentExecutor {
self.complete_pending_as_error(request_id, child_conversation_id, error_msg, ctx);
return;
}
let wait_for_completion = self
.pending
.get(&request_id)
.is_some_and(|pending| pending.wait_for_completion);
if wait_for_completion && matches!(conversation.status(), ConversationStatus::Success) {
self.complete_pending_as_completed(request_id, child_conversation_id, ctx);
return;
}
if conversation.orchestration_agent_id().is_some() {
self.complete_pending_as_started(request_id, child_conversation_id, ctx);
}
@@ -256,7 +302,7 @@ impl StartAgentExecutor {
let Some(request_id) = self.find_pending_by_child(conversation_id) else {
return;
};
self.complete_pending_as_started(request_id, *conversation_id, ctx);
self.maybe_complete_pending_for_child_state(request_id, *conversation_id, ctx);
}
BlocklistAIHistoryEvent::UpdatedConversationStatus {
conversation_id, ..
@@ -264,17 +310,7 @@ impl StartAgentExecutor {
let Some(request_id) = self.find_pending_by_child(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().as_deref(),
);
if let Some(error_msg) = error_msg {
self.complete_pending_as_error(request_id, *conversation_id, error_msg, ctx);
}
self.maybe_complete_pending_for_child_state(request_id, *conversation_id, ctx);
}
BlocklistAIHistoryEvent::NewConversationRequestComplete {
request_id,
@@ -346,36 +382,19 @@ impl StartAgentExecutor {
harness_type: None,
model_id,
} => {
// Oz local children resolve their parent's run id from the
// parent conversation. This mirrors the third-party-harness
// and remote-child branches below; the child task row is
// created eagerly at dispatch (see
// `launch_local_no_harness_child`) using this value as the
// `parent_run_id` on `CreateAgentTask`. Bail out if the
// parent has no `run_id` yet — the eager-create path has no
// late-binding fallback (the pre-change lazy path would have
// linked via `Request.metadata.parent_agent_id` later), so
// proceeding would mint an orphan child with no server-side
// parent linkage.
// Server-backed parents launch an Oz child and return its run
// id immediately. Direct Bedrock/OpenAI parents have no run
// id; terminal_pane creates a local hidden child instead and
// this executor waits for its final output.
let parent_run_id = BlocklistAIHistoryModel::as_ref(ctx)
.conversation(&parent_conversation_id)
.and_then(|conversation| conversation.run_id());
let Some(parent_run_id) = parent_run_id else {
return ActionExecution::Sync(AIAgentActionResultType::StartAgent(
StartAgentResult::Error {
error:
"Local Oz child agents require the parent run_id to be available."
.to_string(),
version,
},
));
};
(
StartAgentExecutionMode::Local {
harness_type: None,
model_id,
},
Some(parent_run_id),
parent_run_id,
)
}
StartAgentExecutionMode::Local {
@@ -485,7 +504,7 @@ 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 wait_for_completion = parent_run_id.is_none();
let (sender, receiver) = async_channel::bounded(1);
let request_id = self.next_request_id();
@@ -495,6 +514,7 @@ impl StartAgentExecutor {
parent_conversation_id,
child_conversation_id: None,
sender,
wait_for_completion,
},
);
@@ -518,6 +538,12 @@ impl StartAgentExecutor {
version,
})
}
Ok(StartAgentOutcome::Completed { agent_id, output }) => {
AIAgentActionResultType::StartAgent(StartAgentResult::Success {
agent_id: format!("{agent_id}\n\nAgent output:\n{output}"),
version,
})
}
Ok(StartAgentOutcome::Error(error)) => {
AIAgentActionResultType::StartAgent(StartAgentResult::Error { error, version })
}
@@ -551,6 +577,7 @@ impl StartAgentExecutor {
parent_conversation_id,
child_conversation_id: None,
sender,
wait_for_completion: parent_run_id.is_none(),
},
);
ctx.emit(StartAgentExecutorEvent::CreateAgent(Box::new(
@@ -576,6 +603,22 @@ impl StartAgentExecutor {
}
}
/// Extracts the text output from every completed exchange in a local child.
fn extract_child_output(conversation: &AIConversation) -> String {
let output_parts = conversation
.all_exchanges()
.into_iter()
.filter_map(|exchange| exchange.output_status.output())
.map(|output| output.get().format_for_copy(None))
.filter(|text| !text.is_empty())
.collect::<Vec<_>>();
if output_parts.is_empty() {
"Agent completed but produced no text output.".to_string()
} else {
output_parts.join("\n\n")
}
}
/// Whether a child that failed before launch should have its hidden pane and
/// conversation cleaned up. Only terminal launch failures qualify; recoverable
/// `Blocked` startup states (e.g. awaiting GitHub auth) and non-terminal
app/src/ai/blocklist/action_model/execute/start_agent_tests.rs
+85 -3
View File
@@ -17,9 +17,8 @@ use crate::test_util::settings::initialize_history_persistence_for_tests;
const FIRST_REQUEST_ID: StartAgentRequestId = StartAgentRequestId::from_raw_for_test(0);
/// Stable placeholder run_id assigned to the parent conversation in tests
/// that dispatch an Oz local child. The Oz `Local` arm of
/// `StartAgentExecutor::execute` bails out synchronously if the parent has
/// no `run_id`, so every `local_with_defaults` test needs to assign one.
/// that exercise the server-backed Oz child path. Tests without this id
/// exercise the direct-provider local child path instead.
const PARENT_RUN_ID: &str = "00000000-0000-0000-0000-000000000001";
fn build_start_agent_action(
@@ -375,6 +374,89 @@ fn execute_resolves_success_when_request_linkage_happens_after_child_already_sta
});
}
#[test]
fn execute_waits_for_direct_provider_child_and_returns_its_output() {
App::test((), |mut app| async move {
initialize_history_persistence_for_tests(&mut app);
let terminal_view_id = EntityId::new();
let history_model = app.add_singleton_model(|_| BlocklistAIHistoryModel::new_for_test());
let executor = app.add_model(StartAgentExecutor::new);
let parent_conversation_id = history_model.update(&mut app, |history_model, ctx| {
history_model.start_new_conversation(terminal_view_id, false, false, false, ctx)
});
let action = build_start_agent_action(
StartAgentVersion::V1,
StartAgentExecutionMode::local_with_defaults(),
);
let execution = executor.update(&mut app, |executor, ctx| {
let input = ExecuteActionInput {
action: &action,
conversation_id: parent_conversation_id,
};
let result: AnyActionExecution = executor.execute(input, ctx).into();
result
});
let AnyActionExecution::Async {
execute_future,
on_complete,
} = execution
else {
panic!("expected async execution");
};
let child_conversation_id = history_model.update(&mut app, |history_model, ctx| {
history_model.start_new_child_conversation(
terminal_view_id,
"Agent 1".to_string(),
parent_conversation_id,
None,
ctx,
)
});
history_model.update(&mut app, |model, ctx| {
model.record_new_conversation_request_complete(
FIRST_REQUEST_ID,
child_conversation_id,
ctx,
);
});
executor.read(&app, |executor, _| {
let pending = executor
.pending
.get(&FIRST_REQUEST_ID)
.expect("direct child should remain pending until completion");
assert!(pending.wait_for_completion);
});
history_model.update(&mut app, |history_model, ctx| {
history_model.update_conversation_status(
terminal_view_id,
child_conversation_id,
ConversationStatus::Success,
ctx,
);
});
let async_result = execute_future.await;
let result = app.update(|ctx| on_complete(async_result, ctx));
assert!(matches!(
result,
AIAgentActionResultType::StartAgent(StartAgentResult::Success {
agent_id,
version,
}) if agent_id.contains(&child_conversation_id.to_string())
&& agent_id.contains("Agent output:")
&& agent_id.contains("Agent completed but produced no text output.")
&& version == StartAgentVersion::V1
));
executor.read(&app, |executor, _| {
assert!(executor.pending.is_empty());
});
});
}
#[test]
fn execute_returns_detailed_error_when_child_startup_fails_before_initialization() {
App::test((), |mut app| async move {