From 72cd38b54efe13a51df151d44603beb0308c2495 Mon Sep 17 00:00:00 2001 From: Ryan Ward Date: Mon, 14 Sep 2026 22:32:18 -0500 Subject: [PATCH] Yield orchestrators for child groups and correct subscription context catalogs --- AGENTS.md | 4 +- app/src/ai/agent/api/convert_conversation.rs | 2 + app/src/ai/agent/conversation.rs | 17 ++ .../action_model/execute/run_agents.rs | 43 ++- .../action_model/execute/run_agents_tests.rs | 5 +- .../action_model/execute/start_agent.rs | 34 +++ .../action_model/execute/start_agent_tests.rs | 10 +- app/src/ai/blocklist/controller.rs | 29 +- app/src/ai/blocklist/history_model.rs | 16 + app/src/ai/blocklist/local_agent_groups.rs | 279 ++++++++++++++++++ app/src/ai/blocklist/orchestration_events.rs | 29 +- .../blocklist/orchestration_events_tests.rs | 62 ++++ app/src/ai/llms.rs | 16 +- app/src/ai/llms_tests.rs | 8 +- app/src/ai/provider/request_translator.rs | 2 +- app/src/ai/runtime/rig_request.rs | 2 +- app/src/ai/runtime/rig_tool.rs | 27 +- app/src/pane_group/mod_tests.rs | 1 + app/src/settings_view/ai_page.rs | 25 +- app/src/terminal/view/load_ai_conversation.rs | 1 + crates/ai/src/agent/action_result/mod.rs | 2 +- crates/galaxy_agent_core/src/provider_run.rs | 39 ++- .../src/provider_run_tests.rs | 69 ++++- crates/persistence/src/model.rs | 14 + crates/persistence/src/model_tests.rs | 6 + vendor/block/src/lib.rs | 59 +--- vendor/block/src/lib_tests.rs | 55 ++++ 27 files changed, 741 insertions(+), 115 deletions(-) create mode 100644 app/src/ai/blocklist/local_agent_groups.rs create mode 100644 vendor/block/src/lib_tests.rs diff --git a/AGENTS.md b/AGENTS.md index 5ed8005f..043ec000 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -166,8 +166,8 @@ Key invariants: - Direct-provider completed-command assessments are hidden, tool-free root-task turns whose output and hidden input survive CLI-task deactivation and restoration - ACP remains a separate session-owned runtime; direct-provider cleanup must not move ACP lifecycle into `ProviderRun` - Orchestrated child conversations are leaf workers: nested `RunAgents` and legacy `StartAgent` calls are rejected before autonomy or permission bypasses, and child requests do not advertise delegation tools -- Direct-provider `RunAgents` remains pending until every local child reaches `Success`, `Error`, or `Cancelled`, or is removed/deleted; recoverable `Blocked`, `TransientError`, and `WaitingForEvents` states remain pending, and the hosted 30-second startup timeout must not apply to these completion waits -- `StartAgentWaitPolicy` is selected from child execution mode, not parent `run_id`: local children wait for completion and only remote/hosted children use startup acknowledgement +- Direct-provider `RunAgents` acknowledges linked local children and yields the parent generation after committing its launch results. Other tools in that launch batch are explicitly not executed. Persistent `LocalAgentGroup` membership survives parent follow-ups; only when every child reaches `Success`, `Error`, or `Cancelled` (with provider cleanup complete), or is removed/deleted, does one combined completion message wake the parent. `Blocked`, `TransientError`, and `WaitingForEvents` children remain pending. Restarts invalidate undelivered group results, and group delivery is confirmed only after the parent input is accepted. +- `StartAgentWaitPolicy` follows child execution ownership, not parent `run_id`: grouped local children use `LocalStartup`, legacy local completion waits retain `Completion`, and hosted children use `Startup` with the hosted timeout. Direct-provider `start_agent` is normalized to a one-child `RunAgents` group. Cancelling or replacing a parent turn must not cancel independently running children or erase their group membership. - Hosted `RunAgents` startup timeouts detach the exact `StartAgentRequestId`; late launch callbacks must not register the child after the timeout result, while children linked before cancellation remain independently running ### Platform Setup diff --git a/app/src/ai/agent/api/convert_conversation.rs b/app/src/ai/agent/api/convert_conversation.rs index 0798134c..afef7244 100644 --- a/app/src/ai/agent/api/convert_conversation.rs +++ b/app/src/ai/agent/api/convert_conversation.rs @@ -74,6 +74,7 @@ pub fn convert_conversation_data_to_ai_conversation( RestorationMode::Fork => AgentConversationData { agent_backend: Default::default(), active_provider_run_json: None, + local_agent_groups: Vec::new(), server_conversation_token: None, conversation_usage_metadata: usage_metadata, reverted_action_ids: None, @@ -98,6 +99,7 @@ pub fn convert_conversation_data_to_ai_conversation( RestorationMode::Continue => AgentConversationData { agent_backend: Default::default(), active_provider_run_json: None, + local_agent_groups: Vec::new(), server_conversation_token: Some( metadata.server_conversation_token.as_str().to_string(), ), diff --git a/app/src/ai/agent/conversation.rs b/app/src/ai/agent/conversation.rs index ca8b3ba6..0639d66c 100644 --- a/app/src/ai/agent/conversation.rs +++ b/app/src/ai/agent/conversation.rs @@ -17,6 +17,7 @@ use galaxy_core::ui::theme::GalaxyTheme; use galaxyui::color::ColorU; use galaxyui::{AppContext, EntityId, ModelContext, SingletonEntity}; use itertools::Itertools as _; +use persistence::model::LocalAgentGroup; use serde::{Deserialize, Serialize}; use uuid::Uuid; use vec1::{Size0Error, Vec1}; @@ -232,6 +233,7 @@ pub struct AIConversation { /// Opaque, versioned snapshot of the active direct-provider run. active_provider_run_json: Option, + local_agent_groups: Vec, /// The server-generated unique "token" for this conversation. /// @@ -386,6 +388,7 @@ impl AIConversation { conversation_usage_metadata: ConversationUsageMetadata::default(), agent_backend, active_provider_run_json: None, + local_agent_groups: Vec::new(), server_conversation_token: None, task_id: None, forked_from_server_conversation_token: None, @@ -544,6 +547,10 @@ impl AIConversation { (task_store, todo_lists, status) }; + let local_agent_groups = conversation_data + .as_ref() + .map(|data| data.local_agent_groups.clone()) + .unwrap_or_default(); let ( agent_backend, active_provider_run_json, @@ -674,6 +681,7 @@ impl AIConversation { conversation_usage_metadata, agent_backend, active_provider_run_json, + local_agent_groups, server_conversation_token, task_id: run_id.as_deref().and_then(|id| id.parse().ok()), forked_from_server_conversation_token, @@ -717,6 +725,14 @@ impl AIConversation { &self.agent_backend } + pub(crate) fn local_agent_groups(&self) -> &[LocalAgentGroup] { + &self.local_agent_groups + } + + pub(crate) fn set_local_agent_groups(&mut self, groups: Vec) { + self.local_agent_groups = groups; + } + pub(crate) fn active_provider_run_json(&self) -> Option<&str> { self.active_provider_run_json.as_deref() } @@ -4033,6 +4049,7 @@ impl AIConversation { conversation_data: AgentConversationData { agent_backend: self.agent_backend.clone(), active_provider_run_json: self.active_provider_run_json.clone(), + local_agent_groups: self.local_agent_groups.clone(), server_conversation_token: self .server_conversation_token .clone() diff --git a/app/src/ai/blocklist/action_model/execute/run_agents.rs b/app/src/ai/blocklist/action_model/execute/run_agents.rs index b27eeed7..16d8286d 100644 --- a/app/src/ai/blocklist/action_model/execute/run_agents.rs +++ b/app/src/ai/blocklist/action_model/execute/run_agents.rs @@ -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 = 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)) diff --git a/app/src/ai/blocklist/action_model/execute/run_agents_tests.rs b/app/src/ai/blocklist/action_model/execute/run_agents_tests.rs index aabc1d7c..a5d03071 100644 --- a/app/src/ai/blocklist/action_model/execute/run_agents_tests.rs +++ b/app/src/ai/blocklist/action_model/execute/run_agents_tests.rs @@ -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); 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 6cae7bb8..39e8698c 100644 --- a/app/src/ai/blocklist/action_model/execute/start_agent.rs +++ b/app/src/ai/blocklist/action_model/execute/start_agent.rs @@ -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)] diff --git a/app/src/ai/blocklist/action_model/execute/start_agent_tests.rs b/app/src/ai/blocklist/action_model/execute/start_agent_tests.rs index 38676a86..af16bad4 100644 --- a/app/src/ai/blocklist/action_model/execute/start_agent_tests.rs +++ b/app/src/ai/blocklist/action_model/execute/start_agent_tests.rs @@ -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, _| { diff --git a/app/src/ai/blocklist/controller.rs b/app/src/ai/blocklist/controller.rs index d347c435..c2dbf8e8 100644 --- a/app/src/ai/blocklist/controller.rs +++ b/app/src/ai/blocklist/controller.rs @@ -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 { + // 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, ) { 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); diff --git a/app/src/ai/blocklist/history_model.rs b/app/src/ai/blocklist/history_model.rs index 644a91bb..8b3be9cb 100644 --- a/app/src/ai/blocklist/history_model.rs +++ b/app/src/ai/blocklist/history_model.rs @@ -782,6 +782,19 @@ impl BlocklistAIHistoryModel { title, }); } + pub(crate) fn set_local_agent_groups( + &mut self, + conversation_id: AIConversationId, + groups: Vec, + ctx: &mut ModelContext, + ) { + 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 diff --git a/app/src/ai/blocklist/local_agent_groups.rs b/app/src/ai/blocklist/local_agent_groups.rs new file mode 100644 index 00000000..9dc7898a --- /dev/null +++ b/app/src/ai/blocklist/local_agent_groups.rs @@ -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.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.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, + ctx: &mut ModelContext, + ) { + 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, + 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, + ) { + 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, + ) -> 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::>(); + 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, + ) { + 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::>(); + 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); + } + } +} diff --git a/app/src/ai/blocklist/orchestration_events.rs b/app/src/ai/blocklist/orchestration_events.rs index 71cee50f..387d602b 100644 --- a/app/src/ai/blocklist/orchestration_events.rs +++ b/app/src/ai/blocklist/orchestration_events.rs @@ -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(¤t_status) { + if !is_restored + && previous.as_ref() != Some(¤t_status) + && !self.reconcile_local_groups_for_child(conversation_id, ctx) + { self.report_restarted_child(conversation_id, previous.as_ref(), ¤t_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, + ) { + 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::>(); + 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 { .. })); diff --git a/app/src/ai/blocklist/orchestration_events_tests.rs b/app/src/ai/blocklist/orchestration_events_tests.rs index 60c8cf62..33afbd24 100644 --- a/app/src/ai/blocklist/orchestration_events_tests.rs +++ b/app/src/ai/blocklist/orchestration_events_tests.rs @@ -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]); + }); + }); +} diff --git a/app/src/ai/llms.rs b/app/src/ai/llms.rs index 63ffcc2e..606c3d16 100644 --- a/app/src/ai/llms.rs +++ b/app/src/ai/llms.rs @@ -2688,8 +2688,10 @@ fn get_new_agent_mode_choices( .collect() } -#[cfg(not(target_family = "wasm"))] -fn openai_model_context_size(model: &OpenAIModelConfig, provider_kind: OpenAIProviderKind) -> u32 { +pub(crate) fn openai_model_context_size( + model: &OpenAIModelConfig, + provider_kind: OpenAIProviderKind, +) -> u32 { let input_limit = model .max_input_tokens .unwrap_or(model.context_size) @@ -2972,9 +2974,6 @@ fn chatgpt_models_from_codex_response(body: &serde_json::Value) -> Vec Vec Vec { }, ToolDefinition { name: "run_agents".to_string(), - description: "Start one or more child agents that share run-wide configuration. Use independent, uniquely named tasks and do not launch duplicate agents for follow-up work. Omitted run-wide fields use the embedded local child runtime and the configured terminal model. After launch, call wait_for_events when you need child-agent results instead of repeating their work yourself.".to_string(), + description: "Start one or more child agents that share run-wide configuration. Use independent, uniquely named tasks and do not launch duplicate agents for follow-up work. Omitted run-wide fields use the embedded local child runtime and the configured terminal model. Local launch automatically yields this turn. Do not batch other tools with run_agents. The user can keep talking to you while children work independently. Each group sends one combined update after all children finish; wait for that update before processing their results. User follow-ups do not cancel the group, and must not cause duplicate children or repeated delegated work.".to_string(), input_schema: serde_json::json!({ "type": "object", "properties": { diff --git a/app/src/ai/runtime/rig_request.rs b/app/src/ai/runtime/rig_request.rs index f49435b7..d56623bd 100644 --- a/app/src/ai/runtime/rig_request.rs +++ b/app/src/ai/runtime/rig_request.rs @@ -116,7 +116,7 @@ pub(crate) fn add_orchestration_model_options( required.push(serde_json::Value::String("model_id".to_string())); } - tool.description = "Start one or more child agents. Assign each child the best-fit, cost-effective model from its required model_id choices, prioritizing capability and likelihood of success over price. Use independent, uniquely named tasks and do not launch duplicate agents for follow-up work. After launch, call wait_for_events when you need child-agent results instead of repeating their work yourself.".to_string(); + tool.description = "Start one or more child agents. Assign each child the best-fit, cost-effective model from its required model_id choices, prioritizing capability and likelihood of success over price. Use independent, uniquely named tasks and do not launch duplicate agents for follow-up work. Local launch automatically yields this turn. Do not batch other tools with run_agents. The user can keep talking to you while children work independently. Each group sends one combined update after all children finish; wait for that update before processing their results. User follow-ups do not cancel the group, and must not cause duplicate children or repeated delegated work.".to_string(); } #[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)] diff --git a/app/src/ai/runtime/rig_tool.rs b/app/src/ai/runtime/rig_tool.rs index 94bfcb31..65137ce1 100644 --- a/app/src/ai/runtime/rig_tool.rs +++ b/app/src/ai/runtime/rig_tool.rs @@ -13,8 +13,7 @@ use crate::ai::agent::{ AskUserQuestionOption, AskUserQuestionType, CreateDocumentsRequest, DocumentDiff, DocumentToCreate, EditDocumentsRequest, FileEdit, FileLocations, ReadDocumentsRequest, ReadFilesRequest, ReadSkillRequest, RunAgentsAgentRunConfig, RunAgentsExecutionMode, - RunAgentsRequest, SearchCodebaseRequest, ShellCommandDelay, StartAgentExecutionMode, - StartAgentVersion, + RunAgentsRequest, SearchCodebaseRequest, ShellCommandDelay, }; use crate::ai::document::ai_document_model::AIDocumentId; @@ -188,13 +187,23 @@ pub(crate) fn action_from_tool_call( plan_id: optional_string(input, "plan_id")?.unwrap_or_default(), harness_auth_secret_name: None, }), - "start_agent" => AIAgentActionType::StartAgent { - version: StartAgentVersion::V1, - name: required_nonempty_string(input, "name")?, - prompt: required_nonempty_string(input, "prompt")?, - execution_mode: StartAgentExecutionMode::local_with_defaults(), - lifecycle_subscription: None, - }, + // Direct-provider single-child delegation uses the same durable group lifecycle. + "start_agent" => AIAgentActionType::RunAgents(RunAgentsRequest { + summary: required_nonempty_string(input, "name")?, + base_prompt: String::new(), + skills: Vec::new(), + model_id: String::new(), + harness_type: String::new(), + execution_mode: RunAgentsExecutionMode::Local, + agent_run_configs: vec![RunAgentsAgentRunConfig { + name: required_nonempty_string(input, "name")?, + prompt: required_nonempty_string(input, "prompt")?, + title: String::new(), + model_id: String::new(), + }], + plan_id: String::new(), + harness_auth_secret_name: None, + }), "send_message_to_agent" => AIAgentActionType::SendMessageToAgent { addresses: vec![required_string(input, "agent_id")?], subject: String::new(), diff --git a/app/src/pane_group/mod_tests.rs b/app/src/pane_group/mod_tests.rs index 27e0fe24..9abdcd2f 100644 --- a/app/src/pane_group/mod_tests.rs +++ b/app/src/pane_group/mod_tests.rs @@ -389,6 +389,7 @@ fn persisted_remote_child_conversation( conversation_data: serde_json::to_string(&AgentConversationData { agent_backend: Default::default(), active_provider_run_json: None, + local_agent_groups: Vec::new(), server_conversation_token: Some("restored-child-token".to_string()), conversation_usage_metadata: None, reverted_action_ids: None, diff --git a/app/src/settings_view/ai_page.rs b/app/src/settings_view/ai_page.rs index 7abed1ad..87b6ed37 100644 --- a/app/src/settings_view/ai_page.rs +++ b/app/src/settings_view/ai_page.rs @@ -73,8 +73,8 @@ use crate::ai::execution_profiles::{ ActionPermission, WriteToPtyPermission, }; use crate::ai::llms::{ - is_using_api_key_for_provider, LLMContextWindow, LLMId, LLMPreferences, LLMPreferencesEvent, - LLMProvider, + is_using_api_key_for_provider, openai_model_context_size, LLMContextWindow, LLMId, + LLMPreferences, LLMPreferencesEvent, LLMProvider, }; use crate::ai::mcp::TemplatableMCPServerManager; use crate::ai::paths::host_native_absolute_path; @@ -7236,13 +7236,26 @@ impl ProviderSettingsWidget { } } - fn format_openai_model_details(model: &OpenAIModelConfig) -> String { + fn format_openai_model_details( + model: &OpenAIModelConfig, + provider: &OpenAIProviderConfig, + ) -> String { + let context_size = if provider.kind == OpenAIProviderKind::ChatGPTSubscription { + // Use the same limit as requests, including for older saved catalogs + // that still contain the optional long-context maximum. + openai_model_context_size(model, provider.kind) + } else { + model.context_size + }; let mut details = vec![ if model.enabled { "Enabled" } else { "Disabled" }.to_string(), - format!("Context: {}", model.context_size), + format!("Context: {context_size}"), ]; - if let Some(default_context_size) = model.default_context_size { + if let Some(default_context_size) = model + .default_context_size + .filter(|_| provider.kind != OpenAIProviderKind::ChatGPTSubscription) + { details.push(format!("Default context: {default_context_size}")); } if let Some(max_input_tokens) = model.max_input_tokens { @@ -7338,7 +7351,7 @@ impl ProviderSettingsWidget { Self::render_model_row( model.display_name.clone(), model.model_id.clone(), - Self::format_openai_model_details(model), + Self::format_openai_model_details(model, provider), appearance, ) }) diff --git a/app/src/terminal/view/load_ai_conversation.rs b/app/src/terminal/view/load_ai_conversation.rs index 9ab4a6d3..2a3cbaa0 100644 --- a/app/src/terminal/view/load_ai_conversation.rs +++ b/app/src/terminal/view/load_ai_conversation.rs @@ -944,6 +944,7 @@ impl TerminalView { let conversation_data = AgentConversationData { agent_backend: Default::default(), active_provider_run_json: None, + local_agent_groups: Vec::new(), server_conversation_token: None, conversation_usage_metadata: None, reverted_action_ids: None, diff --git a/crates/ai/src/agent/action_result/mod.rs b/crates/ai/src/agent/action_result/mod.rs index feeaf30b..dee3a774 100644 --- a/crates/ai/src/agent/action_result/mod.rs +++ b/crates/ai/src/agent/action_result/mod.rs @@ -1766,7 +1766,7 @@ impl RunAgentsResult { "instruction": if children_completed { "Child agents reached terminal states. Use their structured outputs and errors to complete the task." } else { - "Child agents have only been launched, not completed. Wait for child-agent updates before repeating this work or reporting final results." + "Child agents are running independently. This turn yields automatically; the user may continue talking to the orchestrator. One group-completion message will arrive after every child finishes. Do not repeat delegated work or report final results before that message." }, "model_id": model_id, "harness_type": harness_type, diff --git a/crates/galaxy_agent_core/src/provider_run.rs b/crates/galaxy_agent_core/src/provider_run.rs index 5c1ee3ba..c38f3423 100644 --- a/crates/galaxy_agent_core/src/provider_run.rs +++ b/crates/galaxy_agent_core/src/provider_run.rs @@ -1261,9 +1261,26 @@ impl ProviderRun { }); }; let next_epoch = self.next_epoch()?; + let yielded_to_children = batch.calls.iter().any(|pending| { + is_local_agent_launch(&pending.call) + && pending + .state + .result() + .is_some_and(|result| result.status == ToolResultStatus::Success) + }); self.transcript.push(tool_result_message(results)); self.epoch = next_epoch; - self.state = ProviderRunState::ReadyToCallModel; + // Launch acknowledgement is a yield, not an invitation to issue more tools. + // The client persists child membership and supplies a new input once the group finishes. + self.state = if yielded_to_children { + ProviderRunState::Done { + completion: ProviderRunCompletion { + stop_reason: StopReason::Completed, + }, + } + } else { + ProviderRunState::ReadyToCallModel + }; Ok(()) } @@ -1382,12 +1399,25 @@ impl ProviderRun { fn build_tool_batch(&self, turn: &CompletedModelTurn) -> PendingToolBatch { let policy = ToolPolicy::from_names(turn.advertised_tools.iter().cloned()); + let has_launch = turn.tool_calls.iter().any(|call| { + is_local_agent_launch(call) + && call.arguments.is_object() + && turn.advertised_tools.contains(&call.name) + }); let calls = turn .tool_calls .iter() .cloned() .map(|call| { - let state = if !call.arguments.is_object() { + let state = if has_launch && !is_local_agent_launch(&call) { + PendingToolCallState::Resolved { + result: ToolResult { + call_id: call.id.clone(), + content: "Not executed: launching a child group yields this turn. Wait for the group's completion before scheduling dependent work.".to_string(), + status: ToolResultStatus::Error, + }, + } + } else if !call.arguments.is_object() { PendingToolCallState::Resolved { result: ToolResult { call_id: call.id.clone(), @@ -1626,6 +1656,11 @@ fn usage_contains(total: &Usage, part: &Usage) -> bool { && total.cache_creation_input_tokens >= part.cache_creation_input_tokens } +fn is_local_agent_launch(call: &ToolCall) -> bool { + // Galaxy normalizes direct-provider delegation to local child execution. + matches!(call.name.as_str(), "run_agents" | "start_agent") +} + fn validate_model_turn(turn: &CompletedModelTurn) -> Result<(), ProviderRunProtocolError> { for part in &turn.assistant_content { match part { diff --git a/crates/galaxy_agent_core/src/provider_run_tests.rs b/crates/galaxy_agent_core/src/provider_run_tests.rs index 447080b7..7a445708 100644 --- a/crates/galaxy_agent_core/src/provider_run_tests.rs +++ b/crates/galaxy_agent_core/src/provider_run_tests.rs @@ -1206,20 +1206,20 @@ fn restore_normalization_preserves_selected_executing_tools_for_recovery() { &mut run, tool_turn( vec![ - tool_call("run-agents", "run_agents"), + tool_call("command", "run_command"), tool_call("read", "read_files"), ], - &["run_agents", "read_files"], + &["run_command", "read_files"], ), ); - run.start_tool(&batch.work_id, "run-agents").unwrap(); + run.start_tool(&batch.work_id, "command").unwrap(); run.start_tool(&batch.work_id, "read").unwrap(); let normalization = run - .normalize_after_restore_with_recoverable_calls(&HashSet::from(["run-agents".to_string()])) + .normalize_after_restore_with_recoverable_calls(&HashSet::from(["command".to_string()])) .unwrap(); - assert_eq!(normalization.recovery_call_ids, vec!["run-agents"]); + assert_eq!(normalization.recovery_call_ids, vec!["command"]); assert_eq!(normalization.interrupted_call_ids, vec!["read"]); let ProviderRunState::AwaitingTools { batch } = run.state() else { panic!("recovered tool batch should remain pending"); @@ -1252,7 +1252,7 @@ fn recovery_pending_tool_survives_another_restore_and_completes_once() { restored .complete_tool( &batch.work_id, - successful_result("run-agents", "children completed"), + successful_result("run-agents", "children launched"), ) .unwrap(); assert!(matches!( @@ -1263,7 +1263,7 @@ fn recovery_pending_tool_survives_another_restore_and_completes_once() { Err(ProviderRunProtocolError::DuplicateToolUpdate { .. }) )); restored.commit_tool_batch(&batch.work_id).unwrap(); - assert_eq!(restored.state().phase(), ProviderRunPhase::ReadyToCallModel); + assert_eq!(restored.state().phase(), ProviderRunPhase::Done); } #[test] @@ -1465,3 +1465,58 @@ fn multipart_archive_is_not_multiplied_and_legacy_duplicates_are_repaired() { archive_tool_results(&mut archive, transcript); assert_eq!(archive, expected); } + +#[test] +fn local_child_launch_yields_and_does_not_execute_batched_follow_up_tools() { + let mut run = run(); + let batch = accept_tool_turn( + &mut run, + tool_turn( + vec![ + tool_call("children", "run_agents"), + tool_call("edit", "apply_file_diffs"), + ], + &["run_agents", "apply_file_diffs"], + ), + ); + assert!(matches!( + batch.calls[0].state, + PendingToolCallState::Proposed + )); + assert!( + matches!(&batch.calls[1].state, PendingToolCallState::Resolved { result } + if result.status == ToolResultStatus::Error) + ); + run.complete_tool( + &batch.work_id, + successful_result("children", "Children launched"), + ) + .unwrap(); + run.commit_tool_batch(&batch.work_id).unwrap(); + assert!(matches!(run.state(), ProviderRunState::Done { .. })); + assert_serialization_round_trip(&run); + assert!(matches!( + run.next_step().unwrap(), + Some(ProviderRunStep::Done(_)) + )); +} + +#[test] +fn failed_child_launch_allows_a_corrective_model_turn() { + let mut run = run(); + let batch = accept_tool_turn( + &mut run, + tool_turn(vec![tool_call("children", "run_agents")], &["run_agents"]), + ); + run.complete_tool( + &batch.work_id, + ToolResult { + call_id: "children".to_string(), + content: "Launch failed".to_string(), + status: ToolResultStatus::Error, + }, + ) + .unwrap(); + run.commit_tool_batch(&batch.work_id).unwrap(); + assert!(matches!(run.state(), ProviderRunState::ReadyToCallModel)); +} diff --git a/crates/persistence/src/model.rs b/crates/persistence/src/model.rs index 1783ff7a..dd849508 100644 --- a/crates/persistence/src/model.rs +++ b/crates/persistence/src/model.rs @@ -1075,6 +1075,17 @@ pub struct AcpConversationData { } // Serializes to `conversation_data` column in `agent_conversations`. +/// Durable membership and delivery revision for one local delegation group. +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +pub struct LocalAgentGroup { + pub id: String, + pub children: Vec, + pub launch_errors: Vec, + pub sealed: bool, + pub revision: u64, + pub notified: bool, +} + #[derive(Debug, Serialize, Deserialize, Clone, Default)] pub struct AgentConversationData { #[serde(default, skip_serializing_if = "AgentBackend::is_provider")] @@ -1082,6 +1093,9 @@ pub struct AgentConversationData { /// Versioned application-owned snapshot for an active direct-provider run. #[serde(default, skip_serializing_if = "Option::is_none")] pub active_provider_run_json: Option, + /// Local child groups outlive the parent generation that launched them. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub local_agent_groups: Vec, pub server_conversation_token: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub conversation_usage_metadata: Option, diff --git a/crates/persistence/src/model_tests.rs b/crates/persistence/src/model_tests.rs index 6adf00f4..b7be99bf 100644 --- a/crates/persistence/src/model_tests.rs +++ b/crates/persistence/src/model_tests.rs @@ -138,6 +138,7 @@ fn agent_conversation_data_roundtrips_last_event_sequence() { let data = AgentConversationData { agent_backend: Default::default(), active_provider_run_json: None, + local_agent_groups: Vec::new(), server_conversation_token: None, conversation_usage_metadata: None, reverted_action_ids: None, @@ -241,6 +242,7 @@ fn agent_conversation_data_roundtrips_remote_child_marker() { let data = AgentConversationData { agent_backend: Default::default(), active_provider_run_json: None, + local_agent_groups: Vec::new(), server_conversation_token: None, conversation_usage_metadata: None, reverted_action_ids: None, @@ -269,6 +271,7 @@ fn agent_conversation_data_roundtrips_optimistic_root_marker() { let data = AgentConversationData { agent_backend: Default::default(), active_provider_run_json: None, + local_agent_groups: Vec::new(), server_conversation_token: None, conversation_usage_metadata: None, reverted_action_ids: None, @@ -309,6 +312,7 @@ fn agent_conversation_data_skips_serializing_none_last_event_sequence() { let data = AgentConversationData { agent_backend: Default::default(), active_provider_run_json: None, + local_agent_groups: Vec::new(), server_conversation_token: None, conversation_usage_metadata: None, reverted_action_ids: None, @@ -339,6 +343,7 @@ fn agent_conversation_data_roundtrips_pinned() { let data = AgentConversationData { agent_backend: Default::default(), active_provider_run_json: None, + local_agent_groups: Vec::new(), server_conversation_token: None, conversation_usage_metadata: None, reverted_action_ids: None, @@ -367,6 +372,7 @@ fn agent_conversation_data_skips_serializing_unpinned() { let data = AgentConversationData { agent_backend: Default::default(), active_provider_run_json: None, + local_agent_groups: Vec::new(), server_conversation_token: None, conversation_usage_metadata: None, reverted_action_ids: None, diff --git a/vendor/block/src/lib.rs b/vendor/block/src/lib.rs index 74baa0a9..52c95ff9 100644 --- a/vendor/block/src/lib.rs +++ b/vendor/block/src/lib.rs @@ -340,60 +340,5 @@ impl BlockDescriptor { } #[cfg(test)] -mod tests { - use test_utils::*; - use super::{ConcreteBlock, RcBlock}; - - #[test] - fn test_call_block() { - let block = get_int_block_with(13); - unsafe { - assert!(block.call(()) == 13); - } - } - - #[test] - fn test_call_block_args() { - let block = get_add_block_with(13); - unsafe { - assert!(block.call((2,)) == 15); - } - } - - #[test] - fn test_create_block() { - let block = ConcreteBlock::new(|| 13); - let result = invoke_int_block(&block); - assert!(result == 13); - } - - #[test] - fn test_create_block_args() { - let block = ConcreteBlock::new(|a: i32| a + 5); - let result = invoke_add_block(&block, 6); - assert!(result == 11); - } - - #[test] - fn test_concrete_block_copy() { - let s = "Hello!".to_string(); - let expected_len = s.len() as i32; - let block = ConcreteBlock::new(move || s.len() as i32); - assert!(invoke_int_block(&block) == expected_len); - - let copied = block.copy(); - assert!(invoke_int_block(&copied) == expected_len); - } - - #[test] - fn test_concrete_block_stack_copy() { - fn make_block() -> RcBlock<(), i32> { - let x = 7; - let block = ConcreteBlock::new(move || x); - block.copy() - } - - let block = make_block(); - assert!(invoke_int_block(&block) == 7); - } -} +#[path = "lib_tests.rs"] +mod tests; diff --git a/vendor/block/src/lib_tests.rs b/vendor/block/src/lib_tests.rs new file mode 100644 index 00000000..d144d534 --- /dev/null +++ b/vendor/block/src/lib_tests.rs @@ -0,0 +1,55 @@ +use test_utils::*; +use super::{ConcreteBlock, RcBlock}; + +#[test] +fn test_call_block() { + let block = get_int_block_with(13); + unsafe { + assert!(block.call(()) == 13); + } +} + +#[test] +fn test_call_block_args() { + let block = get_add_block_with(13); + unsafe { + assert!(block.call((2,)) == 15); + } +} + +#[test] +fn test_create_block() { + let block = ConcreteBlock::new(|| 13); + let result = invoke_int_block(&block); + assert!(result == 13); +} + +#[test] +fn test_create_block_args() { + let block = ConcreteBlock::new(|a: i32| a + 5); + let result = invoke_add_block(&block, 6); + assert!(result == 11); +} + +#[test] +fn test_concrete_block_copy() { + let s = "Hello!".to_string(); + let expected_len = s.len() as i32; + let block = ConcreteBlock::new(move || s.len() as i32); + assert!(invoke_int_block(&block) == expected_len); + + let copied = block.copy(); + assert!(invoke_int_block(&copied) == expected_len); +} + +#[test] +fn test_concrete_block_stack_copy() { + fn make_block() -> RcBlock<(), i32> { + let x = 7; + let block = ConcreteBlock::new(move || x); + block.copy() + } + + let block = make_block(); + assert!(invoke_int_block(&block) == 7); +}