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

Yield orchestrators for child groups and correct subscription context catalogs

Browse Source
This commit is contained in:
rkw6086
2026-09-14 22:32:18 -05:00
parent 506dc5b225
commit 72cd38b54e
27 changed files with 741 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/ai/blocklist/action_model/execute/run_agents.rs
+39 -4
View File
@@ -34,6 +34,7 @@ use crate::ai::agent::{
};
use crate::ai::auth_secret_types::auth_secret_types_for_harness;
use crate::ai::blocklist::inline_action::orchestration_controls::OrchestrationEditState;
use crate::ai::blocklist::orchestration_events::OrchestrationEventService;
use crate::ai::blocklist::{BlocklistAIHistoryModel, BlocklistAIPermissions};
use crate::ai::cloud_agent_settings::CloudAgentSettings;
use crate::ai::document::plan_publication::{
@@ -180,6 +181,9 @@ impl RunAgentsExecutor {
"RunAgents cancellation detached {detached_dispatches} pending child dispatch(es) for action {action_id}"
);
if self.pending.remove(&action_key).is_some() {
OrchestrationEventService::handle(ctx).update(ctx, |service, ctx| {
service.seal_local_group(conversation_id, &action_id.to_string(), Vec::new(), ctx);
});
ctx.emit(RunAgentsExecutorEvent::SpawningFinished {
conversation_id,
action_id: action_id.clone(),
@@ -423,15 +427,30 @@ impl RunAgentsExecutor {
}),
);
if matches!(run_execution_mode, RunAgentsExecutionMode::Local) {
OrchestrationEventService::handle(ctx).update(ctx, |service, ctx| {
service.begin_local_group(parent_conversation_id, &action_id.to_string(), ctx);
});
}
let mut slots: Vec<ChildSlot> = Vec::with_capacity(agent_run_configs.len());
let wait_policy = match &run_execution_mode {
RunAgentsExecutionMode::Local => StartAgentWaitPolicy::Completion,
RunAgentsExecutionMode::Local => StartAgentWaitPolicy::LocalStartup,
RunAgentsExecutionMode::Remote { .. } => StartAgentWaitPolicy::Startup,
};
for cfg in &agent_run_configs {
let normalized_name = normalize_agent_name(&cfg.name)
.expect("validated RunAgents requests have non-empty agent names");
if let Some(child_conversation_id) = recovery_children.remove(&normalized_name) {
if matches!(run_execution_mode, RunAgentsExecutionMode::Local) {
OrchestrationEventService::handle(ctx).update(ctx, |service, ctx| {
service.register_local_group_child(
parent_conversation_id,
&action_id.to_string(),
child_conversation_id,
ctx,
);
});
}
let dispatch = self.start_agent_executor.update(ctx, |executor, exec_ctx| {
executor.reattach(
action_id.clone(),
@@ -509,7 +528,7 @@ impl RunAgentsExecutor {
}),
);
let dispatch = self.start_agent_executor.update(ctx, |executor, exec_ctx| {
executor.dispatch(
let mut dispatch = executor.dispatch(
action_id.clone(),
cfg.name.clone(),
prompt,
@@ -518,7 +537,11 @@ impl RunAgentsExecutor {
parent_conversation_id,
parent_run_id.clone(),
exec_ctx,
)
);
if matches!(run_execution_mode, RunAgentsExecutionMode::Local) {
executor.yield_after_local_start(&mut dispatch);
}
dispatch
});
slots.push(ChildSlot::Pending(dispatch));
}
@@ -583,6 +606,16 @@ impl RunAgentsExecutor {
})
.collect();
me.record_launched_agents(parent_conversation_id_for_result, &agents);
if matches!(run_execution_mode_for_aggr, RunAgentsExecutionMode::Local) {
let errors = agents.iter().filter_map(|agent| match &agent.kind {
RunAgentsAgentOutcomeKind::Failed { error } => Some(format!("{}: {error}", agent.name)),
RunAgentsAgentOutcomeKind::Launched { .. } | RunAgentsAgentOutcomeKind::Completed { .. } => None,
}).collect();
OrchestrationEventService::handle(ctx).update(ctx, |service, ctx| {
service.seal_local_group(parent_conversation_id_for_result, &action_id_for_aggr.to_string(), errors, ctx);
});
}
#[cfg(not(target_family = "wasm"))]
log_run_agents_event(
ctx,
@@ -838,7 +871,9 @@ async fn resolve_child_slot_with_timeout(
let request_id = dispatch.request_id;
let outcome = match dispatch.wait_policy {
StartAgentWaitPolicy::Completion => dispatch.receiver.recv().await.ok(),
StartAgentWaitPolicy::Completion | StartAgentWaitPolicy::LocalStartup => {
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))
app/src/ai/blocklist/action_model/execute/run_agents_tests.rs
+3 -2
View File
@@ -458,12 +458,12 @@ fn recovered_run_agents_reattaches_existing_child_and_launches_only_missing_chil
assert_eq!(agents.len(), 2);
assert!(matches!(
&agents[0].kind,
RunAgentsAgentOutcomeKind::Completed { agent_id, .. }
RunAgentsAgentOutcomeKind::Launched { agent_id }
if agent_id == &existing_child_id.to_string()
));
assert!(matches!(
&agents[1].kind,
RunAgentsAgentOutcomeKind::Completed { agent_id, .. }
RunAgentsAgentOutcomeKind::Launched { agent_id }
if agent_id == &missing_child_id.to_string()
));
});
@@ -698,6 +698,7 @@ fn initialize_run_agents_test(app: &mut App, mode: ExecutionMode) -> RunAgentsTe
let global_resource_handles = GlobalResourceHandles::mock(app);
app.add_singleton_model(|_| GlobalResourceHandlesProvider::new(global_resource_handles));
let history = app.add_singleton_model(|_| BlocklistAIHistoryModel::new(vec![], &[]));
app.add_singleton_model(OrchestrationEventService::new);
app.add_singleton_model(|_| CLIAgentSessionsModel::new());
app.add_singleton_model(|_| ActiveAgentViewsModel::new());
app.add_singleton_model(AgentNotificationsModel::new);
app/src/ai/blocklist/action_model/execute/start_agent.rs
+34
View File
@@ -18,6 +18,7 @@ use crate::ai::agent::{
StartAgentExecutionMode, StartAgentResult,
};
use crate::ai::blocklist::orchestration_event_streamer::OrchestrationEventStreamer;
use crate::ai::blocklist::orchestration_events::OrchestrationEventService;
use crate::ai::blocklist::{BlocklistAIHistoryEvent, BlocklistAIHistoryModel};
use crate::ai::llms::{LLMId, LLMPreferences};
use crate::ai::local_harness_setup::local_harness_product_disabled_message;
@@ -42,6 +43,8 @@ pub enum StartAgentOutcome {
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StartAgentWaitPolicy {
Startup,
/// A local child has been linked; its persistent group owns completion.
LocalStartup,
Completion,
}
@@ -227,6 +230,20 @@ impl StartAgentExecutor {
}
};
if let Some(pending) = self.pending.get(&request_id) {
if matches!(pending.wait_policy, StartAgentWaitPolicy::LocalStartup) {
let parent = pending.parent_conversation_id;
let group_id = pending.action_id.to_string();
OrchestrationEventService::handle(ctx).update(ctx, |service, ctx| {
service.register_local_group_child(
parent,
&group_id,
child_conversation_id,
ctx,
);
});
}
}
if let Some(event) = child_link_event {
ctx.emit(event);
}
@@ -373,6 +390,14 @@ impl StartAgentExecutor {
| ConversationStatus::Blocked { .. }
| ConversationStatus::WaitingForEvents => {}
},
Some(StartAgentWaitPolicy::LocalStartup) => {
// Membership is durable before launch acknowledgement reaches the parent.
if let Some(pending) = self.pending.remove(&request_id) {
let _ = pending.sender.try_send(StartAgentOutcome::Started {
agent_id: child_conversation_id.to_string(),
});
}
}
Some(StartAgentWaitPolicy::Startup) => {
if let Some(error_msg) = start_agent_startup_error_message_for_status(
conversation.status(),
@@ -683,6 +708,15 @@ impl StartAgentExecutor {
})
}
pub(super) fn yield_after_local_start(&mut self, dispatch: &mut StartAgentDispatch) {
if matches!(dispatch.wait_policy, StartAgentWaitPolicy::Completion) {
dispatch.wait_policy = StartAgentWaitPolicy::LocalStartup;
if let Some(pending) = self.pending.get_mut(&dispatch.request_id) {
pending.wait_policy = StartAgentWaitPolicy::LocalStartup;
}
}
}
/// Dispatch a pre-validated StartAgent request. Returns a receiver
/// for the resulting [`StartAgentOutcome`].
#[allow(clippy::too_many_arguments)]
app/src/ai/blocklist/action_model/execute/start_agent_tests.rs
+5 -5
View File
@@ -559,7 +559,7 @@ fn execute_resolves_error_when_request_linkage_happens_after_child_already_faile
assert!(matches!(
result,
AIAgentActionResultType::StartAgent(StartAgentResult::Error { error, version })
if error == "'codex' CLI not found on your machine."
if error.contains("'codex' CLI not found on your machine.") && error.contains("send_message_to_agent")
&& version == StartAgentVersion::V1
));
@@ -906,7 +906,7 @@ fn direct_provider_nonterminal_states_remain_pending_until_cancelled() {
assert!(matches!(
state.dispatch.receiver.try_recv(),
Ok(StartAgentOutcome::Error(error))
if error == "Child agent was cancelled by the user."
if error.contains("Child agent was cancelled by the user.") && error.contains("send_message_to_agent")
));
state.executor.read(&app, |executor, _| {
assert!(executor.pending.is_empty());
@@ -933,7 +933,7 @@ fn direct_provider_error_preserves_child_for_inspection() {
assert!(matches!(
state.dispatch.receiver.try_recv(),
Ok(StartAgentOutcome::Error(error)) if error == "Child execution failed"
Ok(StartAgentOutcome::Error(error)) if error.contains("Child execution failed") && error.contains("send_message_to_agent")
));
state.captured_cleanup.read(&app, |captured, _| {
assert!(captured.0.is_empty());
@@ -1369,7 +1369,7 @@ fn execute_returns_detailed_error_when_child_startup_fails_before_initialization
assert!(matches!(
result,
AIAgentActionResultType::StartAgent(StartAgentResult::Error { error, version })
if error == "Failed to resolve child agent skills: review-comments"
if error.contains("Failed to resolve child agent skills: review-comments") && error.contains("send_message_to_agent")
&& version == StartAgentVersion::V1
));
});
@@ -1739,7 +1739,7 @@ fn parallel_pendings_each_resolve_independently_via_recorded_child_id() {
assert!(matches!(
result_b,
AIAgentActionResultType::StartAgent(StartAgentResult::Error { error, .. })
if error == "Agent B init failed"
if error.contains("Agent B init failed") && error.contains("send_message_to_agent")
));
executor.read(&app, |executor, _| {
app/src/ai/blocklist/controller.rs
+23 -6
View File
@@ -3949,8 +3949,8 @@ impl BlocklistAIController {
}
if has_piggybacked_events && result.is_ok() {
OrchestrationEventService::handle(ctx).update(ctx, |svc, _| {
svc.confirm_local_messages(conversation_id);
OrchestrationEventService::handle(ctx).update(ctx, |svc, ctx| {
svc.confirm_local_delivery(conversation_id, ctx);
});
}
self.pending_passive_follow_ups.remove(&conversation_id);
@@ -4371,8 +4371,8 @@ impl BlocklistAIController {
svc.requeue_awaiting_events(conversation_id, ctx);
});
} else {
OrchestrationEventService::handle(ctx).update(ctx, |svc, _| {
svc.confirm_local_messages(conversation_id);
OrchestrationEventService::handle(ctx).update(ctx, |svc, ctx| {
svc.confirm_local_delivery(conversation_id, ctx);
});
}
}
@@ -7321,6 +7321,13 @@ impl BlocklistAIController {
slot: &mut ActiveProviderRunSlot,
run: &mut ActiveProviderRun,
) -> Result<ProviderBoundaryDisposition, String> {
// A launch yield can complete the run as its tool batch commits. Once the
// action projection is committed, drive terminal cleanup rather than parking it.
if run.coordinator.run().is_terminal() && slot.committed_provider_batch.is_none() {
return Ok(ProviderBoundaryDisposition::Advance {
completed_block_id: None,
});
}
let state = run.coordinator.run().state();
let phase = provider_boundary_phase(state);
let intent = provider_boundary_intent(
@@ -8192,16 +8199,23 @@ impl BlocklistAIController {
ctx: &mut ModelContext<Self>,
) {
let history_model = BlocklistAIHistoryModel::handle(ctx);
let status = if OrchestrationEventService::as_ref(ctx)
.has_waiting_local_groups(conversation_id, ctx)
{
ConversationStatus::WaitingForEvents
} else {
ConversationStatus::Success
};
let should_finalize = history_model
.as_ref(ctx)
.conversation_status(&conversation_id)
.is_some_and(|status| status != &ConversationStatus::Success);
.is_some_and(|current| current != &status);
if should_finalize {
history_model.update(ctx, |history_model, ctx| {
history_model.update_conversation_status(
self.terminal_surface_id,
conversation_id,
ConversationStatus::Success,
status,
ctx,
);
});
@@ -8397,6 +8411,9 @@ impl BlocklistAIController {
});
self.maybe_refresh_ai_overages(ctx);
self.start_next_queued_provider_run(conversation_id, ctx);
OrchestrationEventService::handle(ctx).update(ctx, |service, ctx| {
service.reconcile_local_groups_for_child(conversation_id, ctx);
});
// Status notifications may arrive before the stream releases its slot.
// Retry queued child messages now that terminal cleanup is complete.
self.handle_pending_events_ready(conversation_id, ctx);
app/src/ai/blocklist/history_model.rs
+16
View File
@@ -782,6 +782,19 @@ impl BlocklistAIHistoryModel {
title,
});
}
pub(crate) fn set_local_agent_groups(
&mut self,
conversation_id: AIConversationId,
groups: Vec<persistence::model::LocalAgentGroup>,
ctx: &mut ModelContext<Self>,
) {
let Some(conversation) = self.conversations_by_id.get_mut(&conversation_id) else {
return;
};
conversation.set_local_agent_groups(groups);
self.persist_conversation_state(conversation_id, ctx);
}
pub fn mark_conversation_as_remote_child(
&mut self,
conversation_id: AIConversationId,
@@ -1691,6 +1704,7 @@ impl BlocklistAIHistoryModel {
let conversation_data = AgentConversationData {
agent_backend: source_conversation.agent_backend().for_fork(),
active_provider_run_json: None,
local_agent_groups: Vec::new(),
server_conversation_token: None,
conversation_usage_metadata: Some(source_conversation.usage_metadata()),
reverted_action_ids,
@@ -1856,6 +1870,7 @@ impl BlocklistAIHistoryModel {
let conversation_data = AgentConversationData {
agent_backend: conversation.agent_backend().for_fork(),
active_provider_run_json: None,
local_agent_groups: Vec::new(),
server_conversation_token: None,
conversation_usage_metadata: None,
reverted_action_ids,
@@ -2866,6 +2881,7 @@ fn merged_remote_child_placeholder_conversation_data(
// Active process-local provider runs cannot be merged from a cloud transcript.
active_provider_run_json: None,
local_agent_groups: Vec::new(),
// Cloud authoritative.
server_conversation_token: cloud_conversation
app/src/ai/blocklist/local_agent_groups.rs
+279
View File
@@ -0,0 +1,279 @@
//! Local delegation membership survives parent turns; only complete groups wake the parent.
use galaxyui::AppContext;
use persistence::model::LocalAgentGroup;
use super::*;
fn group_event_id(group: &LocalAgentGroup) -> String {
format!("local-group/{}/{}", group.id, group.revision)
}
fn terminal_child(status: &ConversationStatus) -> bool {
matches!(
status,
ConversationStatus::Success | ConversationStatus::Error | ConversationStatus::Cancelled
)
}
impl OrchestrationEventService {
pub(crate) fn has_waiting_local_groups(
&self,
parent: AIConversationId,
ctx: &AppContext,
) -> bool {
BlocklistAIHistoryModel::as_ref(ctx)
.conversation(&parent)
.is_some_and(|conversation| {
conversation
.local_agent_groups()
.iter()
.any(|group| !group.notified && !group.children.is_empty())
})
}
pub(crate) fn begin_local_group(
&mut self,
parent: AIConversationId,
id: &str,
ctx: &mut ModelContext<Self>,
) {
self.update_local_group(parent, id, ctx, |_| {});
}
pub(crate) fn register_local_group_child(
&mut self,
parent: AIConversationId,
id: &str,
child: AIConversationId,
ctx: &mut ModelContext<Self>,
) {
self.update_local_group(parent, id, ctx, |group| {
let child = child.to_string();
if !group.children.contains(&child) {
group.children.push(child);
}
});
}
pub(crate) fn seal_local_group(
&mut self,
parent: AIConversationId,
id: &str,
errors: Vec<String>,
ctx: &mut ModelContext<Self>,
) {
self.update_local_group(parent, id, ctx, |group| {
group.sealed = true;
group.launch_errors.extend(errors);
});
self.reconcile_local_groups(parent, ctx);
}
fn update_local_group(
&mut self,
parent: AIConversationId,
id: &str,
ctx: &mut ModelContext<Self>,
edit: impl FnOnce(&mut LocalAgentGroup),
) {
let Some(conversation) = BlocklistAIHistoryModel::as_ref(ctx).conversation(&parent) else {
return;
};
let mut groups = conversation.local_agent_groups().to_vec();
if !groups.iter().any(|group| group.id == id) {
groups.push(LocalAgentGroup {
id: id.to_string(),
..Default::default()
});
}
edit(
groups
.iter_mut()
.find(|group| group.id == id)
.expect("group was inserted"),
);
BlocklistAIHistoryModel::handle(ctx).update(ctx, |history, ctx| {
history.set_local_agent_groups(parent, groups, ctx)
});
}
pub(super) fn confirm_group_messages(
&mut self,
parent: AIConversationId,
event_ids: &[String],
ctx: &mut ModelContext<Self>,
) {
let Some(conversation) = BlocklistAIHistoryModel::as_ref(ctx).conversation(&parent) else {
return;
};
let mut groups = conversation.local_agent_groups().to_vec();
let mut changed = false;
for group in &mut groups {
if event_ids.contains(&group_event_id(group)) && !group.notified {
group.notified = true;
changed = true;
}
}
if changed {
BlocklistAIHistoryModel::handle(ctx).update(ctx, |history, ctx| {
history.set_local_agent_groups(parent, groups, ctx)
});
}
}
/// Re-evaluate current statuses, rather than counting completion callbacks. A restarted
/// child becomes pending again and invalidates any undelivered earlier group result.
pub(crate) fn reconcile_local_groups_for_child(
&mut self,
child: AIConversationId,
ctx: &mut ModelContext<Self>,
) -> bool {
let parents = BlocklistAIHistoryModel::as_ref(ctx)
.all_live_conversations()
.iter()
.filter_map(|(_, conversation)| {
conversation
.local_agent_groups()
.iter()
.any(|group| group.children.contains(&child.to_string()))
.then_some(conversation.id())
})
.collect::<Vec<_>>();
let grouped = !parents.is_empty();
for parent in parents {
self.reconcile_local_groups(parent, ctx);
}
grouped
}
pub(super) fn reconcile_local_groups(
&mut self,
parent: AIConversationId,
ctx: &mut ModelContext<Self>,
) {
let history = BlocklistAIHistoryModel::as_ref(ctx);
let Some(conversation) = history.conversation(&parent) else {
return;
};
let mut groups = conversation.local_agent_groups().to_vec();
let original = groups.clone();
let mut ready = Vec::new();
let mut stale = Vec::new();
for group in &mut groups {
if group.children.is_empty() {
continue;
}
let children = group
.children
.iter()
.map(|id| {
AIConversationId::try_from(id.clone())
.ok()
.and_then(|id| history.conversation(&id))
.filter(|child| {
child.parent_conversation_id() == Some(parent)
&& !child.is_remote_child()
})
})
.collect::<Vec<_>>();
let finished = children.iter().all(|child| {
child.is_none_or(|child| {
terminal_child(child.status()) && child.active_provider_run_json().is_none()
})
});
let event_id = group_event_id(group);
let queued = self
.pending_events
.get(&parent)
.into_iter()
.chain(self.awaiting_server_echo_events.get(&parent))
.flatten()
.any(|event| event.event_id == event_id);
if !finished {
if group.notified || queued {
stale.push(event_id);
group.revision = group.revision.saturating_add(1);
group.notified = false;
}
continue;
}
if !group.sealed || group.notified || queued {
continue;
}
let mut result = format!("All children in delegation group {} have finished. Process their results together now. Failed or cancelled children are explicit blockers, not successful results. Use a corrective follow-up to the same child for recoverable errors; do not override cancellation.\n", group.id);
for (id, child) in group.children.iter().zip(children) {
let Some(child) = child else {
result.push_str(&format!("\nChild {id}: removed or unavailable.\n"));
continue;
};
result.push_str(&format!(
"\n{} ({id}): {:?}\n",
child.agent_name().unwrap_or("Child"),
child.status()
));
if matches!(child.status(), ConversationStatus::Success) {
let output = child
.all_exchanges()
.into_iter()
.rev()
.filter_map(|exchange| exchange.output_status.output())
.map(|output| output.get().format_for_copy(None))
.find(|output| !output.is_empty())
.unwrap_or_default();
// Bound the whole group, including large fan-outs, to roughly 32 KB.
let limit = 32_000 / group.children.len().max(1);
let mut end = output.len().min(limit);
while !output.is_char_boundary(end) {
end -= 1;
}
result.push_str(&output[..end]);
if end < output.len() {
result.push_str("\n[Result excerpt truncated; inspect the child conversation for more.]");
}
} else if let Some(error) = child.status_error_message() {
result.push_str(&error);
}
}
for error in &group.launch_errors {
result.push_str(&format!("\nLaunch failure: {error}\n"));
}
if result.len() > 48_000 {
let mut end = 48_000;
while !result.is_char_boundary(end) {
end -= 1;
}
result.truncate(end);
result.push_str(
"\n[Group result truncated; inspect the child conversations for full details.]",
);
}
ready.push(PendingEvent {
event_id: event_id.clone(),
source_agent_id: parent.to_string(),
attempt_count: 0,
detail: PendingEventDetail::LocalMessage {
message_id: event_id,
subject: "Child agent group completed".to_string(),
message_body: result,
},
});
}
for events in [
self.pending_events.get_mut(&parent),
self.awaiting_server_echo_events.get_mut(&parent),
]
.into_iter()
.flatten()
{
events.retain(|event| !stale.contains(&event.event_id));
}
if groups != original {
BlocklistAIHistoryModel::handle(ctx).update(ctx, |history, ctx| {
history.set_local_agent_groups(parent, groups, ctx)
});
}
if !ready.is_empty() {
self.enqueue_event_batch(parent, ready, ctx);
}
}
}
app/src/ai/blocklist/orchestration_events.rs
+28 -1
View File
@@ -1,5 +1,8 @@
#![allow(dead_code)]
#[path = "local_agent_groups.rs"]
mod local_agent_groups;
use std::collections::{HashMap, HashSet};
use uuid::Uuid;
@@ -146,6 +149,9 @@ impl OrchestrationEventService {
for conversation_id in conversation_ids {
self.sync_conversation_status(*conversation_id, ctx);
}
for conversation_id in conversation_ids {
self.reconcile_local_groups(*conversation_id, ctx);
}
}
BlocklistAIHistoryEvent::RemoveConversation {
conversation_id, ..
@@ -157,6 +163,7 @@ impl OrchestrationEventService {
self.awaiting_server_echo_events.remove(conversation_id);
self.conversation_statuses.remove(conversation_id);
self.restarted_children.remove(conversation_id);
self.reconcile_local_groups_for_child(*conversation_id, ctx);
}
_ => {}
}
@@ -196,7 +203,10 @@ impl OrchestrationEventService {
let previous = self
.conversation_statuses
.insert(conversation_id, current_status.clone());
if !is_restored && previous.as_ref() != Some(&current_status) {
if !is_restored
&& previous.as_ref() != Some(&current_status)
&& !self.reconcile_local_groups_for_child(conversation_id, ctx)
{
self.report_restarted_child(conversation_id, previous.as_ref(), &current_status, ctx);
}
let has_pending = self
@@ -231,6 +241,23 @@ impl OrchestrationEventService {
}
/// Local providers commit the input directly and do not emit a hosted message echo.
pub(crate) fn confirm_local_delivery(
&mut self,
parent: AIConversationId,
ctx: &mut ModelContext<Self>,
) {
let ids = self
.awaiting_server_echo_events
.get(&parent)
.into_iter()
.flatten()
.filter(|event| matches!(event.detail, PendingEventDetail::LocalMessage { .. }))
.map(|event| event.event_id.clone())
.collect::<Vec<_>>();
self.confirm_group_messages(parent, &ids, ctx);
self.confirm_local_messages(parent);
}
pub fn confirm_local_messages(&mut self, conversation_id: AIConversationId) {
if let Some(events) = self.awaiting_server_echo_events.get_mut(&conversation_id) {
events.retain(|event| !matches!(event.detail, PendingEventDetail::LocalMessage { .. }));
app/src/ai/blocklist/orchestration_events_tests.rs
+62
View File
@@ -391,3 +391,65 @@ fn local_follow_up_is_delivered_once_and_acknowledged_without_server_echo() {
assert_eq!(awaiting[0].event_id, "hosted-1");
assert!(!service.has_pending_local_message(conversation_id));
}
#[test]
fn local_group_waits_for_every_child_and_invalidates_results_after_restart() {
galaxyui::App::test((), |mut app| async move {
crate::test_util::settings::initialize_history_persistence_for_tests(&mut app);
let surface = galaxyui::EntityId::new();
let history = app.add_singleton_model(|_| BlocklistAIHistoryModel::new_for_test());
let service = app.add_singleton_model(OrchestrationEventService::new);
let (parent, first, second) = history.update(&mut app, |history, ctx| {
let parent = history.start_new_conversation(surface, false, false, false, ctx);
let first =
history.start_new_child_conversation(surface, "one".to_string(), parent, None, ctx);
let second =
history.start_new_child_conversation(surface, "two".to_string(), parent, None, ctx);
(parent, first, second)
});
service.update(&mut app, |service, ctx| {
service.begin_local_group(parent, "group-1", ctx);
service.register_local_group_child(parent, "group-1", first, ctx);
service.register_local_group_child(parent, "group-1", second, ctx);
service.seal_local_group(parent, "group-1", Vec::new(), ctx);
});
history.update(&mut app, |history, ctx| {
history.update_conversation_status(surface, first, ConversationStatus::Success, ctx)
});
service.read(&app, |service, _| {
assert!(!service.has_pending_events(parent))
});
history.update(&mut app, |history, ctx| {
history.update_conversation_status(surface, second, ConversationStatus::Error, ctx)
});
let old_event = service.read(&app, |service, _| {
assert_eq!(service.pending_events[&parent].len(), 1);
service.pending_events[&parent][0].event_id.clone()
});
// A user restarts the failed child before the parent has consumed the result.
history.update(&mut app, |history, ctx| {
history.update_conversation_status(surface, second, ConversationStatus::InProgress, ctx)
});
service.read(&app, |service, _| {
assert!(!service.has_pending_events(parent))
});
history.update(&mut app, |history, ctx| {
history.update_conversation_status(surface, second, ConversationStatus::Success, ctx)
});
service.update(&mut app, |service, ctx| {
assert_eq!(service.pending_events[&parent].len(), 1);
assert_ne!(service.pending_events[&parent][0].event_id, old_event);
service.drain_and_convert_events(parent);
service.confirm_local_delivery(parent, ctx);
service.reconcile_local_groups(parent, ctx);
assert!(!service.has_pending_events(parent));
});
history.read(&app, |history, _| {
let groups = history.conversation(&parent).unwrap().local_agent_groups();
assert!(groups[0].notified);
let restored: persistence::model::LocalAgentGroup =
serde_json::from_str(&serde_json::to_string(&groups[0]).unwrap()).unwrap();
assert_eq!(&restored, &groups[0]);
});
});
}