diff --git a/app/src/ai/agent/api/convert_conversation.rs b/app/src/ai/agent/api/convert_conversation.rs index 4b6f9abb..837d8f4d 100644 --- a/app/src/ai/agent/api/convert_conversation.rs +++ b/app/src/ai/agent/api/convert_conversation.rs @@ -359,7 +359,13 @@ impl ConvertToExchanges for &api::Task { let task_id = TaskId::new(api_message.task_id.clone()); // Check if request_id has changed - if so, create an exchange from accumulated messages let message_request_id = if api_message.request_id.is_empty() { - None + // Older local snapshots omitted request IDs. A user query still + // gives us an unambiguous boundary; keep following output with it. + if matches!(message, api::message::Message::UserQuery(_)) { + Some(api_message.id.clone()) + } else { + current_request_id.clone() + } } else { Some(api_message.request_id.clone()) }; @@ -466,12 +472,18 @@ impl ConvertToExchanges for &api::Task { } api::message::Message::ToolCallResult(tool_call_result) => { // Try to convert tool call result - returns None for ServerToolCalls - if let Some(input) = convert_tool_call_result_to_input( + if let Some(mut input) = convert_tool_call_result_to_input( &task_id, tool_call_result, &tool_call_map, &mut document_versions, ) { + crate::ai::agent::provider_history::restore_result_details( + &mut input, + &api_message.server_message_data, + message_map.get(tool_call_result.tool_call_id.as_str()) + .map(|proposal| proposal.server_message_data.as_str()), + ); // Add tool call result as input current_inputs.push(input); } @@ -1867,15 +1879,19 @@ fn create_exchange_from_messages( _ => None, }) }) - // Fall back to any timestamp from the messages in this exchange + // Message IDs are a set, so choose the earliest timestamp rather than + // whichever message happens to be visited first. .or_else(|| { - message_ids.iter().find_map(|message_id| { - message_map.get(message_id.as_str()).and_then(|message| { - message.timestamp.as_ref().map(|timestamp| { - proto_timestamp_to_local_datetime(timestamp.seconds, timestamp.nanos) + message_ids + .iter() + .filter_map(|message_id| { + message_map.get(message_id.as_str()).and_then(|message| { + message.timestamp.as_ref().map(|timestamp| { + proto_timestamp_to_local_datetime(timestamp.seconds, timestamp.nanos) + }) }) }) - }) + .min() }) .unwrap_or_default(); diff --git a/app/src/ai/agent/api/convert_from.rs b/app/src/ai/agent/api/convert_from.rs index 1c2190e2..661ac7a9 100644 --- a/app/src/ai/agent/api/convert_from.rs +++ b/app/src/ai/agent/api/convert_from.rs @@ -215,6 +215,8 @@ fn convert_start_agent_v2_execution_mode( /// Unexpected errors when trying to convert an [`api::Message`] to an [`AIAgentOutputMessage`]. #[derive(Debug, thiserror::Error)] pub enum MessageToAIAgentOutputMessageError { + #[error("Invalid saved provider tool: {0}")] + SavedProviderTool(String), #[error("Missing expected message")] MissingMessage, #[error("Error converting tool to action: {0:?}")] @@ -262,6 +264,16 @@ impl ConvertAPIMessageToClientOutputMessage for api::Message { self, params: ConversionParams, ) -> Result { + if let Some(action) = + crate::ai::agent::provider_history::restore_tool(&self.server_message_data) + { + return Ok(MaybeAIAgentOutputMessage::Message( + AIAgentOutputMessage::action( + MessageId::new(self.id), + action.map_err(MessageToAIAgentOutputMessageError::SavedProviderTool)?, + ), + )); + } let Some(message) = self.message else { // In shared-session streams we can receive skeleton placeholder task messages without payloads. // Treat them as having no client representation rather than erroring and aborting ingestion entirely. diff --git a/app/src/ai/agent/conversation.rs b/app/src/ai/agent/conversation.rs index 3d49e451..602a03f5 100644 --- a/app/src/ai/agent/conversation.rs +++ b/app/src/ai/agent/conversation.rs @@ -2430,6 +2430,61 @@ impl AIConversation { Ok(()) } + pub(crate) fn save_provider_tool( + &mut self, + stream_id: &ResponseStreamId, + context: &crate::ai::runtime::ProviderActionContext, + call: &galaxy_agent_core::ToolCall, + action: &AIAgentAction, + ctx: &mut ModelContext, + ) -> anyhow::Result<()> { + let task_id = TaskId::new(context.task_id().to_owned()); + let exchange_id = self + .added_exchanges_by_response + .get(stream_id) + .and_then(|exchanges| { + exchanges + .iter() + .find(|exchange| exchange.task_id == task_id) + }) + .map(|exchange| exchange.exchange_id) + .ok_or_else(|| anyhow::anyhow!("missing provider tool exchange"))?; + let message = super::provider_history::tool_message(context, call, action)?; + self.task_store + .modify_task(&task_id, |task| { + task.save_display_message(message, exchange_id.to_string()) + }) + .ok_or_else(|| anyhow::anyhow!("missing provider tool task"))??; + self.write_updated_conversation_state(ctx); + Ok(()) + } + + pub(crate) fn save_provider_tool_result( + &mut self, + result: &super::AIAgentActionResult, + ctx: &mut ModelContext, + ) -> anyhow::Result<()> { + let request_id = self + .get_task(&result.task_id) + .and_then(|task| { + task.messages() + .find(|message| message.id == result.id.to_string()) + }) + .map(|message| message.request_id.clone()); + // Hosted/session-owned tools already persist their results through API messages. + let Some(request_id) = request_id else { + return Ok(()); + }; + let message = super::provider_history::result_message(result)?; + self.task_store + .modify_task(&result.task_id, |task| { + task.save_display_message(message, request_id) + }) + .ok_or_else(|| anyhow::anyhow!("missing provider result task"))??; + self.write_updated_conversation_state(ctx); + Ok(()) + } + fn ensure_response_exchange_for_task( &mut self, stream_id: &ResponseStreamId, diff --git a/app/src/ai/agent/mod.rs b/app/src/ai/agent/mod.rs index f835f36a..48b67130 100644 --- a/app/src/ai/agent/mod.rs +++ b/app/src/ai/agent/mod.rs @@ -6,6 +6,7 @@ pub(crate) mod api; pub(crate) mod comment; pub(crate) mod icons; pub(crate) mod linearization; +pub(crate) mod provider_history; pub(crate) mod redaction; pub(crate) mod runtime_activity; pub(crate) mod task; diff --git a/app/src/ai/agent/provider_history.rs b/app/src/ai/agent/provider_history.rs new file mode 100644 index 00000000..4371903e --- /dev/null +++ b/app/src/ai/agent/provider_history.rs @@ -0,0 +1,239 @@ +//! Durable display history for direct providers, independent of their compacted model transcript. +use galaxy_agent_core::ToolCall; +use serde::{Deserialize, Serialize}; +use warp_multi_agent_api as api; + +use super::{AIAgentAction, AIAgentActionResult}; +use crate::ai::runtime::ProviderActionContext; + +const TOOL_PREFIX: &str = "galaxy:provider-tool:v1:"; + +#[derive(Serialize, Deserialize)] +struct SavedTool { + context: ProviderActionContext, + call: ToolCall, + #[serde(default)] + delegated_base_prompt: Option, + #[serde(default)] + question_ids: Vec, +} + +pub(crate) fn tool_message( + context: &ProviderActionContext, + call: &ToolCall, + action: &AIAgentAction, +) -> anyhow::Result { + // Keep the native call as well as the compatibility representation. The native + // form retains MCP aliases and distinctions such as notebook vs. plan tools. + let compatible = context.history_compatibility_call(call); + let event = crate::ai::provider::response_translator::build_tool_call_message( + context.task_id(), + &call.id, + &compatible.name, + &compatible.arguments.to_string(), + ); + let mut message = match event.r#type { + Some(api::response_event::Type::ClientActions(actions)) => actions + .actions + .into_iter() + .find_map(|action| match action.action { + Some(api::client_action::Action::AddMessagesToTask(messages)) => { + messages.messages.into_iter().next() + } + _ => None, + }) + .ok_or_else(|| anyhow::anyhow!("missing tool history message"))?, + _ => anyhow::bail!("missing tool history event"), + }; + // Native action IDs must remain identical across execution and restoration. + message.id = call.id.clone(); + if let Some(api::message::Message::ToolCall(tool)) = &mut message.message { + tool.tool_call_id = call.id.clone(); + } else { + message.message = Some(api::message::Message::ToolCall(api::message::ToolCall { + tool_call_id: call.id.clone(), + tool: None, + })); + } + message.server_message_data = format!( + "{TOOL_PREFIX}{}", + serde_json::to_string(&SavedTool { + context: context.clone(), + call: call.clone(), + question_ids: match &action.action { + super::AIAgentActionType::AskUserQuestion { questions } => questions + .iter() + .map(|question| question.question_id.clone()) + .collect(), + _ => Vec::new(), + }, + delegated_base_prompt: match &action.action { + super::AIAgentActionType::RunAgents(request) => Some(request.base_prompt.clone()), + _ => None, + }, + })? + ); + Ok(message) +} + +pub(crate) fn restore_tool(data: &str) -> Option> { + let json = data.strip_prefix(TOOL_PREFIX)?; + Some( + serde_json::from_str::(json) + .map_err(|error| error.to_string()) + .and_then(|saved| { + let mut action = saved.context.restored_action_from_tool_call(&saved.call)?; + if let (super::AIAgentActionType::RunAgents(request), Some(base_prompt)) = + (&mut action.action, saved.delegated_base_prompt) + { + request.base_prompt = base_prompt; + } + if let super::AIAgentActionType::AskUserQuestion { questions } = &mut action.action + { + for (question, id) in questions.iter_mut().zip(saved.question_ids) { + question.question_id = id; + } + } + Ok(action) + }), + ) +} + +const RESULT_PREFIX: &str = "galaxy:provider-result:v1:"; + +#[derive(Default, Serialize, Deserialize)] +struct SavedResultDetails { + cancelled: bool, + file_edit: Option<(String, usize, usize)>, + command_error: Option<(String, String)>, +} + +/// Supplement fields that the server's request/result protocol does not round-trip. +/// This is display metadata and is never used to execute an action. +pub(crate) fn restore_result_details( + input: &mut super::AIAgentInput, + data: &str, + proposal_data: Option<&str>, +) { + let Some(json) = data.strip_prefix(RESULT_PREFIX) else { + return; + }; + let Ok(details) = serde_json::from_str::(json) else { + return; + }; + let super::AIAgentInput::ActionResult { result, .. } = input else { + return; + }; + if details.cancelled { + if let Some(Ok(action)) = proposal_data.and_then(restore_tool) { + result.result = action.action.cancelled_result(); + } + } + if let ( + super::AIAgentActionResultType::RequestFileEdits(super::RequestFileEditsResult::Success { + diff, + lines_added, + lines_removed, + .. + }), + Some((saved_diff, added, removed)), + ) = (&mut result.result, details.file_edit) + { + *diff = saved_diff; + *lines_added = added; + *lines_removed = removed; + } + if let Some((command, message)) = details.command_error { + result.result = super::AIAgentActionResultType::RequestCommandOutput( + super::RequestCommandOutputResult::ExecutionError { command, message }, + ); + } +} + +// FileGlob remains part of saved history even though the server deprecated it. +#[allow(deprecated)] +pub(crate) fn result_message(result: &AIAgentActionResult) -> anyhow::Result { + use api::message::tool_call_result::Result as Output; + use api::request::input::tool_call_result::Result as Input; + let details = SavedResultDetails { + cancelled: result.result.is_cancelled(), + file_edit: match &result.result { + super::AIAgentActionResultType::RequestFileEdits( + super::RequestFileEditsResult::Success { + diff, + lines_added, + lines_removed, + .. + }, + ) => Some((diff.clone(), *lines_added, *lines_removed)), + _ => None, + }, + command_error: match &result.result { + super::AIAgentActionResultType::RequestCommandOutput( + super::RequestCommandOutputResult::ExecutionError { command, message }, + ) => Some((command.clone(), message.clone())), + _ => None, + }, + }; + let input: api::request::input::user_inputs::user_input::Input = result.clone().try_into()?; + let api::request::input::user_inputs::user_input::Input::ToolCallResult(input) = input else { + anyhow::bail!("expected tool result"); + }; + // These protobuf oneofs use different field numbers; map variants explicitly. + let payload = input.result.map(|result| match result { + Input::RunShellCommand(value) => Output::RunShellCommand(value), + Input::ReadFiles(value) => Output::ReadFiles(value), + Input::SearchCodebase(value) => Output::SearchCodebase(value), + Input::ApplyFileDiffs(value) => Output::ApplyFileDiffs(value), + Input::SuggestPlan(value) => Output::SuggestPlan(value), + Input::SuggestCreatePlan(value) => Output::SuggestCreatePlan(value), + Input::Grep(value) => Output::Grep(value), + Input::FileGlob(value) => Output::FileGlob(value), + Input::ReadMcpResource(value) => Output::ReadMcpResource(value), + Input::CallMcpTool(value) => Output::CallMcpTool(value), + Input::WriteToLongRunningShellCommand(value) => { + Output::WriteToLongRunningShellCommand(value) + } + Input::SuggestNewConversation(value) => Output::SuggestNewConversation(value), + Input::FileGlobV2(value) => Output::FileGlobV2(value), + Input::SuggestPrompt(value) => Output::SuggestPrompt(value), + Input::OpenCodeReview(value) => Output::OpenCodeReview(value), + Input::InitProject(value) => Output::InitProject(value), + Input::ReadDocuments(value) => Output::ReadDocuments(value), + Input::EditDocuments(value) => Output::EditDocuments(value), + Input::CreateDocuments(value) => Output::CreateDocuments(value), + Input::ReadShellCommandOutput(value) => Output::ReadShellCommandOutput(value), + Input::UseComputer(value) => Output::UseComputer(value), + Input::InsertReviewComments(value) => Output::InsertReviewComments(value), + Input::RequestComputerUse(value) => Output::RequestComputerUseResult(value), + Input::ReadSkill(value) => Output::ReadSkill(value), + Input::FetchConversation(value) => Output::FetchConversation(value), + Input::StartAgent(value) => Output::StartAgent(value), + Input::SendMessageToAgent(value) => Output::SendMessageToAgent(value), + Input::TransferShellCommandControlToUser(value) => { + Output::TransferShellCommandControlToUser(value) + } + Input::AskUserQuestion(value) => Output::AskUserQuestion(value), + Input::StartAgentV2(value) => Output::StartAgentV2(value), + Input::UploadFileArtifact(value) => Output::UploadFileArtifact(value), + Input::RunAgentsResult(value) => Output::RunAgentsResult(value), + Input::WaitForEvents(value) => Output::WaitForEvents(value), + }); + Ok(api::Message { + id: format!("galaxy-tool-result:{}", result.id), + server_message_data: format!("{RESULT_PREFIX}{}", serde_json::to_string(&details)?), + task_id: result.task_id.to_string(), + message: Some(api::message::Message::ToolCallResult( + api::message::ToolCallResult { + tool_call_id: input.tool_call_id, + context: None, + result: payload, + }, + )), + ..Default::default() + }) +} + +#[cfg(test)] +#[path = "provider_history_tests.rs"] +mod tests; diff --git a/app/src/ai/agent/provider_history_tests.rs b/app/src/ai/agent/provider_history_tests.rs new file mode 100644 index 00000000..0963bd74 --- /dev/null +++ b/app/src/ai/agent/provider_history_tests.rs @@ -0,0 +1,186 @@ +use galaxy_agent_core::ToolCall; +use prost::Message; +use serde_json::json; +use warp_multi_agent_api as api; + +use super::{result_message, tool_message}; +use crate::ai::agent::api::convert_conversation::ConvertToExchanges; +use crate::ai::agent::{ + AIAgentActionResult, AIAgentActionResultType, AIAgentInput, GrepResult, + RequestCommandOutputResult, RunAgentsAgentOutcome, RunAgentsAgentOutcomeKind, + RunAgentsLaunchedExecutionMode, RunAgentsResult, +}; +use crate::ai::runtime::ProviderActionContext; + +fn query(id: &str) -> api::Message { + api::Message { + id: id.to_owned(), + task_id: "task".into(), + message: Some(api::message::Message::UserQuery(api::message::UserQuery { + query: id.to_owned(), + ..Default::default() + })), + ..Default::default() + } +} + +#[test] +fn saved_tools_and_results_restore_as_individual_panes_across_turns() { + let context = ProviderActionContext::new_for_test("task"); + let cases = [ + ( + ToolCall { + id: "edit-call".into(), + name: "apply_file_diffs".into(), + arguments: json!({"summary": "Update file", "diffs": [{"file_path": "/project/file", "search": "before", "replace": "after"}]}), + }, + AIAgentActionResultType::RequestFileEdits( + crate::ai::agent::RequestFileEditsResult::Success { + diff: "-before\n+after".into(), + updated_files: Vec::new(), + deleted_files: Vec::new(), + lines_added: 1, + lines_removed: 1, + }, + ), + ), + ( + ToolCall { + id: "cancelled-grep".into(), + name: "grep".into(), + arguments: json!({"queries": ["needle"], "path": "/project"}), + }, + AIAgentActionResultType::Grep(GrepResult::Cancelled), + ), + ( + ToolCall { + id: "shell-call".into(), + name: "run_shell_command".into(), + arguments: json!({"command": "cat large.txt"}), + }, + AIAgentActionResultType::RequestCommandOutput(RequestCommandOutputResult::Completed { + block_id: "shell-block".to_owned().into(), + command: "cat large.txt".into(), + output: "complete output line\n".repeat(10_000), + exit_code: galaxy_core::command::ExitCode::from(0), + start_ts: None, + completed_ts: None, + }), + ), + ( + ToolCall { + id: "grep-call".into(), + name: "grep".into(), + arguments: json!({"queries": ["needle"], "path": "/project"}), + }, + AIAgentActionResultType::Grep(GrepResult::Error("path unavailable".into())), + ), + ( + ToolCall { + id: "agents-call".into(), + name: "run_agents".into(), + arguments: json!({ + "summary": "Implement and review", "execution_mode": {"type": "local"}, + "agent_run_configs": [{"name": "reviewer", "prompt": "Review changes", "model_id": "review-model"}] + }), + }, + AIAgentActionResultType::RunAgents(RunAgentsResult::Launched { + model_id: "review-model".into(), + harness_type: String::new(), + execution_mode: RunAgentsLaunchedExecutionMode::Local, + agents: vec![RunAgentsAgentOutcome { + name: "reviewer".into(), + kind: RunAgentsAgentOutcomeKind::Launched { + agent_id: "child-conversation".into(), + }, + }], + }), + ), + ]; + let mut task = api::Task { + id: "task".into(), + ..Default::default() + }; + for (index, (call, result)) in cases.iter().enumerate() { + let request_id = format!("turn-{index}"); + let action_result = AIAgentActionResult { + id: call.id.clone().into(), + task_id: crate::ai::agent::task::TaskId::new("task".to_owned()), + result: result.clone(), + }; + for mut message in [ + query(&request_id), + tool_message( + &context, + call, + &context.action_from_tool_call(call).unwrap(), + ) + .unwrap(), + result_message(&action_result).unwrap(), + ] { + message.request_id = request_id.clone(); + task.messages.push(message); + } + } + // Exercise the same protobuf bytes stored in agent_tasks, followed by UI hydration. + let restored = api::Task::decode(task.encode_to_vec().as_slice()).unwrap(); + let exchanges = (&restored).into_exchanges(); + assert_eq!(exchanges.len(), cases.len()); + for (exchange, (call, expected_result)) in exchanges.iter().zip(cases.iter()) { + let output = exchange.output_status.output().unwrap(); + let output = output.get(); + let actions = output.actions().collect::>(); + assert_eq!(actions.len(), 1); + assert_eq!(actions[0], &context.action_from_tool_call(call).unwrap()); + let result = exchange + .input + .iter() + .find_map(|input| match input { + AIAgentInput::ActionResult { result, .. } => Some(result), + _ => None, + }) + .unwrap(); + assert_eq!(&result.result, expected_result); + } +} + +#[test] +fn legacy_messages_without_request_ids_split_at_each_user_query() { + let task = api::Task { + id: "task".into(), + messages: vec![ + query("first"), + api::Message { + id: "reply".into(), + task_id: "task".into(), + message: Some(api::message::Message::AgentOutput( + api::message::AgentOutput { + text: "First reply".into(), + }, + )), + ..Default::default() + }, + query("second"), + ], + ..Default::default() + }; + assert_eq!((&task).into_exchanges().len(), 2); +} + +#[test] +fn restored_question_keeps_the_id_used_by_its_answer() { + let context = ProviderActionContext::new_for_test("task"); + let call = ToolCall { + id: "question-call".into(), + name: "ask_user_question".into(), + arguments: json!({"question": "Which option?", "options": ["A", "B"]}), + }; + let action = context.action_from_tool_call(&call).unwrap(); + let message = tool_message(&context, &call, &action).unwrap(); + assert_eq!( + super::restore_tool(&message.server_message_data) + .unwrap() + .unwrap(), + action + ); +} diff --git a/app/src/ai/agent/task.rs b/app/src/ai/agent/task.rs index c401e094..e774794c 100644 --- a/app/src/ai/agent/task.rs +++ b/app/src/ai/agent/task.rs @@ -691,9 +691,38 @@ impl Task { Ok(source) } + /// Append display-only records without replaying tools or changing the live exchange. + pub(super) fn save_display_message( + &mut self, + mut message: api::Message, + request_id: String, + ) -> Result<(), UpdateTaskError> { + message.request_id = request_id; + if message.timestamp.is_none() { + let now = chrono::Utc::now(); + message.timestamp = Some(prost_types::Timestamp { + seconds: now.timestamp(), + nanos: now.timestamp_subsec_nanos() as i32, + }); + } + let source = self.try_get_source_mut()?; + if let Some(existing) = source + .messages + .iter_mut() + .find(|existing| existing.id == message.id) + { + message.request_id.clone_from(&existing.request_id); + message.timestamp = existing.timestamp.or(message.timestamp); + *existing = message; + } else { + source.messages.push(message); + } + Ok(()) + } + pub(super) fn add_messages( &mut self, - messages: Vec, + mut messages: Vec, exchange_id: AIAgentExchangeId, message_context: TaskMessageContext<'_>, should_convert_input_messages: bool, @@ -701,6 +730,20 @@ impl Task { if self.source().is_none() { return Err(UpdateTaskError::TaskNotInitialized); } + // Locally projected provider messages have no server request identity. + // Persist their actual exchange identity so restart preserves turn boundaries. + for message in &mut messages { + if message.request_id.is_empty() { + message.request_id = exchange_id.to_string(); + } + if message.timestamp.is_none() { + let now = chrono::Utc::now(); + message.timestamp = Some(prost_types::Timestamp { + seconds: now.timestamp(), + nanos: now.timestamp_subsec_nanos() as i32, + }); + } + } self.update_exchange_from_messages( messages.clone(), exchange_id, diff --git a/app/src/ai/blocklist/action_model.rs b/app/src/ai/blocklist/action_model.rs index 8706b44f..77daa9c2 100644 --- a/app/src/ai/blocklist/action_model.rs +++ b/app/src/ai/blocklist/action_model.rs @@ -2455,6 +2455,16 @@ impl BlocklistAIActionModel { let permission_denied = self .denied_permissions .remove(&(conversation_id, action_result.id.clone())); + if execution_ref.is_some() { + BlocklistAIHistoryModel::handle(ctx).update(ctx, |history, ctx| { + if let Some(conversation) = history.conversation_mut(&conversation_id) { + if let Err(error) = conversation.save_provider_tool_result(&action_result, ctx) + { + log::error!("Failed to save provider tool result: {error}"); + } + } + }); + } let tool_result = domain_tool_result(&action_result, permission_denied); if execution_ref.is_none() { self.finished_tool_results diff --git a/app/src/ai/blocklist/action_model/execute/shell_command.rs b/app/src/ai/blocklist/action_model/execute/shell_command.rs index 1a312726..40b6456f 100644 --- a/app/src/ai/blocklist/action_model/execute/shell_command.rs +++ b/app/src/ai/blocklist/action_model/execute/shell_command.rs @@ -277,8 +277,16 @@ impl ShellCommandExecutor { move |result, ctx| { // Remove the senders from the maps. if let Some(handle) = handle.upgrade(ctx) { - handle.update(ctx, |me, _| { + handle.update(ctx, |me, ctx| { me.prune_closed_senders(&block_selector); + if matches!( + result, + ActionResult::BlockNotFound | ActionResult::Cancelled + ) { + ctx.emit(ShellCommandExecutorEvent::CancelExecution { + action_id: action_id.clone(), + }); + } }); } @@ -659,11 +667,9 @@ impl ShellCommandExecutor { }; // Cancelling the wait future alone would report cancellation while the process keeps // running. Terminate the exact requested command before resolving the action as cancelled. - if requested_block_is_running { - ctx.emit(ShellCommandExecutorEvent::CancelExecution { - action_id: id.clone(), - }); - } + ctx.emit(ShellCommandExecutorEvent::CancelExecution { + action_id: id.clone(), + }); if !requested_block_is_running { self.block_finished_senders.remove(&selector); self.force_refresh_senders.remove(&selector); diff --git a/app/src/ai/blocklist/block/cli_controller.rs b/app/src/ai/blocklist/block/cli_controller.rs index 4c5d551d..e071bc91 100644 --- a/app/src/ai/blocklist/block/cli_controller.rs +++ b/app/src/ai/blocklist/block/cli_controller.rs @@ -295,6 +295,20 @@ impl CLISubagentController { }); ctx.subscribe_to_model(model_event_dispatcher, |me, _, event, ctx| { + if let ModelEvent::AfterBlockStarted { block_id, .. } = event { + let identity = { + let model = me.terminal_model.lock(); + model.block_list().block_with_id(block_id).and_then(|block| { + Some(( + block.ai_conversation_id()?, + block.requested_command_action_id()?.clone(), + )) + }) + }; + if let Some((conversation_id, action_id)) = identity { + me.track_requested_command(block_id, conversation_id, &action_id); + } + } if let ModelEvent::BlockCompleted(block_completed_event) = event { let terminal_model = me.terminal_model.lock(); let Some(block) = terminal_model @@ -654,11 +668,10 @@ impl CLISubagentController { .and_then(|state| state.last_snapshot_at) } - /// Begins tracking an agent-requested command before its shell event is dispatched. - /// - /// The placeholder lets command completion and action-result events arrive in either order - /// without losing the completion that a subsequently-created CLI monitor needs. - pub fn track_requested_command( + /// Tracks the block assigned when an agent-requested command actually starts in the shell. + /// This must not use the active block from when the command was queued: that block may still + /// belong to a previous command. The placeholder preserves completion before a monitor exists. + fn track_requested_command( &mut self, block_id: &BlockId, conversation_id: AIConversationId, diff --git a/app/src/ai/blocklist/controller.rs b/app/src/ai/blocklist/controller.rs index b6b4c37f..7d4a522d 100644 --- a/app/src/ai/blocklist/controller.rs +++ b/app/src/ai/blocklist/controller.rs @@ -2558,7 +2558,23 @@ impl BlocklistAIController { terminal_surface_id, conversation_ids, } if *terminal_surface_id == me.terminal_surface_id => { - me.schedule_restored_provider_runs(conversation_ids, ctx); + BlocklistAIHistoryModel::handle(ctx).update(ctx, |history, ctx| { + for conversation_id in conversation_ids { + if history + .conversation(conversation_id) + .is_some_and(|conversation| { + conversation.active_provider_run_json().is_some() + }) + { + history.update_conversation_status( + me.terminal_surface_id, + *conversation_id, + ConversationStatus::Cancelled, + ctx, + ); + } + } + }); } BlocklistAIHistoryEvent::UpdatedConversationStatus { new_status, .. } if new_status.is_done() => @@ -4124,6 +4140,13 @@ impl BlocklistAIController { ); return false; }; + // A saved checkpoint is dormant until the user explicitly resumes. In particular, + // child completion or late terminal events must not wake a restored conversation. + if conversation.active_provider_run_json().is_some() + && !self.active_provider_runs.contains_key(&conversation_id) + { + return false; + } // WaitingForEvents is treated as Success here: pending events // drain via the next outbound request and the server-side // supersede emits the resume signal. @@ -4463,6 +4486,12 @@ impl BlocklistAIController { log::error!("Tried to resume non-existent conversation: {conversation_id:?}"); return; }; + if conversation.active_provider_run_json().is_some() + && !self.active_provider_runs.contains_key(&conversation_id) + { + self.schedule_restored_provider_runs(&[conversation_id], ctx); + return; + } let task_id = { let terminal_model = self.terminal_model.lock(); let active_block = terminal_model.block_list().active_block(); @@ -5573,7 +5602,8 @@ impl BlocklistAIController { self.restoring_provider_runs .extend(conversation_ids.iter().copied()); - // RestoredConversations is emitted before terminal views finish rebuilding their blocks. + // Resume only after the caller releases its view/model borrows. Merely loading a + // conversation never reaches this path; the user must explicitly request Resume. let _ = ctx.spawn(async {}, move |me, _, ctx| { for conversation_id in conversation_ids { me.restore_active_provider_run(conversation_id, ctx); @@ -7664,6 +7694,7 @@ impl BlocklistAIController { else { return; }; + let action_context = run.action_context.clone(); let tool_titles = batch .calls .iter() @@ -7823,6 +7854,32 @@ impl BlocklistAIController { return; } } + if let Some(pending) = batch + .calls + .iter() + .find(|pending| pending.call.id == action.id.to_string()) + { + let saved = BlocklistAIHistoryModel::handle(ctx).update(ctx, |history, ctx| { + let conversation = history + .conversation_mut(&conversation_id) + .ok_or_else(|| anyhow::anyhow!("missing provider conversation"))?; + conversation.save_provider_tool( + &stream_id, + &action_context, + &pending.call, + &action, + ctx, + ) + }); + if let Err(error) = saved { + self.fail_active_provider_run( + conversation_id, + format!("failed to save provider tool history: {error}"), + ctx, + ); + return; + } + } actions.push(action); } if let Some(slot) = self.active_provider_runs.get_mut(&conversation_id) { diff --git a/app/src/ai/blocklist/controller_tests.rs b/app/src/ai/blocklist/controller_tests.rs index 600dfff9..d70bff37 100644 --- a/app/src/ai/blocklist/controller_tests.rs +++ b/app/src/ai/blocklist/controller_tests.rs @@ -3890,3 +3890,57 @@ fn failed_child_without_an_exchange_reaches_terminal_error() { }); }); } + +#[test] +fn restored_provider_checkpoint_waits_for_explicit_resume() { + App::test((), |mut app| async move { + initialize_app_for_terminal_view(&mut app); + let terminal = add_window_with_terminal(&mut app, None); + let conversation_id = terminal.update(&mut app, |terminal, ctx| { + BlocklistAIHistoryModel::handle(ctx).update(ctx, |history, ctx| { + let conversation_id = + history.start_new_conversation(terminal.id(), false, false, false, ctx); + history + .persist_active_provider_run_json( + conversation_id, + Some("unparsed saved checkpoint".to_owned()), + ctx, + ) + .unwrap(); + ctx.emit( + crate::ai::blocklist::BlocklistAIHistoryEvent::RestoredConversations { + terminal_surface_id: terminal.id(), + conversation_ids: vec![conversation_id], + }, + ); + conversation_id + }) + }); + terminal.update(&mut app, |terminal, ctx| { + let conversation = BlocklistAIHistoryModel::as_ref(ctx) + .conversation(&conversation_id) + .unwrap(); + assert_eq!(conversation.status(), &ConversationStatus::Cancelled); + assert_eq!( + conversation.active_provider_run_json(), + Some("unparsed saved checkpoint") + ); + terminal.ai_controller().update(ctx, |controller, ctx| { + assert!(!controller + .restoring_provider_runs + .contains(&conversation_id)); + assert!(!controller + .active_provider_runs + .contains_key(&conversation_id)); + assert!(!controller.conversation_ready_for_pending_events(conversation_id, ctx)); + controller.resume_conversation(conversation_id, Vec::new(), ctx); + assert!(controller + .restoring_provider_runs + .contains(&conversation_id)); + // A second click must not schedule another restoration. + controller.resume_conversation(conversation_id, Vec::new(), ctx); + assert_eq!(controller.restoring_provider_runs.len(), 1); + }); + }); + }); +} diff --git a/app/src/ai/runtime/rig.rs b/app/src/ai/runtime/rig.rs index 153c2fba..c524c8bd 100644 --- a/app/src/ai/runtime/rig.rs +++ b/app/src/ai/runtime/rig.rs @@ -83,6 +83,33 @@ impl ProviderActionContext { .filter(|title| !title.is_empty()) } + pub(crate) fn history_compatibility_call(&self, call: &ToolCall) -> ToolCall { + let mut compatible = call.clone(); + if let Some(target) = self.mcp_tool_aliases.get(&call.name) { + let server = target + .server_id + .map(|id| id.to_string()) + .unwrap_or_default(); + compatible.name = format!("mcp__{server}__{}", target.name); + if let (Some(arguments), Some(parameter)) = ( + compatible.arguments.as_object_mut(), + target.title_parameter.as_ref(), + ) { + arguments.remove(parameter); + } + } + compatible + } + + pub(crate) fn restored_action_from_tool_call( + &self, + call: &ToolCall, + ) -> Result { + let mut context = self.clone(); + context.skill_path_origin = ai::skills::SkillPathOrigin::RestoredDisplayOnly; + context.action_from_tool_call(call) + } + pub(crate) fn action_from_tool_call(&self, call: &ToolCall) -> Result { action_from_tool_call( &self.task_id, diff --git a/app/src/terminal/input_tests.rs b/app/src/terminal/input_tests.rs index c486a2bf..dbe2fc27 100644 --- a/app/src/terminal/input_tests.rs +++ b/app/src/terminal/input_tests.rs @@ -2457,6 +2457,7 @@ fn test_histignorespace_support_in_zsh() { (terminal.model.clone(), terminal.sessions_model().clone()) }); + let block_id = model.lock().active_block_id().clone(); app.update(|ctx| { update_command_history( &ExecuteCommandEvent { @@ -2467,6 +2468,7 @@ fn test_histignorespace_support_in_zsh() { should_add_command_to_history: true, source: CommandExecutionSource::User, }, + &block_id, &model, None, &sessions, @@ -2482,6 +2484,7 @@ fn test_histignorespace_support_in_zsh() { should_add_command_to_history: true, source: CommandExecutionSource::User, }, + &block_id, &model, None, &sessions, diff --git a/app/src/terminal/model/terminal_model.rs b/app/src/terminal/model/terminal_model.rs index 1b8b828e..5d809c71 100644 --- a/app/src/terminal/model/terminal_model.rs +++ b/app/src/terminal/model/terminal_model.rs @@ -3022,7 +3022,10 @@ impl ansi::Handler for TerminalModel { self.is_receiving_in_band_command_output = IsReceivingInBandCommandOutput::No; } IsReceivingInBandCommandOutput::No => { - log::warn!("Received 'end_in_band_command_output' while not expecting to read in-band command output."); + // Command dispatch calls this defensively even when no in-band output is active. + if from_osc_sequence { + log::warn!("Received 'end_in_band_command_output' while not expecting to read in-band command output."); + } } } diff --git a/app/src/terminal/view.rs b/app/src/terminal/view.rs index f3b57b7f..b8adae4d 100644 --- a/app/src/terminal/view.rs +++ b/app/src/terminal/view.rs @@ -1763,6 +1763,7 @@ pub enum Event { size_update: SizeUpdate, }, ExecuteCommand(ExecuteCommandEvent), + CancelQueuedAgentCommand(AIAgentActionId), BlockStarted { is_for_in_band_command: bool, }, @@ -7695,17 +7696,8 @@ impl TerminalView { } } } - let block_id = model.active_block_id().clone(); drop(model); - self.cli_subagent_controller.update(ctx, |controller, _| { - controller.track_requested_command( - &block_id, - parent_conversation_id, - action_id, - ); - }); - ctx.emit(Event::ExecuteCommand(ExecuteCommandEvent { command, session_id, @@ -7721,8 +7713,6 @@ impl TerminalView { }); } - self.schedule_command_monitor_start(block_id, ctx); - if let Some(metadata) = workflow_telem_metadata { send_telemetry_from_ctx!(TelemetryEvent::WorkflowExecuted(metadata), ctx); } @@ -7732,6 +7722,7 @@ impl TerminalView { self.write_agent_bytes_to_pty(input.to_vec(), mode, ctx); } ShellCommandExecutorEvent::CancelExecution { action_id } => { + ctx.emit(Event::CancelQueuedAgentCommand(action_id.clone())); // We need to manually invoke ctrl-c to terminate the running command because the // user's ctrl-c was directed to the AIBlock instead of the command's shell block. let is_exact_active_command = self @@ -12185,6 +12176,15 @@ impl TerminalView { return; } self.did_notify_long_running = false; + let is_agent_command = self + .model + .lock() + .block_list() + .block_with_id(block_id) + .is_some_and(|block| block.requested_command_action_id().is_some()); + if is_agent_command { + self.schedule_command_monitor_start(block_id.clone(), ctx); + } // Snapshot the prompt state as of when the command began executing. // Commands may themselves affect the prompt (if running `git checkout`), for @@ -26257,6 +26257,9 @@ impl PtyIntentEvent for Event { }), Event::Resize { size_update } => Some(PtyIntent::Resize(*size_update)), Event::ExecuteCommand(event) => Some(PtyIntent::ExecuteCommand(event.clone())), + Event::CancelQueuedAgentCommand(action_id) => { + Some(PtyIntent::CancelQueuedAgentCommand(action_id.clone())) + } Event::RunNativeShellCompletions { buffer_text, results_tx, diff --git a/app/src/terminal/writeable_pty/command_history.rs b/app/src/terminal/writeable_pty/command_history.rs index da3e96e0..8208f261 100644 --- a/app/src/terminal/writeable_pty/command_history.rs +++ b/app/src/terminal/writeable_pty/command_history.rs @@ -5,19 +5,23 @@ use galaxyui::{AppContext, ModelHandle, SingletonEntity}; use parking_lot::FairMutex; use crate::persistence::{ModelEvent, StartedCommandMetadata}; +use crate::terminal::model::block::BlockId; use crate::terminal::model::session::Sessions; use crate::terminal::view::ExecuteCommandEvent; use crate::terminal::{History, HistoryEntry, TerminalModel}; pub fn update_command_history( event: &ExecuteCommandEvent, + block_id: &BlockId, model: &Arc>, model_event_sender: Option<&SyncSender>, sessions: &ModelHandle, ctx: &mut AppContext, ) { let model = model.lock(); - let active_block = model.block_list().active_block(); + let Some(active_block) = model.block_list().block_with_id(block_id) else { + return; + }; let session_id = event.session_id; let Some(session) = sessions.as_ref(ctx).get(session_id) else { return; diff --git a/app/src/terminal/writeable_pty/pty_controller.rs b/app/src/terminal/writeable_pty/pty_controller.rs index 620aa527..9627355b 100644 --- a/app/src/terminal/writeable_pty/pty_controller.rs +++ b/app/src/terminal/writeable_pty/pty_controller.rs @@ -10,10 +10,11 @@ use parking_lot::FairMutex; use thiserror::Error; use super::Message; -use crate::ai::agent::AIAgentPtyWriteMode; +use crate::ai::agent::{AIAgentActionId, AIAgentPtyWriteMode}; use crate::terminal::input::CommandExecutionSource; use crate::terminal::line_editor_status::{LineEditorStatus, LineEditorStatusEvent}; use crate::terminal::model::ansi::Handler; +use crate::terminal::model::block::BlockId; use crate::terminal::model::completions::ShellCompletion; use crate::terminal::model::escape_sequences; use crate::terminal::model::session::{ @@ -21,7 +22,7 @@ use crate::terminal::model::session::{ }; use crate::terminal::model_events::{AnsiHandlerEvent, ModelEvent, ModelEventDispatcher}; use crate::terminal::shell::ShellType; -use crate::terminal::view::LINEFEED_REGEX; +use crate::terminal::view::{ExecuteCommandEvent, LINEFEED_REGEX}; #[cfg(not(target_family = "wasm"))] use crate::terminal::writeable_pty::bootstrap_file::{permanent_bootstrap_file, TempBootstrapFile}; use crate::terminal::{bootstrap, SizeUpdate, TerminalModel}; @@ -46,6 +47,7 @@ enum PtyWrite { in_band_command_id: Option, /// If 'some', the given callback is called right before the bytes are written to the PTY. before_write_fn: Option>, + execution: Option>, }, Bytes { /// The bytes to be written. @@ -298,7 +300,17 @@ impl PtyController { cancel_tx: Sender, ctx: &mut ModelContext, ) { - if self.is_user_command_executing { + if self.is_user_command_executing + || self.pending_writes.iter().any(|write| { + matches!( + write, + PtyWrite::Command { + execution: Some(_), + .. + } + ) + }) + { // Send blocking should be okay b/c this is an unbound channel if let Err(err) = block_on(cancel_tx.send(InBandCommandCancelledEvent { command_id })) { log::warn!("Pty Controller failed to cancel in band command: {err:?}"); @@ -311,6 +323,7 @@ impl PtyController { command: command.to_owned(), shell_type, in_band_command_id: Some(command_id), + execution: None, before_write_fn: Some(Box::new(move || { let mut terminal_model = terminal_model.lock(); terminal_model @@ -510,47 +523,19 @@ impl PtyController { /// sequences. pub fn write_command( &mut self, - command: &str, + event: &ExecuteCommandEvent, shell_type: ShellType, - source: CommandExecutionSource, ctx: &mut ModelContext, ) { - { - let mut model = self.terminal_model.lock(); - - // Explicitly start the block now that the command is executed. - match source { - CommandExecutionSource::AI { metadata } => { - model.start_command_execution_with_ai_metadata(metadata) - } - CommandExecutionSource::SharedSession { - participant_id, - ai_metadata, - .. - } => model.start_command_execution_for_shared_session(participant_id, ai_metadata), - CommandExecutionSource::User | CommandExecutionSource::QueuedCommand => { - model.start_command_execution() - } - CommandExecutionSource::EnvVarCollection { metadata } => { - model.start_command_execution_from_env_var_collection(metadata) - } - } - - // Ensure that the `TerminalModel` doesn't interpret any of the PTY output from the - // following commands as in-band command output. If the in-band command output is not - // currently being received by the `TerminalModel`, this is a no-op. - model.end_in_band_command_output(false); - } - self.pending_writes.clear(); - self.is_user_command_executing = true; // Send the write to the PTY event loop. let write = PtyWrite::Command { - command: command.to_owned(), + command: event.command.clone(), shell_type, in_band_command_id: None, before_write_fn: None, + execution: Some(Box::new(event.clone())), }; if self.can_write_to_pty(ctx) { // Cancel the async writer task and clear the async write queue. @@ -561,6 +546,28 @@ impl PtyController { } } + /// Cancel only a command that has not yet been dispatched to the shell. + pub fn cancel_queued_agent_command(&mut self, action_id: &AIAgentActionId) { + self.pending_writes.retain(|write| { + let PtyWrite::Command { + execution: Some(event), + .. + } = write + else { + return true; + }; + let metadata = match &event.source { + CommandExecutionSource::AI { metadata } => Some(metadata), + CommandExecutionSource::SharedSession { ai_metadata, .. } => ai_metadata.as_ref(), + CommandExecutionSource::User + | CommandExecutionSource::QueuedCommand + | CommandExecutionSource::EnvVarCollection { .. } => None, + }; + !metadata + .is_some_and(|metadata| metadata.requested_command_action_id() == Some(action_id)) + }); + } + /// Synchronously writes the EOT (End-of-Transmission) char to the PTY. pub fn write_end_of_transmission_char(&mut self, ctx: &mut ModelContext) { self.write_bytes(&[escape_sequences::C0::EOT][..], ctx); @@ -630,11 +637,12 @@ impl PtyController { /// If the write corresponds to a command, this also calls /// [`LineEditorStatus::did_execute_command()`]. fn send_write_to_event_loop(&mut self, write: PtyWrite, ctx: &mut ModelContext) { - let (bytes_to_write, is_for_command, on_write_fn) = match write { + let (bytes_to_write, is_for_command, on_write_fn, execution) = match write { PtyWrite::Command { command, shell_type, before_write_fn: on_write_fn, + execution, .. } => ( Cow::Owned(bytes_to_execute_command( @@ -644,13 +652,14 @@ impl PtyController { )), true, on_write_fn, + execution, ), PtyWrite::AgentInput { bytes, mode } => { let decorated_bytes = mode.decorate_bytes(bytes.into_owned(), self.is_bracketed_paste_enabled); - (decorated_bytes.into(), false, None) + (decorated_bytes.into(), false, None, None) } - PtyWrite::Bytes { bytes } => (bytes, false, None), + PtyWrite::Bytes { bytes } => (bytes, false, None, None), PtyWrite::RunNativeShellCompletions(state) => { self.in_flight_native_completions_state = Some(state); @@ -658,7 +667,7 @@ impl PtyController { // then wait for an OSC-based signal from the shell before we // send the text that needs to be completed. let bytes = vec![0x19_u8]; - (bytes.into(), false, None) + (bytes.into(), false, None, None) } }; @@ -674,6 +683,41 @@ impl PtyController { }); } + // A queued command must not claim the block still owned by the previous command + // or an in-band request. Attach its identity only once the shell accepts this write. + if let Some(event) = execution { + let block_id = { + let mut model = self.terminal_model.lock(); + match &event.source { + CommandExecutionSource::AI { metadata } => { + model.start_command_execution_with_ai_metadata(metadata.clone()); + } + CommandExecutionSource::SharedSession { + participant_id, + ai_metadata, + .. + } => model.start_command_execution_for_shared_session( + participant_id.clone(), + ai_metadata.clone(), + ), + CommandExecutionSource::User | CommandExecutionSource::QueuedCommand => { + model.start_command_execution(); + } + CommandExecutionSource::EnvVarCollection { metadata } => { + model.start_command_execution_from_env_var_collection(metadata.clone()); + } + } + model + .block_list_mut() + .active_block_mut() + .set_cloud_workflow_state(event.workflow_id); + model.end_in_band_command_output(false); + model.active_block_id().clone() + }; + self.is_user_command_executing = true; + ctx.emit(PtyControllerEvent::CommandStarted { event, block_id }); + } + if let Some(on_write_fn) = on_write_fn { on_write_fn(); } @@ -718,6 +762,11 @@ impl PtyController { } pub enum PtyControllerEvent { + /// The command has acquired its terminal block and is being written to the shell. + CommandStarted { + event: Box, + block_id: BlockId, + }, /// Emitted when the event loop thread has exited. PtyDisconnected, } @@ -790,6 +839,10 @@ fn wrap_bytes_in_bracketed_paste(bytes: impl IntoIterator) -> impl It .chain(escape_sequences::BRACKETED_PASTE_END.iter().copied()) } +#[cfg(test)] +#[path = "pty_controller_queue_tests.rs"] +mod queue_tests; + #[cfg(test)] #[path = "pty_controller_command_bytes_tests.rs"] mod command_bytes_tests; diff --git a/app/src/terminal/writeable_pty/pty_controller_queue_tests.rs b/app/src/terminal/writeable_pty/pty_controller_queue_tests.rs new file mode 100644 index 00000000..be8521c8 --- /dev/null +++ b/app/src/terminal/writeable_pty/pty_controller_queue_tests.rs @@ -0,0 +1,131 @@ +use galaxyui::App; + +use super::*; +use crate::ai::agent::conversation::AIConversationId; +use crate::ai::agent::AIAgentActionId; +use crate::terminal::model::block::AgentInteractionMetadata; + +struct RecordingEventLoopSender(Arc>>); + +impl EventLoopSender for RecordingEventLoopSender { + fn send(&self, message: Message) -> Result<(), EventLoopSendError> { + self.0.lock().push(message); + Ok(()) + } +} + +#[test] +fn queued_agent_command_claims_only_the_block_at_dispatch() { + App::test((), |mut app| async move { + let sessions = app.add_model(|_| Sessions::new_for_test()); + let (_events_tx, events_rx) = async_channel::unbounded(); + let dispatcher = + app.add_model(|ctx| ModelEventDispatcher::new(events_rx, sessions.clone(), ctx)); + let line_editor = + app.add_model(|ctx| LineEditorStatus::new(dispatcher.clone(), sessions.clone(), ctx)); + let (_commands_tx, commands_rx) = async_channel::unbounded(); + let model = Arc::new(FairMutex::new(TerminalModel::mock(None, None))); + let writes = Arc::new(FairMutex::new(Vec::new())); + let controller = app.add_model(|ctx| { + PtyController::new( + RecordingEventLoopSender(writes.clone()), + dispatcher, + line_editor, + sessions, + commands_rx, + model.clone(), + ctx, + ) + }); + let conversation_id = AIConversationId::new(); + let previous_action = AIAgentActionId::from("previous-command".to_string()); + let next_action = AIAgentActionId::from("follow-up-command".to_string()); + let previous_block = { + let mut model = model.lock(); + model.simulate_long_running_block("echo previous", "previous output"); + model + .block_list_mut() + .active_block_mut() + .set_agent_interaction_mode_for_requested_command( + previous_action.clone(), + None, + conversation_id, + ); + model.active_block_id().clone() + }; + controller.update(&mut app, |controller, ctx| { + controller.write_command( + &ExecuteCommandEvent { + command: "echo follow-up".into(), + session_id: 1.into(), + workflow_id: None, + workflow_command: None, + should_add_command_to_history: false, + source: CommandExecutionSource::AI { + metadata: AgentInteractionMetadata::new_hidden( + next_action.clone(), + conversation_id, + ), + }, + }, + ShellType::Zsh, + ctx, + ); + assert_eq!(controller.pending_writes.len(), 1); + controller.cancel_queued_agent_command(&previous_action); + assert_eq!(controller.pending_writes.len(), 1); + let write = controller.pending_writes.pop_front().unwrap(); + let PtyWrite::Command { + execution: Some(event), + .. + } = &write + else { + panic!("expected queued agent command"); + }; + let event = event.clone(); + controller.pending_writes.push_back(write); + controller.cancel_queued_agent_command(&next_action); + assert!(controller.pending_writes.is_empty()); + assert!(!controller.is_user_command_executing); + controller.write_command(&event, ShellType::Zsh, ctx); + }); + assert!(writes.lock().is_empty()); + { + let mut model = model.lock(); + assert_eq!( + model + .block_list() + .active_block() + .requested_command_action_id(), + Some(&previous_action) + ); + assert!(model + .block_list() + .block_for_ai_action_id(&next_action) + .is_none()); + model.finish_block(); + } + // The line editor becomes ready only after the preceding block has finished. + // Dispatch the queued write at that boundary, as the Active event normally does. + controller.update(&mut app, |controller, ctx| { + let write = controller.pending_writes.pop_front().unwrap(); + controller.send_write_to_event_loop(write, ctx); + }); + let model = model.lock(); + let new_block = model + .block_list() + .block_for_ai_action_id(&next_action) + .unwrap(); + assert_ne!(new_block.id(), &previous_block); + assert_eq!(new_block.ai_conversation_id(), Some(conversation_id)); + assert_eq!( + model + .block_list() + .block_with_id(&previous_block) + .unwrap() + .requested_command_action_id(), + Some(&previous_action) + ); + assert_eq!(writes.lock().len(), 1); + }); +} diff --git a/app/src/terminal/writeable_pty/terminal_manager_util.rs b/app/src/terminal/writeable_pty/terminal_manager_util.rs index bcbf7497..a9ad2b6e 100644 --- a/app/src/terminal/writeable_pty/terminal_manager_util.rs +++ b/app/src/terminal/writeable_pty/terminal_manager_util.rs @@ -37,8 +37,8 @@ pub fn wire_up_pty_controller_with_surface { - let Some(shell_type) = sessions + let Some(shell_type) = command_sessions .as_ref(ctx) .get(event.session_id) .map(|s| s.shell().shell_type()) @@ -88,20 +88,14 @@ pub fn wire_up_pty_controller_with_surface { + controller.update(ctx, |controller, _| { + controller.cancel_queued_agent_command(&action_id); }); - - if event.should_add_command_to_history { - update_command_history( - &event, - &model_clone, - model_event_sender.as_ref(), - &sessions, - ctx, - ); - } } PtyIntent::RunNativeShellCompletions { buffer_text, @@ -120,6 +114,18 @@ pub fn wire_up_pty_controller_with_surface { + if event.should_add_command_to_history { + update_command_history( + event, + block_id, + &model, + model_event_sender.as_ref(), + &sessions, + ctx, + ); + } + } PtyControllerEvent::PtyDisconnected => surface.update(ctx, |_, _| { model.lock().exit(ExitReason::PtyDisconnected); }), diff --git a/app/src/terminal/writeable_pty/terminal_surface.rs b/app/src/terminal/writeable_pty/terminal_surface.rs index 221f809c..e1ddbd0f 100644 --- a/app/src/terminal/writeable_pty/terminal_surface.rs +++ b/app/src/terminal/writeable_pty/terminal_surface.rs @@ -5,7 +5,7 @@ use async_channel::Sender; use warpui::AppContext; use warpui::{Entity, ViewContext}; -use crate::ai::agent::AIAgentPtyWriteMode; +use crate::ai::agent::{AIAgentActionId, AIAgentPtyWriteMode}; #[cfg(unix)] use crate::terminal::event::AfterBlockCompletedEvent; use crate::terminal::model::completions::ShellCompletion; @@ -30,6 +30,7 @@ pub enum PtyIntent { }, Resize(SizeUpdate), ExecuteCommand(ExecuteCommandEvent), + CancelQueuedAgentCommand(AIAgentActionId), RunNativeShellCompletions { buffer_text: String, results_tx: Sender>,