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

Fix child startup readiness and ordered provider tools

Browse Source 
Wait for shell bootstrap before dispatching child prompts, serialize provider preprocessing, and make permission callbacks idempotent. Restrict task lists to concrete multistep plans and stop inferring completion from tool activity.

Update regression fixtures and resolve existing lint and test-layout failures. Verified formatting, both presubmit Clippy commands, and 354 targeted nextest tests.
This commit is contained in:
rkw6086
2026-09-09 13:55:34 -05:00
parent 4b703c019f
commit 1845b9e20e
41 changed files with 1097 additions and 895 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/ai/blocklist/action_model.rs
+83 -3
View File
@@ -780,10 +780,15 @@ pub struct BlocklistAIActionModel {
/// Permission-card rejections whose cancelled action result must not emit a second provider event.
denied_permissions: HashSet<(AIConversationId, AIAgentActionId)>,
pending_permissions: HashSet<(AIConversationId, AIAgentActionId)>,
/// Actions parked because their executor-specific UI or state was not ready yet.
not_ready_actions: NotReadyActionTracker,
/// Provider preprocessing waits for execution order, since it can read file contents.
/// The value is true while preprocessing is in flight.
deferred_provider_preprocessing: HashMap<(AIConversationId, AIAgentActionId), bool>,
/// Durable provider work identity for actions owned by an active provider run.
provider_tool_executions:
HashMap<(AIConversationId, AIAgentActionId), ProviderToolExecutionRef>,
@@ -899,7 +904,9 @@ impl BlocklistAIActionModel {
running_actions: Default::default(),
action_order: Default::default(),
denied_permissions: Default::default(),
pending_permissions: Default::default(),
not_ready_actions: Default::default(),
deferred_provider_preprocessing: Default::default(),
provider_tool_executions: Default::default(),
terminal_view_id,
pending_preprocessed_actions: Default::default(),
@@ -1280,6 +1287,13 @@ impl BlocklistAIActionModel {
conversation_id: AIConversationId,
id: &AIAgentActionId,
) -> Option<AIActionStatus> {
if self
.deferred_provider_preprocessing
.get(&(conversation_id, id.clone()))
== Some(&true)
{
return Some(AIActionStatus::Preprocessing);
}
if let Some(status) = pending_action_status(
&self.pending_actions,
&self.running_actions,
@@ -1507,7 +1521,7 @@ impl BlocklistAIActionModel {
}
fn handle_not_executed_action(
&self,
&mut self,
action: &AIAgentAction,
reason: NotExecutedReason,
conversation_id: AIConversationId,
@@ -1515,6 +1529,12 @@ impl BlocklistAIActionModel {
) {
match reason {
NotExecutedReason::NeedsConfirmation => {
if !self
.pending_permissions
.insert((conversation_id, action.id.clone()))
{
return;
}
#[cfg(not(target_family = "wasm"))]
log_tool_event(
ctx,
@@ -1624,6 +1644,42 @@ impl BlocklistAIActionModel {
.get(&conversation_id)
.and_then(|queue| queue.iter().position(|action| &action.id == action_id))?;
let key = (conversation_id, action_id.clone());
if self.provider_tool_executions.contains_key(&key) && idx != 0 {
return None;
}
// Readiness callbacks also enter here directly, bypassing the normal queue loop.
// They must obey the same serial barriers before preprocessing or executing.
if let Some(current_phase) = self.action_execution_phase(conversation_id) {
let action = self.pending_actions[&conversation_id][idx].clone();
if !self.can_start_action_in_current_phase(&action, conversation_id, current_phase, ctx)
{
return None;
}
}
if let Some(in_flight) = self.deferred_provider_preprocessing.get_mut(&key) {
if *in_flight {
return None;
}
*in_flight = true;
let action = self.pending_actions[&conversation_id][idx].clone();
let execution_ref = self.provider_tool_execution_ref(conversation_id, action_id);
let future = self.preprocess_action(&action, conversation_id, ctx);
ctx.spawn(future, move |me, (), ctx| {
// A cancelled or replaced generation must never restart this action.
if me.provider_tool_execution_ref(conversation_id, &action.id) != execution_ref {
return;
}
me.deferred_provider_preprocessing.remove(&key);
if is_user_initiated {
me.execute_action(&action.id, conversation_id, ctx);
} else {
me.try_to_execute_available_actions(conversation_id, ctx);
}
});
return None;
}
let action = self
.pending_actions
.get_mut(&conversation_id)?
@@ -1640,7 +1696,13 @@ impl BlocklistAIActionModel {
"permission_kind": format!("{:?}", permission_kind_for_action(&action.action)),
"phase": format!("{phase:?}"),
});
if is_user_initiated {
let resolves_permission = self.pending_permissions.contains(&key)
&& (is_user_initiated
|| self.executor.update(ctx, |executor, ctx| {
executor.can_autoexecute_action(&action, conversation_id, ctx)
}));
if resolves_permission {
self.pending_permissions.remove(&key);
#[cfg(not(target_family = "wasm"))]
log_tool_event(
ctx,
@@ -1884,7 +1946,15 @@ impl BlocklistAIActionModel {
for action in actions.iter() {
action_ids.insert(action.id.clone());
preprocess_future.push(self.preprocess_action(action, conversation_id, ctx));
if self
.provider_tool_executions
.contains_key(&(conversation_id, action.id.clone()))
{
self.deferred_provider_preprocessing
.insert((conversation_id, action.id.clone()), false);
} else {
preprocess_future.push(self.preprocess_action(action, conversation_id, ctx));
}
}
let preprocess_id = self
@@ -2053,6 +2123,8 @@ impl BlocklistAIActionModel {
reason: Option<CancellationReason>,
ctx: &mut ModelContext<Self>,
) {
self.deferred_provider_preprocessing
.retain(|(conv_id, _), _| *conv_id != conversation_id);
self.executor.update(ctx, |executor, ctx| {
executor.cancel_all_running_async_actions_for_conversation(conversation_id, reason, ctx)
});
@@ -2253,12 +2325,16 @@ impl BlocklistAIActionModel {
conversation_id: AIConversationId,
ctx: &mut ModelContext<Self>,
) {
self.deferred_provider_preprocessing
.retain(|(conv_id, _), _| *conv_id != conversation_id);
self.past_action_results
.retain(|(conv_id, _), _| *conv_id != conversation_id);
self.provider_tool_executions
.retain(|(conv_id, _), _| *conv_id != conversation_id);
self.denied_permissions
.retain(|(conv_id, _)| *conv_id != conversation_id);
self.pending_permissions
.retain(|(conv_id, _)| *conv_id != conversation_id);
self.pending_actions.remove(&conversation_id);
self.running_actions.remove(&conversation_id);
self.finished_action_results.remove(&conversation_id);
@@ -2372,6 +2448,10 @@ impl BlocklistAIActionModel {
let execution_ref = self
.provider_tool_executions
.remove(&(conversation_id, action_result.id.clone()));
self.deferred_provider_preprocessing
.remove(&(conversation_id, action_result.id.clone()));
self.pending_permissions
.remove(&(conversation_id, action_result.id.clone()));
let permission_denied = self
.denied_permissions
.remove(&(conversation_id, action_result.id.clone()));
app/src/ai/blocklist/action_model/execute/start_agent_tests.rs
+20 -20
View File
@@ -16,11 +16,23 @@ use crate::test_util::settings::initialize_history_persistence_for_tests;
const FIRST_REQUEST_ID: StartAgentRequestId = StartAgentRequestId::from_raw_for_test(0);
/// Stable placeholder run_id assigned to the parent conversation in tests
/// that exercise the server-backed Oz child path. Tests without this id
/// exercise the direct-provider local child path instead.
/// Stable server run ID for parent conversations. Child execution mode, rather
/// than the presence of this ID, selects startup versus completion waits.
const PARENT_RUN_ID: &str = "00000000-0000-0000-0000-000000000001";
fn remote_execution_mode() -> StartAgentExecutionMode {
StartAgentExecutionMode::Remote {
environment_id: "env-123".to_string(),
skill_references: vec![],
model_id: String::new(),
computer_use_enabled: false,
worker_host: String::new(),
harness_type: "oz".to_string(),
title: String::new(),
auth_secret_name: None,
}
}
#[derive(Default)]
struct CapturedDirectProviderChildLinks(Vec<(AIAgentActionId, AIConversationId, AIConversationId)>);
@@ -394,10 +406,7 @@ fn execute_returns_error_when_child_startup_is_blocked_before_initialization() {
ctx,
);
});
let action = build_start_agent_action(
StartAgentVersion::V1,
StartAgentExecutionMode::local_with_defaults(),
);
let action = build_start_agent_action(StartAgentVersion::V1, remote_execution_mode());
let execution = executor.update(&mut app, |executor, ctx| {
let input = ExecuteActionInput {
@@ -581,10 +590,7 @@ fn execute_resolves_success_when_request_linkage_happens_after_child_already_sta
ctx,
);
});
let action = build_start_agent_action(
StartAgentVersion::V1,
StartAgentExecutionMode::local_with_defaults(),
);
let action = build_start_agent_action(StartAgentVersion::V1, remote_execution_mode());
let execution = executor.update(&mut app, |executor, ctx| {
let input = ExecuteActionInput {
@@ -739,10 +745,7 @@ fn hosted_child_link_does_not_publish_direct_provider_panel_event() {
ctx,
);
});
let action = build_start_agent_action(
StartAgentVersion::V1,
StartAgentExecutionMode::local_with_defaults(),
);
let action = build_start_agent_action(StartAgentVersion::V1, remote_execution_mode());
let execution = executor.update(&mut app, |executor, ctx| {
let input = ExecuteActionInput {
@@ -1769,7 +1772,7 @@ struct PendingChildLaunch {
child_conversation_id: AIConversationId,
}
/// Dispatches a local child launch and creates (but does not yet link) its
/// Dispatches a remote child launch and creates (but does not yet link) its
/// child conversation, leaving one in-flight pending in the executor with a
/// model subscribed to capture `CleanupFailedChildLaunch` events. Tests link
/// the child and drive it to a terminal state, then assert on cleanup. The
@@ -1803,10 +1806,7 @@ fn dispatch_pending_child_launch(
ctx,
);
});
let action = build_start_agent_action(
StartAgentVersion::V1,
StartAgentExecutionMode::local_with_defaults(),
);
let action = build_start_agent_action(StartAgentVersion::V1, remote_execution_mode());
// The pending lives in the executor regardless of the returned execution,
// and cleanup is emitted synchronously from the child status update, so the
// action-result plumbing is discarded.
app/src/ai/blocklist/action_model/execute/wait_for_events_tests.rs
+17 -13
View File
@@ -19,6 +19,7 @@ use crate::ai::blocklist::orchestration_event_streamer::OrchestrationEventStream
use crate::ai::blocklist::BlocklistAIHistoryModel;
use crate::server::server_api::ai::{AIClient, MockAIClient};
use crate::server::server_api::ServerApiProvider;
use crate::test_util::settings::initialize_history_persistence_for_tests;
#[test]
fn watchdog_timeout_constants_match_documented_values() {
@@ -88,24 +89,12 @@ fn execute_invokes_parent_registration_and_honors_child_short_circuit() {
// without a server fetch (asserted via the mock's times(0) expectation),
// and the wait still flips the conversation into WaitingForEvents.
App::test((), |mut app| async move {
initialize_history_persistence_for_tests(&mut app);
let _flag_guard = FeatureFlag::WaitForEventsParentRegistration.override_enabled(true);
let terminal_view_id = EntityId::new();
let history_model = app.add_singleton_model(|_| BlocklistAIHistoryModel::new(vec![], &[]));
// A streamer whose server fetch must never be called: the child
// short-circuit precedes any `get_ambient_agent_task` call.
let mut mock = MockAIClient::new();
mock.expect_get_ambient_agent_task().times(0);
let ai_client: Arc<dyn AIClient> = Arc::new(mock);
let server_api = ServerApiProvider::new_for_test().get();
// Held for the lifetime of the test so the mock's times(0) expectation
// is verified on drop; resolved internally by `execute()` via
// `OrchestrationEventStreamer::handle`.
let _streamer = app.add_singleton_model(|ctx| {
OrchestrationEventStreamer::new_with_clients_for_test(ai_client, server_api, ctx)
});
let executor = app.add_model(|ctx| WaitForEventsExecutor::new(terminal_view_id, ctx));
// Child conversation: own run_id plus a parent_agent_id.
@@ -123,6 +112,21 @@ fn execute_invokes_parent_registration_and_honors_child_short_circuit() {
);
});
// Subscribe after restoring the fixture so unrelated restore-time harness
// discovery does not count as a parent-registration fetch.
// A streamer whose server fetch must never be called: the child
// short-circuit precedes any `get_ambient_agent_task` call.
let mut mock = MockAIClient::new();
mock.expect_get_ambient_agent_task().times(0);
let ai_client: Arc<dyn AIClient> = Arc::new(mock);
let server_api = ServerApiProvider::new_for_test().get();
// Held for the lifetime of the test so the mock's times(0) expectation
// is verified on drop; resolved internally by `execute()` via
// `OrchestrationEventStreamer::handle`.
let _streamer = app.add_singleton_model(|ctx| {
OrchestrationEventStreamer::new_with_clients_for_test(ai_client, server_api, ctx)
});
let action = AIAgentAction {
id: AIAgentActionId::from("wait-action".to_string()),
action: AIAgentActionType::WaitForEvents {
app/src/ai/blocklist/action_model_tests.rs
+70
View File
@@ -7,6 +7,7 @@ use crate::ai::agent::{
AIAgentAction, AIAgentActionResultType, AIAgentActionType, AnyFileContent, FileContext,
GrepResult, ReadFilesResult,
};
use crate::test_util::terminal::{add_window_with_terminal, initialize_app_for_terminal_view};
fn make_action_result(id: &str) -> Arc<AIAgentActionResult> {
Arc::new(AIAgentActionResult {
@@ -54,6 +55,75 @@ fn pending_tool_batch(call_ids: &[&str]) -> PendingToolBatch {
}
}
#[test]
fn readiness_retry_cannot_preprocess_across_a_serial_barrier_or_skip_a_pending_tool() {
warpui::App::test((), |mut app| async move {
initialize_app_for_terminal_view(&mut app);
let terminal = add_window_with_terminal(&mut app, None);
terminal.update(&mut app, |view, ctx| {
let terminal_view_id = ctx.view_id();
let relevant_files = ctx.add_model(GetRelevantFilesController::new);
let action_model = ctx.add_model(|ctx| {
BlocklistAIActionModel::new(
view.model.clone(),
view.active_session().clone(),
view.model_event_dispatcher(),
relevant_files,
terminal_view_id,
ctx,
)
});
action_model.update(ctx, |model, ctx| {
let conversation_id = AIConversationId::new();
let batch = pending_tool_batch(&["first", "second"]);
let actions = vec![action("first"), action("second")];
model.provider_tool_executions.extend(
provider_action_correlations(&actions, conversation_id, &batch).unwrap(),
);
for action in &actions {
model
.deferred_provider_preprocessing
.insert((conversation_id, action.id.clone()), false);
}
model
.pending_actions
.insert(conversation_id, actions.clone().into());
model.add_running_action(
conversation_id,
AIAgentActionId::from("running".to_string()),
RunningActionPhase::Serial,
);
assert!(model
.start_pending_action_by_id(
&actions[0].id,
conversation_id,
ActionExecutionInitiator::Automatic,
ctx
)
.is_none());
assert!(
!model.deferred_provider_preprocessing
[&(conversation_id, actions[0].id.clone())]
);
model.running_actions.remove(&conversation_id);
assert!(model
.start_pending_action_by_id(
&actions[1].id,
conversation_id,
ActionExecutionInitiator::User,
ctx
)
.is_none());
assert!(
!model.deferred_provider_preprocessing
[&(conversation_id, actions[1].id.clone())]
);
assert_eq!(model.pending_actions[&conversation_id].len(), 2);
});
});
});
}
fn count_startable_actions_for_pass(phases: &[(RunningActionPhase, bool)]) -> usize {
let mut current_phase = None;
let mut count = 0;
app/src/ai/blocklist/block/cli_controller.rs
+2 -83
View File
@@ -1158,86 +1158,5 @@ fn matches_requested_command_identity(
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn stop_takeover_does_not_request_a_completion_assessment() {
let state = LongRunningCommandControlState::User {
reason: UserTakeOverReason::Stop,
};
assert!(!should_request_completion_assessment(Some(&state)));
}
#[test]
fn non_stop_control_states_can_request_a_completion_assessment() {
let agent_state = LongRunningCommandControlState::Agent {
is_blocked: false,
should_hide_responses: false,
};
let transfer_state = LongRunningCommandControlState::User {
reason: UserTakeOverReason::TransferFromAgent {
reason: "needs user input".to_owned(),
},
};
assert!(should_request_completion_assessment(None));
assert!(should_request_completion_assessment(Some(&agent_state)));
assert!(should_request_completion_assessment(Some(&transfer_state)));
}
#[test]
fn prose_monitor_turn_is_nudged_once_until_a_tool_action_runs() {
assert!(should_nudge_monitor_turn(false, false));
assert!(!should_nudge_monitor_turn(false, true));
assert!(!should_nudge_monitor_turn(true, false));
}
#[test]
fn shell_control_event_must_match_conversation_and_requested_command() {
let active_conversation_id = AIConversationId::new();
let other_conversation_id = AIConversationId::new();
let active_action_id = AIAgentActionId::from("same-action".to_owned());
let other_action_id = AIAgentActionId::from("other-action".to_owned());
assert!(matches_active_requested_command(
active_conversation_id,
&active_action_id,
Some(active_conversation_id),
Some(&active_action_id),
));
assert!(!matches_active_requested_command(
other_conversation_id,
&active_action_id,
Some(active_conversation_id),
Some(&active_action_id),
));
assert!(!matches_active_requested_command(
active_conversation_id,
&other_action_id,
Some(active_conversation_id),
Some(&active_action_id),
));
}
#[test]
fn requested_command_identity_rejects_duplicate_id_from_another_conversation() {
let active_conversation_id = AIConversationId::new();
let other_conversation_id = AIConversationId::new();
let duplicate_action_id = AIAgentActionId::from("duplicate-action".to_owned());
assert!(matches_requested_command_identity(
active_conversation_id,
&duplicate_action_id,
Some(active_conversation_id),
Some(&duplicate_action_id),
));
assert!(!matches_requested_command_identity(
other_conversation_id,
&duplicate_action_id,
Some(active_conversation_id),
Some(&duplicate_action_id),
));
}
}
#[path = "cli_controller_tests.rs"]
mod tests;
app/src/ai/blocklist/block/cli_controller_tests.rs
+81
View File
@@ -0,0 +1,81 @@
use super::*;
#[test]
fn stop_takeover_does_not_request_a_completion_assessment() {
let state = LongRunningCommandControlState::User {
reason: UserTakeOverReason::Stop,
};
assert!(!should_request_completion_assessment(Some(&state)));
}
#[test]
fn non_stop_control_states_can_request_a_completion_assessment() {
let agent_state = LongRunningCommandControlState::Agent {
is_blocked: false,
should_hide_responses: false,
};
let transfer_state = LongRunningCommandControlState::User {
reason: UserTakeOverReason::TransferFromAgent {
reason: "needs user input".to_owned(),
},
};
assert!(should_request_completion_assessment(None));
assert!(should_request_completion_assessment(Some(&agent_state)));
assert!(should_request_completion_assessment(Some(&transfer_state)));
}
#[test]
fn prose_monitor_turn_is_nudged_once_until_a_tool_action_runs() {
assert!(should_nudge_monitor_turn(false, false));
assert!(!should_nudge_monitor_turn(false, true));
assert!(!should_nudge_monitor_turn(true, false));
}
#[test]
fn shell_control_event_must_match_conversation_and_requested_command() {
let active_conversation_id = AIConversationId::new();
let other_conversation_id = AIConversationId::new();
let active_action_id = AIAgentActionId::from("same-action".to_owned());
let other_action_id = AIAgentActionId::from("other-action".to_owned());
assert!(matches_active_requested_command(
active_conversation_id,
&active_action_id,
Some(active_conversation_id),
Some(&active_action_id),
));
assert!(!matches_active_requested_command(
other_conversation_id,
&active_action_id,
Some(active_conversation_id),
Some(&active_action_id),
));
assert!(!matches_active_requested_command(
active_conversation_id,
&other_action_id,
Some(active_conversation_id),
Some(&active_action_id),
));
}
#[test]
fn requested_command_identity_rejects_duplicate_id_from_another_conversation() {
let active_conversation_id = AIConversationId::new();
let other_conversation_id = AIConversationId::new();
let duplicate_action_id = AIAgentActionId::from("duplicate-action".to_owned());
assert!(matches_requested_command_identity(
active_conversation_id,
&duplicate_action_id,
Some(active_conversation_id),
Some(&duplicate_action_id),
));
assert!(!matches_requested_command_identity(
other_conversation_id,
&duplicate_action_id,
Some(active_conversation_id),
Some(&duplicate_action_id),
));
}
app/src/ai/blocklist/controller_tests.rs
-1
View File
@@ -267,7 +267,6 @@ fn provider_snapshot(conversation_id: AIConversationId) -> super::ActiveProvider
max_context_tokens: Some(128_000),
capabilities: RuntimeCapabilities::provider(),
empty_output_message: None,
todo_items: None,
},
action_context: crate::ai::runtime::ProviderActionContext::new_for_test(
task_id.to_string(),