From 3fa7f2f00cceab030f8350c461d1dbcc5f038c53 Mon Sep 17 00:00:00 2001 From: Ryan Ward Date: Fri, 6 Nov 2026 21:25:17 +0000 Subject: [PATCH] Prevent idle event waits from stalling user follow-ups --- app/src/ai/blocklist/controller.rs | 40 +++- app/src/ai/blocklist/controller_tests.rs | 178 +++++++++++++++++- app/src/ai/provider/request_translator.rs | 6 +- .../ai/provider/request_translator_tests.rs | 40 ++++ 4 files changed, 256 insertions(+), 8 deletions(-) diff --git a/app/src/ai/blocklist/controller.rs b/app/src/ai/blocklist/controller.rs index f3c2c4a8..80bf1c14 100644 --- a/app/src/ai/blocklist/controller.rs +++ b/app/src/ai/blocklist/controller.rs @@ -1092,6 +1092,7 @@ enum ProviderCommandResult { fn convert_provider_tool_batch( action_context: &ProviderActionContext, batch: &PendingToolBatch, + has_expected_orchestration_event: bool, ) -> (Vec<(AIAgentAction, bool)>, Vec) { let mut actions = Vec::new(); let mut invalid_results = Vec::new(); @@ -1100,7 +1101,18 @@ fn convert_provider_tool_batch( .iter() .filter(|pending| pending.state.result().is_none()) { - match action_context.action_from_tool_call(&pending.call) { + let action = action_context.action_from_tool_call(&pending.call).and_then(|action| { + if matches!(action.action, AIAgentActionType::WaitForEvents { .. }) { + if batch.calls.len() != 1 { + return Err("wait_for_events must be the only tool in its batch. Finish the other tool results first. Background commands deliver completion automatically; do not wait for them with this tool.".to_string()); + } + if !has_expected_orchestration_event { + return Err("No child-agent or orchestration notification source is pending. Handle the latest user request now: inspect the requested logs or answer from available evidence. Background commands deliver completion automatically. This tool cannot schedule a follow-up, watch a log file, or wait for a future incident.".to_string()); + } + } + Ok(action) + }); + match action { Ok(action) => actions.push(( action, matches!(pending.state, PendingToolCallState::RecoveryPending), @@ -8748,6 +8760,25 @@ impl BlocklistAIController { batch: PendingToolBatch, ctx: &mut ModelContext, ) { + // A generic wait has no way to observe files, incidents, or command output. + // Require a real orchestration source before parking a durable generation. + // Hosted tasks retain owner-side registration for out-of-band children/messages. + let has_expected_orchestration_event = batch.calls.len() == 1 + && batch.calls[0].call.name == "wait_for_events" + && { + let history = BlocklistAIHistoryModel::as_ref(ctx); + let events = OrchestrationEventService::as_ref(ctx); + #[cfg(not(target_family = "wasm"))] + let has_pending_events = events.has_pending_events(conversation_id); + #[cfg(target_family = "wasm")] + let has_pending_events = false; + history + .conversation(&conversation_id) + .is_some_and(|conversation| conversation.task_id().is_some()) + || !active_descendant_conversation_ids(history, conversation_id).is_empty() + || events.has_waiting_local_groups(conversation_id, ctx) + || has_pending_events + }; let Some(run) = self .active_provider_runs .get_mut(&conversation_id) @@ -8765,8 +8796,11 @@ impl BlocklistAIController { .map(|title| (pending.call.id.clone(), title)) }) .collect::>(); - let (mut converted_actions, invalid_results) = - convert_provider_tool_batch(&run.action_context, &batch); + let (mut converted_actions, invalid_results) = convert_provider_tool_batch( + &run.action_context, + &batch, + has_expected_orchestration_event, + ); // Seed each worker from the current durable transcript, including discoveries // made in this run that have not reached the persisted conversation yet. let history = BlocklistAIHistoryModel::as_ref(ctx); diff --git a/app/src/ai/blocklist/controller_tests.rs b/app/src/ai/blocklist/controller_tests.rs index af97c000..40f3f5f3 100644 --- a/app/src/ai/blocklist/controller_tests.rs +++ b/app/src/ai/blocklist/controller_tests.rs @@ -1455,7 +1455,8 @@ fn malformed_provider_tool_inputs_become_correlated_errors_without_dropping_vali panic!("expected provider tool batch"); }; - let (actions, errors) = super::convert_provider_tool_batch(&snapshot.action_context, &batch); + let (actions, errors) = + super::convert_provider_tool_batch(&snapshot.action_context, &batch, false); assert_eq!(actions.len(), 1); assert_eq!(actions[0].0.id.to_string(), "good-grep"); @@ -1492,7 +1493,8 @@ fn malformed_provider_tool_error_can_be_committed_and_run_continues() { let Some(ProviderRunStep::DispatchTools(batch)) = snapshot.run.next_step().unwrap() else { panic!("expected provider tool batch"); }; - let (actions, errors) = super::convert_provider_tool_batch(&snapshot.action_context, &batch); + let (actions, errors) = + super::convert_provider_tool_batch(&snapshot.action_context, &batch, false); assert!(actions.is_empty()); snapshot @@ -4595,3 +4597,175 @@ fn restored_provider_checkpoint_waits_for_explicit_resume() { }); }); } + +fn dispatch_follow_up_tool_batch( + snapshot: &mut super::ActiveProviderRunSnapshot, + calls: Vec, +) -> galaxy_agent_core::PendingToolBatch { + let Some(ProviderRunStep::CallModel(call)) = snapshot.run.next_step().unwrap() else { + panic!("expected model call"); + }; + let advertised_tools = calls.iter().map(|call| call.name.clone()).collect(); + snapshot + .run + .accept_model_turn( + &call.work_id, + CompletedModelTurn { + assistant_content: Vec::new(), + tool_calls: calls, + usage: Usage::default(), + stop_reason: StopReason::Completed, + advertised_tools, + }, + ) + .unwrap(); + let Some(ProviderRunStep::DispatchTools(batch)) = snapshot.run.next_step().unwrap() else { + panic!("expected tool batch"); + }; + batch +} + +#[test] +fn wait_without_an_event_source_returns_a_correlated_error_and_allows_another_model_turn() { + let mut snapshot = provider_snapshot(AIConversationId::new()); + let batch = dispatch_follow_up_tool_batch( + &mut snapshot, + vec![ToolCall { + id: "unsupported-wait".to_owned(), + name: "wait_for_events".to_owned(), + arguments: serde_json::json!({}), + }], + ); + let (actions, errors) = + super::convert_provider_tool_batch(&snapshot.action_context, &batch, false); + assert_eq!(actions.len(), 0); + assert_eq!(errors.len(), 1); + assert_eq!(errors[0].call_id, "unsupported-wait"); + assert_eq!(errors[0].status, galaxy_agent_core::ToolResultStatus::Error); + assert!(errors[0] + .content + .contains("No child-agent or orchestration notification source")); + assert!(errors[0].content.contains("cannot schedule a follow-up")); + + snapshot + .run + .complete_tool(&batch.work_id, errors[0].clone()) + .unwrap(); + snapshot.run.commit_tool_batch(&batch.work_id).unwrap(); + let MessageContent::MultiPart(parts) = &snapshot.run.transcript().last().unwrap().content + else { + panic!("expected committed tool result"); + }; + assert!(matches!(&parts[0], ContentPart::ToolResult { + tool_use_id, is_error: true, .. + } if tool_use_id == "unsupported-wait")); + assert!(matches!( + snapshot.run.next_step().unwrap(), + Some(ProviderRunStep::CallModel(_)) + )); +} + +#[test] +fn sole_wait_with_an_expected_orchestration_notification_remains_executable() { + let mut snapshot = provider_snapshot(AIConversationId::new()); + let batch = dispatch_follow_up_tool_batch( + &mut snapshot, + vec![ToolCall { + id: "child-wait".to_owned(), + name: "wait_for_events".to_owned(), + arguments: serde_json::json!({"idle_timeout_seconds": 45}), + }], + ); + let (actions, errors) = + super::convert_provider_tool_batch(&snapshot.action_context, &batch, true); + assert_eq!(errors.len(), 0); + assert_eq!(actions.len(), 1); + assert_eq!(actions[0].0.id.to_string(), "child-wait"); + assert!( + matches!(&actions[0].0.action, AIAgentActionType::WaitForEvents { + tool_call_id, idle_timeout_seconds: 45, + } if tool_call_id == "child-wait") + ); + assert!( + matches!(snapshot.run.state(), ProviderRunState::AwaitingTools { batch } if !batch.is_complete()) + ); +} + +#[test] +fn mixed_shell_launch_and_wait_preserves_the_launch_and_yields_after_every_result_commits() { + let mut snapshot = provider_snapshot(AIConversationId::new()); + let batch = dispatch_follow_up_tool_batch( + &mut snapshot, + vec![ + ToolCall { + id: "job-launch".to_owned(), + name: "run_shell_command".to_owned(), + arguments: serde_json::json!({"command": "echo ready"}), + }, + ToolCall { + id: "mixed-wait".to_owned(), + name: "wait_for_events".to_owned(), + arguments: serde_json::json!({}), + }, + ], + ); + let (actions, errors) = + super::convert_provider_tool_batch(&snapshot.action_context, &batch, true); + assert_eq!(actions.len(), 1); + assert_eq!(actions[0].0.id.to_string(), "job-launch"); + assert!( + matches!(&actions[0].0.action, AIAgentActionType::RequestCommandOutput { command, .. } if command == "echo ready") + ); + assert_eq!(errors.len(), 1); + assert_eq!(errors[0].call_id, "mixed-wait"); + assert_eq!(errors[0].status, galaxy_agent_core::ToolResultStatus::Error); + assert!(errors[0].content.contains("only tool in its batch")); + + snapshot + .run + .complete_tool(&batch.work_id, errors[0].clone()) + .unwrap(); + let error = snapshot.run.commit_tool_batch(&batch.work_id).unwrap_err(); + assert!(error.to_string().contains("job-launch")); + assert!(matches!( + snapshot.run.state(), + ProviderRunState::AwaitingTools { .. } + )); + snapshot + .run + .start_tool(&batch.work_id, "job-launch") + .unwrap(); + snapshot + .run + .complete_tool( + &batch.work_id, + galaxy_agent_core::ToolResult { + call_id: "job-launch".to_owned(), + content: serde_json::json!({ + "background_job_started": {"job_id": "job-1", "command": "echo ready"} + }) + .to_string(), + status: galaxy_agent_core::ToolResultStatus::Success, + images: Vec::new(), + }, + ) + .unwrap(); + snapshot.run.commit_tool_batch(&batch.work_id).unwrap(); + + let MessageContent::MultiPart(parts) = &snapshot.run.transcript().last().unwrap().content + else { + panic!("expected both committed tool results"); + }; + assert_eq!(parts.len(), 2); + assert!(matches!(&parts[0], ContentPart::ToolResult { + tool_use_id, is_error: false, .. + } if tool_use_id == "job-launch")); + assert!(matches!(&parts[1], ContentPart::ToolResult { + tool_use_id, is_error: true, .. + } if tool_use_id == "mixed-wait")); + assert!( + matches!(snapshot.run.next_step().unwrap(), Some(ProviderRunStep::Done( + ProviderRunOutcome::Completed(completion) + )) if completion.stop_reason == StopReason::Completed) + ); +} diff --git a/app/src/ai/provider/request_translator.rs b/app/src/ai/provider/request_translator.rs index 303a3b0f..8f2686f1 100644 --- a/app/src/ai/provider/request_translator.rs +++ b/app/src/ai/provider/request_translator.rs @@ -1718,11 +1718,11 @@ pub(crate) fn tool_name_is_supported(name: &str, supported: &HashSet serde_json::Value { serde_json::json!({ @@ -2087,7 +2087,7 @@ pub fn default_tool_definitions() -> Vec { }, ToolDefinition { name: "wait_for_events".to_string(), - description: "Yield after starting child agents or other asynchronous work. Use this when you are waiting for child-agent updates instead of repeating the same investigation yourself.".to_string(), + description: "Yield only for an expected update from an already-started child agent or orchestration event source. Call this tool alone after other tool results have committed. Do not use this for background-shell completion, reading or monitoring log files, reminders, or waiting for another user message. Handle a fresh user question or status/log-review request now using available evidence and permitted inspection tools; never park that request in a wait.".to_string(), input_schema: serde_json::json!({ "type": "object", "properties": { diff --git a/app/src/ai/provider/request_translator_tests.rs b/app/src/ai/provider/request_translator_tests.rs index c542376c..40cb20fa 100644 --- a/app/src/ai/provider/request_translator_tests.rs +++ b/app/src/ai/provider/request_translator_tests.rs @@ -697,3 +697,43 @@ fn compatibility_background_acknowledgment_does_not_invent_monitoring() { }; assert_eq!(super::long_running_command_content(&snapshot), content); } + +#[test] +fn status_and_log_followups_are_not_parked_behind_event_waits() { + let request = api::Request { + settings: Some(api::request::Settings { + supported_tools: vec![ + api::ToolType::RunShellCommand.into(), + api::ToolType::ReadFiles.into(), + api::ToolType::Subagent.into(), + ], + ..Default::default() + }), + ..Default::default() + }; + let prompt = extract_system_prompt(&request, &[]).expect("system prompt"); + for instruction in [ + "Background jobs continue independently", + "Do not use wait_for_events for background commands", + "inspect the available evidence and answer now", + "read the retained log path for the requested snapshot", + "do not defer that inspection until command exit", + ] { + assert!( + prompt.contains(instruction), + "missing instruction: {instruction}" + ); + } + assert!(!super::PROGRESS_COMMUNICATION_GUIDANCE + .contains("command or subagent, use its wait tool")); + + let tools = extract_tools(&request); + let wait = tools + .iter() + .find(|tool| tool.name == "wait_for_events") + .expect("child wait tool"); + assert!(wait.description.contains("expected update from an already-started child agent or orchestration event source")); + assert!(wait.description.contains("never park that request in a wait")); + assert!(!wait.description.contains("other asynchronous work")); + assert!(tools.iter().any(|tool| tool.name == "read_files")); +}