From 35ff43de7fa190a6e29cd9468722000147dae5aa Mon Sep 17 00:00:00 2001 From: Ryan Ward Date: Sun, 20 Sep 2026 17:11:41 -0500 Subject: [PATCH] Restore agent tool panes and require explicit session resume Persist direct-provider tool calls, results, display details, and turn identities so saved conversations rebuild individual panes. Keep restored runs dormant until explicitly resumed and bind queued shell commands to their actual execution blocks. --- app/src/ai/agent/api/convert_conversation.rs | 32 ++- app/src/ai/agent/api/convert_from.rs | 12 + app/src/ai/agent/conversation.rs | 55 ++++ app/src/ai/agent/mod.rs | 1 + app/src/ai/agent/provider_history.rs | 239 ++++++++++++++++++ app/src/ai/agent/provider_history_tests.rs | 186 ++++++++++++++ app/src/ai/agent/task.rs | 45 +++- app/src/ai/blocklist/action_model.rs | 10 + .../action_model/execute/shell_command.rs | 18 +- app/src/ai/blocklist/block/cli_controller.rs | 23 +- app/src/ai/blocklist/controller.rs | 61 ++++- app/src/ai/blocklist/controller_tests.rs | 54 ++++ app/src/ai/runtime/rig.rs | 27 ++ app/src/terminal/input_tests.rs | 3 + app/src/terminal/model/terminal_model.rs | 5 +- app/src/terminal/view.rs | 25 +- .../terminal/writeable_pty/command_history.rs | 6 +- .../terminal/writeable_pty/pty_controller.rs | 129 +++++++--- .../pty_controller_queue_tests.rs | 131 ++++++++++ .../writeable_pty/terminal_manager_util.rs | 34 ++- .../writeable_pty/terminal_surface.rs | 3 +- 21 files changed, 1011 insertions(+), 88 deletions(-) create mode 100644 app/src/ai/agent/provider_history.rs create mode 100644 app/src/ai/agent/provider_history_tests.rs create mode 100644 app/src/terminal/writeable_pty/pty_controller_queue_tests.rs 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>,