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

Wait for direct-provider child agents

Browse Source
This commit is contained in:
rkw6086
2026-08-15 19:23:28 -05:00
parent d10deb80a2
commit 93d6172072
11 changed files with 1096 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/ai/blocklist/action_model/execute.rs
+3 -2
View File
@@ -355,8 +355,9 @@ impl BlocklistAIActionExecutor {
let read_skill_executor = ctx.add_model(|_| ReadSkillExecutor::new(active_session.clone()));
let fetch_conversation_executor = ctx.add_model(|_| FetchConversationExecutor::new());
let start_agent_executor = ctx.add_model(StartAgentExecutor::new);
let run_agents_executor = ctx
.add_model(|_| RunAgentsExecutor::new(start_agent_executor.clone(), terminal_view_id));
let run_agents_executor = ctx.add_model(|ctx| {
RunAgentsExecutor::new(start_agent_executor.clone(), terminal_view_id, ctx)
});
let send_message_executor = ctx.add_model(|_| SendMessageToAgentExecutor::new());
let ask_user_question_executor =
ctx.add_model(|_| AskUserQuestionExecutor::new(terminal_view_id));
app/src/ai/blocklist/action_model/execute/run_agents.rs
+90 -50
View File
@@ -12,14 +12,17 @@ use ai::agent::action_result::{
};
use ai::agent::orchestration_config::OrchestrationConfig;
use ai::skills::SkillReference;
use futures::future::BoxFuture;
use futures::future::{join_all, BoxFuture};
use futures::FutureExt;
use galaxy_core::execution_mode::AppExecutionMode;
use settings::Setting;
use warp_cli::agent::Harness;
use warpui::{Entity, EntityId, ModelContext, ModelHandle, SingletonEntity};
use super::start_agent::{StartAgentExecutor, StartAgentOutcome};
use super::start_agent::{
StartAgentDispatch, StartAgentExecutor, StartAgentExecutorEvent, StartAgentOutcome,
StartAgentWaitPolicy,
};
use super::{
child_agent_delegation_denial_reason, ActionExecution, AnyActionExecution, ExecuteActionInput,
PreprocessActionInput,
@@ -80,6 +83,12 @@ pub enum RunAgentsExecutorEvent {
SpawningFinished {
action_id: AIAgentActionId,
},
ChildConversationCreated {
action_id: AIAgentActionId,
agent_name: String,
parent_conversation_id: AIConversationId,
child_conversation_id: AIConversationId,
},
}
impl Entity for RunAgentsExecutor {
@@ -90,7 +99,24 @@ impl RunAgentsExecutor {
pub fn new(
start_agent_executor: ModelHandle<StartAgentExecutor>,
terminal_view_id: EntityId,
ctx: &mut ModelContext<Self>,
) -> Self {
ctx.subscribe_to_model(&start_agent_executor, |_, _, event, ctx| {
if let StartAgentExecutorEvent::RunAgentsChildConversationCreated {
action_id,
agent_name,
parent_conversation_id,
child_conversation_id,
} = event
{
ctx.emit(RunAgentsExecutorEvent::ChildConversationCreated {
action_id: action_id.clone(),
agent_name: agent_name.clone(),
parent_conversation_id: *parent_conversation_id,
child_conversation_id: *child_conversation_id,
});
}
});
Self {
pending: HashMap::new(),
launched_agents: HashMap::new(),
@@ -103,6 +129,10 @@ impl RunAgentsExecutor {
self.pending.contains_key(action_id)
}
pub(crate) fn terminal_view_id(&self) -> EntityId {
self.terminal_view_id
}
/// Cancels a pending run so publication completion cannot fan out children.
pub(super) fn cancel_execution(
&mut self,
@@ -361,8 +391,9 @@ impl RunAgentsExecutor {
"execution_mode": start_agent_execution_mode_label(&mode),
}),
);
let recv = self.start_agent_executor.update(ctx, |executor, exec_ctx| {
let dispatch = self.start_agent_executor.update(ctx, |executor, exec_ctx| {
executor.dispatch(
action_id.clone(),
cfg.name.clone(),
prompt,
mode,
@@ -372,7 +403,7 @@ impl RunAgentsExecutor {
exec_ctx,
)
});
slots.push(ChildSlot::Pending(recv));
slots.push(ChildSlot::Pending(dispatch));
}
let agent_run_configs_for_result = agent_run_configs.clone();
@@ -393,49 +424,9 @@ impl RunAgentsExecutor {
ctx.spawn(
async move {
let mut outcomes: Vec<RunAgentsAgentOutcomeKind> = Vec::with_capacity(slots.len());
for (slot_index, slot) in slots.into_iter().enumerate() {
let kind = match slot {
ChildSlot::Failed(error) => RunAgentsAgentOutcomeKind::Failed { error },
ChildSlot::Pending(recv) => {
let timeout = warpui::r#async::Timer::after(SPAWN_TIMEOUT);
match futures::future::select(Box::pin(recv.recv()), Box::pin(timeout))
.await
{
futures::future::Either::Left((
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)),
_,
)) => RunAgentsAgentOutcomeKind::Failed { error },
futures::future::Either::Left((Err(_), _)) => {
RunAgentsAgentOutcomeKind::Failed {
error: "Cancelled before launch".to_string(),
}
}
futures::future::Either::Right((_, _)) => {
log::warn!(
"Agent spawn timed out after {} seconds",
SPAWN_TIMEOUT.as_secs()
);
RunAgentsAgentOutcomeKind::Failed {
error: format!(
"Agent failed to start within {} seconds. \
The harness binary may not be installed.",
SPAWN_TIMEOUT.as_secs()
),
}
}
}
}
};
#[cfg(not(target_family = "wasm"))]
let outcomes = join_all(slots.into_iter().map(resolve_child_slot)).await;
#[cfg(not(target_family = "wasm"))]
for (slot_index, kind) in outcomes.iter().enumerate() {
log::info!(
"RunAgents child launch outcome action_id={} parent_conversation_id={} \
agent_name={} slot_index={} outcome={}",
@@ -446,9 +437,8 @@ impl RunAgentsExecutor {
.map(String::as_str)
.unwrap_or("<unknown>"),
slot_index,
run_agents_agent_outcome_kind_label(&kind)
run_agents_agent_outcome_kind_label(kind)
);
outcomes.push(kind);
}
outcomes
},
@@ -651,7 +641,57 @@ fn run_agents_agent_outcome_kind_label(kind: &RunAgentsAgentOutcomeKind) -> &'st
enum ChildSlot {
Failed(String),
Pending(async_channel::Receiver<StartAgentOutcome>),
Pending(StartAgentDispatch),
}
async fn resolve_child_slot(slot: ChildSlot) -> RunAgentsAgentOutcomeKind {
resolve_child_slot_with_timeout(slot, SPAWN_TIMEOUT).await
}
async fn resolve_child_slot_with_timeout(
slot: ChildSlot,
spawn_timeout: Duration,
) -> RunAgentsAgentOutcomeKind {
let dispatch = match slot {
ChildSlot::Failed(error) => return RunAgentsAgentOutcomeKind::Failed { error },
ChildSlot::Pending(dispatch) => dispatch,
};
let outcome = match dispatch.wait_policy {
StartAgentWaitPolicy::Completion => dispatch.receiver.recv().await.ok(),
StartAgentWaitPolicy::Startup => {
let timeout = warpui::r#async::Timer::after(spawn_timeout);
match futures::future::select(Box::pin(dispatch.receiver.recv()), Box::pin(timeout))
.await
{
futures::future::Either::Left((outcome, _)) => outcome.ok(),
futures::future::Either::Right((_, _)) => {
log::warn!(
"Agent spawn timed out after {} seconds",
spawn_timeout.as_secs()
);
return RunAgentsAgentOutcomeKind::Failed {
error: format!(
"Agent failed to start within {} seconds. \
The harness binary may not be installed.",
spawn_timeout.as_secs()
),
};
}
}
}
};
match outcome {
Some(StartAgentOutcome::Started { agent_id })
| Some(StartAgentOutcome::Completed { agent_id, .. }) => {
RunAgentsAgentOutcomeKind::Launched { agent_id }
}
Some(StartAgentOutcome::Error(error)) => RunAgentsAgentOutcomeKind::Failed { error },
None => RunAgentsAgentOutcomeKind::Failed {
error: "Child agent was cancelled before completion".to_string(),
},
}
}
fn approved_orchestration_config_can_autoexecute(
app/src/ai/blocklist/action_model/execute/run_agents_tests.rs
+110 -2
View File
@@ -358,6 +358,113 @@ fn validate_request_rejects_remote_dispatch() {
);
}
#[test]
fn completion_slots_are_polled_concurrently_and_preserve_request_order() {
App::test((), |_app| async move {
let (first_sender, first_receiver) = async_channel::bounded(1);
let (second_sender, second_receiver) = async_channel::bounded(1);
let slots = vec![
ChildSlot::Pending(StartAgentDispatch {
receiver: first_receiver,
wait_policy: StartAgentWaitPolicy::Completion,
}),
ChildSlot::Pending(StartAgentDispatch {
receiver: second_receiver,
wait_policy: StartAgentWaitPolicy::Completion,
}),
ChildSlot::Failed("prelaunch failure".to_string()),
];
let mut outcomes =
Box::pin(join_all(slots.into_iter().map(|slot| {
resolve_child_slot_with_timeout(slot, Duration::from_millis(1))
})));
second_sender
.try_send(StartAgentOutcome::Completed {
agent_id: "second-agent".to_string(),
output: "done".to_string(),
})
.unwrap();
assert!(futures::poll!(&mut outcomes).is_pending());
assert!(
!second_sender.is_full(),
"join_all should poll and drain the second slot while the first is pending"
);
first_sender
.try_send(StartAgentOutcome::Error("first failed".to_string()))
.unwrap();
let outcomes = outcomes.await;
assert!(matches!(
&outcomes[0],
RunAgentsAgentOutcomeKind::Failed { error } if error == "first failed"
));
assert!(matches!(
&outcomes[1],
RunAgentsAgentOutcomeKind::Launched { agent_id } if agent_id == "second-agent"
));
assert!(matches!(
&outcomes[2],
RunAgentsAgentOutcomeKind::Failed { error } if error == "prelaunch failure"
));
});
}
#[test]
fn completion_wait_ignores_spawn_timeout() {
App::test((), |_app| async move {
let (sender, receiver) = async_channel::bounded(1);
let completion = Box::pin(resolve_child_slot_with_timeout(
ChildSlot::Pending(StartAgentDispatch {
receiver,
wait_policy: StartAgentWaitPolicy::Completion,
}),
Duration::from_millis(1),
));
let wait = warpui::r#async::Timer::after(Duration::from_millis(20));
let completion = match futures::future::select(completion, Box::pin(wait)).await {
futures::future::Either::Left((outcome, _)) => {
panic!("completion wait unexpectedly resolved before child completion: {outcome:?}")
}
futures::future::Either::Right((_, completion)) => completion,
};
sender
.try_send(StartAgentOutcome::Completed {
agent_id: "child-agent".to_string(),
output: "done".to_string(),
})
.unwrap();
assert!(matches!(
completion.await,
RunAgentsAgentOutcomeKind::Launched { agent_id } if agent_id == "child-agent"
));
});
}
#[test]
fn startup_wait_retains_spawn_timeout() {
App::test((), |_app| async move {
let (_sender, receiver) = async_channel::bounded(1);
let outcome = resolve_child_slot_with_timeout(
ChildSlot::Pending(StartAgentDispatch {
receiver,
wait_policy: StartAgentWaitPolicy::Startup,
}),
Duration::from_millis(1),
)
.await;
assert!(matches!(
outcome,
RunAgentsAgentOutcomeKind::Failed { error }
if error.contains("Agent failed to start within")
));
});
}
fn initialize_run_agents_test(app: &mut App, mode: ExecutionMode) -> RunAgentsTestState {
initialize_settings_for_tests_with_mode(app, mode, false);
let global_resource_handles = GlobalResourceHandles::mock(app);
@@ -389,8 +496,9 @@ fn initialize_run_agents_test(app: &mut App, mode: ExecutionMode) -> RunAgentsTe
history_model.start_new_conversation(terminal_view_id, false, false, false, ctx)
});
let start_agent_executor = app.add_model(StartAgentExecutor::new);
let executor =
app.add_model(|_| RunAgentsExecutor::new(start_agent_executor.clone(), terminal_view_id));
let executor = app.add_model(|ctx| {
RunAgentsExecutor::new(start_agent_executor.clone(), terminal_view_id, ctx)
});
RunAgentsTestState {
conversation_id,
app/src/ai/blocklist/action_model/execute/start_agent.rs
+157 -57
View File
@@ -30,10 +30,23 @@ pub enum StartAgentOutcome {
agent_id: String,
output: String,
},
/// An error occurred while starting the agent.
/// An error occurred while starting or running the agent.
Error(String),
}
/// Determines whether a dispatch receiver acknowledges startup or waits for a
/// direct-provider child to reach a terminal state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StartAgentWaitPolicy {
Startup,
Completion,
}
pub struct StartAgentDispatch {
pub receiver: async_channel::Receiver<StartAgentOutcome>,
pub wait_policy: StartAgentWaitPolicy,
}
fn invalid_local_child_harness_error(harness_type: &str) -> String {
let harness_name = harness_type.trim();
if harness_name.is_empty() {
@@ -118,9 +131,10 @@ pub struct StartAgentRequest {
}
struct PendingStartAgent {
/// Present for standalone StartAgent tool calls. RunAgents dispatches use
/// the same executor but do not have a one-to-one StartAgent action card.
action_id: Option<AIAgentActionId>,
action_id: AIAgentActionId,
/// Present when RunAgents owns this dispatch. Standalone StartAgent calls
/// use the action id only for their one-to-one inline child panel.
run_agents_child_name: Option<String>,
parent_conversation_id: AIConversationId,
/// Set once the child conversation is synchronously created.
child_conversation_id: Option<AIConversationId>,
@@ -128,7 +142,7 @@ struct PendingStartAgent {
/// 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,
wait_policy: StartAgentWaitPolicy,
}
pub struct StartAgentExecutor {
@@ -161,34 +175,33 @@ impl StartAgentExecutor {
child_conversation_id: AIConversationId,
ctx: &mut ModelContext<Self>,
) {
let direct_provider_panel_link = {
let child_link_event = {
let Some(pending) = self.pending.get_mut(&request_id) else {
return;
};
pending.child_conversation_id = Some(child_conversation_id);
if pending.wait_for_completion {
pending.action_id.clone().map(|action_id| {
(
action_id,
pending.parent_conversation_id,
child_conversation_id,
)
if let Some(agent_name) = pending.run_agents_child_name.clone() {
Some(StartAgentExecutorEvent::RunAgentsChildConversationCreated {
action_id: pending.action_id.clone(),
agent_name,
parent_conversation_id: pending.parent_conversation_id,
child_conversation_id,
})
} else if matches!(pending.wait_policy, StartAgentWaitPolicy::Completion) {
Some(
StartAgentExecutorEvent::DirectProviderChildConversationCreated {
action_id: pending.action_id.clone(),
parent_conversation_id: pending.parent_conversation_id,
child_conversation_id,
},
)
} else {
None
}
};
if let Some((action_id, parent_conversation_id, child_conversation_id)) =
direct_provider_panel_link
{
ctx.emit(
StartAgentExecutorEvent::DirectProviderChildConversationCreated {
action_id,
parent_conversation_id,
child_conversation_id,
},
);
if let Some(event) = child_link_event {
ctx.emit(event);
}
self.maybe_complete_pending_for_child_state(request_id, child_conversation_id, ctx);
}
@@ -274,14 +287,15 @@ impl StartAgentExecutor {
return;
};
let _ = pending.sender.try_send(StartAgentOutcome::Error(error_msg));
// A child that reaches `complete_pending_as_error` never obtained an
// agent id, so it failed at the launch stage. Clean up its hidden
// pane + conversation so the orchestration pill bar does not retain a
// dead chip — but only for terminal failures, leaving recoverable
// `Blocked` startup states (e.g. awaiting GitHub auth) intact.
let should_cleanup = BlocklistAIHistoryModel::as_ref(ctx)
.conversation(&child_conversation_id)
.is_some_and(|conversation| should_cleanup_failed_child_launch(conversation.status()));
// Only startup acknowledgements may clean up a conversation that never
// initialized. Direct-provider completion waits preserve the terminal
// child so its transcript and failure remain inspectable.
let should_cleanup = matches!(pending.wait_policy, StartAgentWaitPolicy::Startup)
&& BlocklistAIHistoryModel::as_ref(ctx)
.conversation(&child_conversation_id)
.is_some_and(|conversation| {
should_cleanup_failed_child_launch(conversation.status())
});
if should_cleanup {
ctx.emit(StartAgentExecutorEvent::CleanupFailedChildLaunch {
conversation_id: child_conversation_id,
@@ -300,23 +314,49 @@ impl StartAgentExecutor {
else {
return;
};
if let Some(error_msg) = start_agent_error_message_for_status(
conversation.status(),
conversation.status_error_message().as_deref(),
) {
self.complete_pending_as_error(request_id, child_conversation_id, error_msg, ctx);
return;
}
let wait_for_completion = self
let wait_policy = 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);
.map(|pending| pending.wait_policy);
match wait_policy {
Some(StartAgentWaitPolicy::Completion) => match conversation.status() {
ConversationStatus::Success => {
self.complete_pending_as_completed(request_id, child_conversation_id, ctx);
}
ConversationStatus::Error | ConversationStatus::Cancelled => {
let error_msg = direct_child_error_message_for_status(
conversation.status(),
conversation.status_error_message().as_deref(),
)
.expect("terminal direct child status should produce an error");
self.complete_pending_as_error(
request_id,
child_conversation_id,
error_msg,
ctx,
);
}
ConversationStatus::InProgress
| ConversationStatus::TransientError
| ConversationStatus::Blocked { .. }
| ConversationStatus::WaitingForEvents => {}
},
Some(StartAgentWaitPolicy::Startup) => {
if let Some(error_msg) = start_agent_startup_error_message_for_status(
conversation.status(),
conversation.status_error_message().as_deref(),
) {
self.complete_pending_as_error(
request_id,
child_conversation_id,
error_msg,
ctx,
);
} else if conversation.orchestration_agent_id().is_some() {
self.complete_pending_as_started(request_id, child_conversation_id, ctx);
}
}
None => {}
}
}
@@ -349,6 +389,22 @@ impl StartAgentExecutor {
} => {
self.record_child_conversation(*request_id, *conversation_id, ctx);
}
BlocklistAIHistoryEvent::RemoveConversation {
conversation_id, ..
}
| BlocklistAIHistoryEvent::DeletedConversation {
conversation_id, ..
} => {
let Some(request_id) = self.find_pending_by_child(conversation_id) else {
return;
};
let Some(pending) = self.pending.remove(&request_id) else {
return;
};
let _ = pending.sender.try_send(StartAgentOutcome::Error(
"Child agent conversation was removed by the user.".to_string(),
));
}
BlocklistAIHistoryEvent::StartedNewConversation { .. }
| BlocklistAIHistoryEvent::CreatedSubtask { .. }
| BlocklistAIHistoryEvent::UpgradedTask { .. }
@@ -361,8 +417,6 @@ impl StartAgentExecutor {
| BlocklistAIHistoryEvent::UpdatedTodoList { .. }
| BlocklistAIHistoryEvent::UpdatedAutoexecuteOverride { .. }
| BlocklistAIHistoryEvent::SplitConversation { .. }
| BlocklistAIHistoryEvent::RemoveConversation { .. }
| BlocklistAIHistoryEvent::DeletedConversation { .. }
| BlocklistAIHistoryEvent::RestoredConversations { .. }
| BlocklistAIHistoryEvent::UpdatedConversationTitle { .. }
| BlocklistAIHistoryEvent::UpdatedConversationMetadata { .. }
@@ -543,18 +597,23 @@ 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_policy = if parent_run_id.is_none() {
StartAgentWaitPolicy::Completion
} else {
StartAgentWaitPolicy::Startup
};
let (sender, receiver) = async_channel::bounded(1);
let request_id = self.next_request_id();
self.pending.insert(
request_id,
PendingStartAgent {
action_id: Some(action_id),
action_id,
run_agents_child_name: None,
parent_conversation_id,
child_conversation_id: None,
sender,
wait_for_completion,
wait_policy,
},
);
@@ -599,6 +658,7 @@ impl StartAgentExecutor {
#[allow(clippy::too_many_arguments)]
pub fn dispatch(
&mut self,
action_id: AIAgentActionId,
name: String,
prompt: String,
execution_mode: StartAgentExecutionMode,
@@ -606,11 +666,19 @@ impl StartAgentExecutor {
parent_conversation_id: AIConversationId,
parent_run_id: Option<String>,
ctx: &mut ModelContext<Self>,
) -> async_channel::Receiver<StartAgentOutcome> {
) -> StartAgentDispatch {
let wait_policy = if parent_run_id.is_none() {
StartAgentWaitPolicy::Completion
} else {
StartAgentWaitPolicy::Startup
};
let (sender, receiver) = async_channel::bounded(1);
if let Some(error) = child_agent_delegation_denial_reason(parent_conversation_id, ctx) {
let _ = sender.try_send(StartAgentOutcome::Error(error));
return receiver;
return StartAgentDispatch {
receiver,
wait_policy,
};
}
let (prompt, execution_mode) =
@@ -620,11 +688,12 @@ impl StartAgentExecutor {
self.pending.insert(
request_id,
PendingStartAgent {
action_id: None,
action_id,
run_agents_child_name: Some(name.clone()),
parent_conversation_id,
child_conversation_id: None,
sender,
wait_for_completion: parent_run_id.is_none(),
wait_policy,
},
);
ctx.emit(StartAgentExecutorEvent::CreateAgent(Box::new(
@@ -638,7 +707,10 @@ impl StartAgentExecutor {
parent_run_id,
},
)));
receiver
StartAgentDispatch {
receiver,
wait_policy,
}
}
pub(super) fn preprocess_action(
@@ -682,7 +754,7 @@ fn should_cleanup_failed_child_launch(status: &ConversationStatus) -> bool {
}
}
fn start_agent_error_message_for_status(
fn start_agent_startup_error_message_for_status(
status: &ConversationStatus,
error_message: Option<&str>,
) -> Option<String> {
@@ -717,6 +789,26 @@ fn start_agent_error_message_for_status(
}
}
fn direct_child_error_message_for_status(
status: &ConversationStatus,
error_message: Option<&str>,
) -> Option<String> {
match status {
ConversationStatus::Error => Some(
error_message
.filter(|message| !message.trim().is_empty())
.unwrap_or("Child agent failed")
.to_string(),
),
ConversationStatus::Cancelled => Some("Child agent was cancelled by the user.".to_string()),
ConversationStatus::InProgress
| ConversationStatus::TransientError
| ConversationStatus::Success
| ConversationStatus::Blocked { .. }
| ConversationStatus::WaitingForEvents => None,
}
}
impl Entity for StartAgentExecutor {
type Event = StartAgentExecutorEvent;
}
@@ -731,6 +823,14 @@ pub enum StartAgentExecutorEvent {
parent_conversation_id: AIConversationId,
child_conversation_id: AIConversationId,
},
/// A RunAgents child conversation is available for live status and
/// navigation in the owning action card.
RunAgentsChildConversationCreated {
action_id: AIAgentActionId,
agent_name: String,
parent_conversation_id: AIConversationId,
child_conversation_id: AIConversationId,
},
/// A child agent failed at the launch stage (never started a server-side
/// run). The owning terminal view removes its hidden pane and conversation
/// so the orchestration pill bar does not retain a dead chip.
app/src/ai/blocklist/action_model/execute/start_agent_tests.rs
+264 -3
View File
@@ -35,6 +35,15 @@ impl Entity for CapturedStartAgentPrompts {
type Event = ();
}
#[derive(Default)]
struct CapturedRunAgentsChildLinks(
Vec<(AIAgentActionId, String, AIConversationId, AIConversationId)>,
);
impl Entity for CapturedRunAgentsChildLinks {
type Event = ();
}
fn capture_start_agent_prompts(
app: &mut App,
executor: &ModelHandle<StartAgentExecutor>,
@@ -74,6 +83,32 @@ fn capture_direct_provider_child_links(
captured
}
fn capture_run_agents_child_links(
app: &mut App,
executor: &ModelHandle<StartAgentExecutor>,
) -> ModelHandle<CapturedRunAgentsChildLinks> {
let captured = app.add_model(|_| CapturedRunAgentsChildLinks::default());
captured.update(app, |_, ctx| {
ctx.subscribe_to_model(executor, |captured, _, event, _ctx| {
if let StartAgentExecutorEvent::RunAgentsChildConversationCreated {
action_id,
agent_name,
parent_conversation_id,
child_conversation_id,
} = event
{
captured.0.push((
action_id.clone(),
agent_name.clone(),
*parent_conversation_id,
*child_conversation_id,
));
}
});
});
captured
}
fn build_start_agent_action(
version: StartAgentVersion,
execution_mode: StartAgentExecutionMode,
@@ -199,8 +234,9 @@ fn dispatch_denies_child_conversation_defense_in_depth() {
conversation_id
});
let receiver = executor.update(&mut app, |executor, ctx| {
let dispatch = executor.update(&mut app, |executor, ctx| {
executor.dispatch(
AIAgentActionId::from("run-agents-action".to_string()),
"grandchild".to_string(),
"Do more work".to_string(),
StartAgentExecutionMode::local_with_defaults(),
@@ -212,7 +248,7 @@ fn dispatch_denies_child_conversation_defense_in_depth() {
});
assert!(matches!(
receiver.try_recv(),
dispatch.receiver.try_recv(),
Ok(StartAgentOutcome::Error(error)) if error.contains("leaf workers")
));
executor.read(&app, |executor, _ctx| {
@@ -685,6 +721,231 @@ fn hosted_child_link_does_not_publish_direct_provider_panel_event() {
});
}
struct PendingDirectProviderChild {
action_id: AIAgentActionId,
parent_conversation_id: AIConversationId,
history_model: ModelHandle<BlocklistAIHistoryModel>,
executor: ModelHandle<StartAgentExecutor>,
captured_cleanup: ModelHandle<CapturedCleanupEvents>,
direct_links: ModelHandle<CapturedDirectProviderChildLinks>,
run_agents_links: ModelHandle<CapturedRunAgentsChildLinks>,
terminal_view_id: EntityId,
child_conversation_id: AIConversationId,
dispatch: StartAgentDispatch,
}
fn dispatch_pending_direct_provider_child(app: &mut App) -> PendingDirectProviderChild {
initialize_history_persistence_for_tests(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 captured_cleanup = app.add_model(|_| CapturedCleanupEvents::default());
captured_cleanup.update(app, |_, ctx| {
ctx.subscribe_to_model(&executor, |captured, _, event, _ctx| {
if let StartAgentExecutorEvent::CleanupFailedChildLaunch { conversation_id } = event {
captured.0.push(*conversation_id);
}
});
});
let direct_links = capture_direct_provider_child_links(app, &executor);
let run_agents_links = capture_run_agents_child_links(app, &executor);
let parent_conversation_id = history_model.update(app, |history_model, ctx| {
history_model.start_new_conversation(terminal_view_id, false, false, false, ctx)
});
let action_id = AIAgentActionId::from("run-agents-action".to_string());
let dispatch = executor.update(app, |executor, ctx| {
executor.dispatch(
action_id.clone(),
"child".to_string(),
"Investigate the failure".to_string(),
StartAgentExecutionMode::local_with_defaults(),
None,
parent_conversation_id,
None,
ctx,
)
});
let child_conversation_id = history_model.update(app, |history_model, ctx| {
history_model.start_new_child_conversation(
terminal_view_id,
"child".to_string(),
parent_conversation_id,
None,
ctx,
)
});
history_model.update(app, |history_model, ctx| {
history_model.record_new_conversation_request_complete(
FIRST_REQUEST_ID,
child_conversation_id,
ctx,
);
});
PendingDirectProviderChild {
action_id,
parent_conversation_id,
history_model,
executor,
captured_cleanup,
direct_links,
run_agents_links,
terminal_view_id,
child_conversation_id,
dispatch,
}
}
#[test]
fn direct_provider_nonterminal_states_remain_pending_until_cancelled() {
App::test((), |mut app| async move {
let state = dispatch_pending_direct_provider_child(&mut app);
assert_eq!(state.dispatch.wait_policy, StartAgentWaitPolicy::Completion);
for status in [
ConversationStatus::Blocked {
blocked_action: "Waiting for user input".to_string(),
},
ConversationStatus::TransientError,
ConversationStatus::WaitingForEvents,
] {
state.history_model.update(&mut app, |history_model, ctx| {
history_model.update_conversation_status(
state.terminal_view_id,
state.child_conversation_id,
status,
ctx,
);
});
assert!(matches!(
state.dispatch.receiver.try_recv(),
Err(async_channel::TryRecvError::Empty)
));
state.executor.read(&app, |executor, _| {
assert!(executor.pending.contains_key(&FIRST_REQUEST_ID));
});
}
state.history_model.update(&mut app, |history_model, ctx| {
history_model.update_conversation_status(
state.terminal_view_id,
state.child_conversation_id,
ConversationStatus::Cancelled,
ctx,
);
});
assert!(matches!(
state.dispatch.receiver.try_recv(),
Ok(StartAgentOutcome::Error(error))
if error == "Child agent was cancelled by the user."
));
state.executor.read(&app, |executor, _| {
assert!(executor.pending.is_empty());
});
state.captured_cleanup.read(&app, |captured, _| {
assert!(captured.0.is_empty());
});
});
}
#[test]
fn direct_provider_error_preserves_child_for_inspection() {
App::test((), |mut app| async move {
let state = dispatch_pending_direct_provider_child(&mut app);
state.history_model.update(&mut app, |history_model, ctx| {
history_model.update_conversation_status_with_error(
state.terminal_view_id,
state.child_conversation_id,
ConversationStatus::Error,
Some(RenderableAIError::other("Child execution failed", false)),
ctx,
);
});
assert!(matches!(
state.dispatch.receiver.try_recv(),
Ok(StartAgentOutcome::Error(error)) if error == "Child execution failed"
));
state.captured_cleanup.read(&app, |captured, _| {
assert!(captured.0.is_empty());
});
state.history_model.read(&app, |history_model, _| {
assert!(history_model
.conversation(&state.child_conversation_id)
.is_some());
});
});
}
#[test]
fn removing_direct_provider_child_resolves_pending_wait() {
App::test((), |mut app| async move {
let state = dispatch_pending_direct_provider_child(&mut app);
state.history_model.update(&mut app, |history_model, ctx| {
history_model.remove_conversation(
state.child_conversation_id,
state.terminal_view_id,
ctx,
);
});
assert!(matches!(
state.dispatch.receiver.try_recv(),
Ok(StartAgentOutcome::Error(error))
if error == "Child agent conversation was removed by the user."
));
state.executor.read(&app, |executor, _| {
assert!(executor.pending.is_empty());
});
});
}
#[test]
fn deleting_direct_provider_child_resolves_pending_wait() {
App::test((), |mut app| async move {
let state = dispatch_pending_direct_provider_child(&mut app);
state.history_model.update(&mut app, |history_model, ctx| {
history_model.delete_conversation(
state.child_conversation_id,
Some(state.terminal_view_id),
ctx,
);
});
assert!(matches!(
state.dispatch.receiver.try_recv(),
Ok(StartAgentOutcome::Error(error))
if error == "Child agent conversation was removed by the user."
));
state.executor.read(&app, |executor, _| {
assert!(executor.pending.is_empty());
});
});
}
#[test]
fn run_agents_dispatch_publishes_only_run_agents_child_link() {
App::test((), |mut app| async move {
let state = dispatch_pending_direct_provider_child(&mut app);
state.direct_links.read(&app, |captured, _| {
assert!(captured.0.is_empty());
});
state.run_agents_links.read(&app, |captured, _| {
assert_eq!(
captured.0,
vec![(
state.action_id.clone(),
"child".to_string(),
state.parent_conversation_id,
state.child_conversation_id,
)]
);
});
});
}
#[test]
fn execute_waits_for_direct_provider_child_and_returns_its_output() {
App::test((), |mut app| async move {
@@ -738,7 +999,7 @@ fn execute_waits_for_direct_provider_child_and_returns_its_output() {
.pending
.get(&FIRST_REQUEST_ID)
.expect("direct child should remain pending until completion");
assert!(pending.wait_for_completion);
assert_eq!(pending.wait_policy, StartAgentWaitPolicy::Completion);
});
history_model.update(&mut app, |history_model, ctx| {