diff --git a/app/src/ai/agent/conversation.rs b/app/src/ai/agent/conversation.rs index 46422064..cd0f78c6 100644 --- a/app/src/ai/agent/conversation.rs +++ b/app/src/ai/agent/conversation.rs @@ -2349,15 +2349,12 @@ impl AIConversation { if self.contains_action(&action.id) { return Ok(()); } - let added_exchanges = self - .added_exchanges_by_response - .get(stream_id) - .ok_or(UpdateConversationError::NoPendingRequest)?; - let exchange_id = added_exchanges - .iter() - .find(|added| added.task_id == action.task_id) - .map(|added| added.exchange_id) - .ok_or(UpdateConversationError::TaskNotFound)?; + let exchange_id = self.ensure_response_exchange_for_task( + stream_id, + &action.task_id, + terminal_surface_id, + ctx, + )?; let message_id = MessageId::new(action.id.to_string()); let exchange = self.get_exchange_to_update(exchange_id)?; match &exchange.output_status { @@ -2383,6 +2380,59 @@ impl AIConversation { Ok(()) } + fn ensure_response_exchange_for_task( + &mut self, + stream_id: &ResponseStreamId, + task_id: &TaskId, + terminal_surface_id: EntityId, + ctx: &mut ModelContext, + ) -> Result { + let added_exchanges = self + .added_exchanges_by_response + .get(stream_id) + .ok_or(UpdateConversationError::NoPendingRequest)?; + if let Some(exchange_id) = added_exchanges + .iter() + .find_map(|added| (added.task_id == *task_id).then_some(added.exchange_id)) + { + return Ok(exchange_id); + } + + // Direct-provider command monitoring can switch tasks within one response stream. A + // tool-first monitor turn needs an exchange before any message event can create it. + let source_exchange = added_exchanges.last().clone(); + let existing_exchange = self + .task_store + .get(&source_exchange.task_id) + .ok_or(UpdateConversationError::TaskNotFound)? + .exchange(source_exchange.exchange_id) + .cloned() + .ok_or(UpdateConversationError::ExchangeNotFound)?; + let mut task = self + .task_store + .remove(task_id) + .ok_or(UpdateConversationError::TaskNotFound)?; + let exchange_id = task.append_new_exchange(&existing_exchange); + self.task_store.insert(task); + self.added_exchanges_by_response + .get_mut(stream_id) + .ok_or(UpdateConversationError::NoPendingRequest)? + .push(AddedExchange { + task_id: task_id.clone(), + exchange_id, + }); + let is_hidden = self.hidden_exchanges.contains(&exchange_id); + ctx.emit(BlocklistAIHistoryEvent::AppendedExchange { + response_stream_id: Some(stream_id.clone()), + exchange_id, + task_id: task_id.clone(), + terminal_surface_id, + conversation_id: self.id, + is_hidden, + }); + Ok(exchange_id) + } + pub fn update_cost_and_usage_for_request( &mut self, request_cost: Option, diff --git a/app/src/ai/blocklist/history_model_tests.rs b/app/src/ai/blocklist/history_model_tests.rs index adc88436..c2c69a9b 100644 --- a/app/src/ai/blocklist/history_model_tests.rs +++ b/app/src/ai/blocklist/history_model_tests.rs @@ -21,10 +21,10 @@ use crate::ai::agent::conversation::{ ServerAIConversationMetadata, }; use crate::ai::agent::{ - AIAgentExchange, AIAgentExchangeId, AIAgentInput, AIAgentOutput, AIAgentOutputMessage, - AIAgentOutputMessageType, AIAgentOutputStatus, AIAgentText, AIAgentTextSection, - AgentOutputText, FinishedAIAgentOutput, MessageId, RenderableAIError, RunningCommand, Shared, - TransientNetworkErrorKind, UserQueryMode, + AIAgentAction, AIAgentActionId, AIAgentActionType, AIAgentExchange, AIAgentExchangeId, + AIAgentInput, AIAgentOutput, AIAgentOutputMessage, AIAgentOutputMessageType, + AIAgentOutputStatus, AIAgentText, AIAgentTextSection, AgentOutputText, FinishedAIAgentOutput, + MessageId, RenderableAIError, RunningCommand, Shared, TransientNetworkErrorKind, UserQueryMode, }; use crate::ai::ambient_agents::{ conversation_output_status_from_conversation, AmbientAgentTaskId, AmbientConversationStatus, @@ -223,6 +223,101 @@ fn repeated_command_steering_reuses_the_active_cli_subtask() { }); } +#[test] +fn provider_tool_proposal_creates_exchange_for_tool_first_cli_turn() { + App::test((), |mut app| async move { + initialize_history_persistence_for_tests(&mut app); + let terminal_view_id = EntityId::new(); + let history_model = app.add_singleton_model(|_| BlocklistAIHistoryModel::new_for_test()); + let stream_id = ResponseStreamId::new_for_test(); + let action_id = AIAgentActionId::from("monitor-tool-call".to_owned()); + + let (conversation_id, cli_task_id, action) = + history_model.update(&mut app, |model, ctx| { + let conversation_id = + model.start_new_conversation(terminal_view_id, false, false, false, ctx); + let root_task_id = model + .conversation(&conversation_id) + .expect("conversation should exist") + .get_root_task_id() + .clone(); + model + .update_conversation_for_new_request_input( + RequestInput { + conversation_id, + input_messages: HashMap::from([(root_task_id, Vec::new())]), + working_directory: None, + model_id: LLMId::from("test-model"), + coding_model_id: LLMId::from("test-coding-model"), + cli_agent_model_id: LLMId::from("test-cli-agent-model"), + computer_use_model_id: LLMId::from("test-computer-use-model"), + shared_session_response_initiator: None, + request_start_ts: Local::now(), + supported_tools_override: None, + }, + stream_id.clone(), + terminal_view_id, + ctx, + ) + .expect("root response exchange should be recorded"); + model.initialize_output_for_response_stream( + &stream_id, + conversation_id, + terminal_view_id, + warp_multi_agent_api::response_event::StreamInit { + request_id: "provider-request".to_owned(), + conversation_id: "provider-conversation".to_owned(), + run_id: "provider-run".to_owned(), + }, + ctx, + ); + let cli_task_id = model + .create_cli_subagent_task_for_conversation( + BlockId::new(), + conversation_id, + terminal_view_id, + ctx, + ) + .expect("CLI subtask should be created"); + let action = AIAgentAction { + id: action_id.clone(), + task_id: cli_task_id.clone(), + action: AIAgentActionType::FileGlob { + patterns: vec!["*.rs".to_owned()], + path: None, + }, + requires_result: true, + tool_name: Some("file_glob".to_owned()), + }; + model + .apply_domain_tool_proposal( + &stream_id, + conversation_id, + terminal_view_id, + action.clone(), + ctx, + ) + .expect("tool-first CLI proposal should attach to a lazy exchange"); + (conversation_id, cli_task_id, action) + }); + + history_model.read(&app, |model, _| { + let conversation = model + .conversation(&conversation_id) + .expect("conversation should exist"); + let cli_task = conversation + .get_task(&cli_task_id) + .expect("CLI subtask should exist"); + assert_eq!(cli_task.exchanges_len(), 1); + assert_eq!( + conversation.exchange_id_for_action(&action.id), + cli_task.last_exchange().map(|exchange| exchange.id) + ); + assert!(conversation.contains_action(&action.id)); + }); + }); +} + #[test] fn deactivating_cli_subtask_clears_activity_without_deleting_task() { App::test((), |mut app| async move {