diff --git a/app/src/ai/agent/conversation.rs b/app/src/ai/agent/conversation.rs index 0fcb11e3..37d0c799 100644 --- a/app/src/ai/agent/conversation.rs +++ b/app/src/ai/agent/conversation.rs @@ -2090,6 +2090,59 @@ impl AIConversation { Ok(()) } + /// Force-cancels all streaming exchanges in this conversation. + /// + /// This is used as a fallback when Ctrl+C is pressed but there are no in-flight + /// response streams to cancel (e.g. the stream ended without proper cleanup, or a + /// subagent got stuck). It walks all exchanges and transitions any that are still + /// in `Streaming` state to `Cancelled`, emitting the appropriate UI update events. + pub fn force_cancel_all_streaming_exchanges( + &mut self, + terminal_view_id: EntityId, + reason: CancellationReason, + ctx: &mut ModelContext, + ) { + if self.transaction.is_some() { + self.commit_transaction(); + } + + let streaming_exchange_ids: Vec<_> = self + .task_store + .all_exchanges() + .filter(|exchange| matches!(exchange.output_status, AIAgentOutputStatus::Streaming { .. })) + .map(|exchange| exchange.id) + .collect(); + + for exchange_id in streaming_exchange_ids { + if let Ok(exchange) = self.get_exchange_to_update(exchange_id) { + let output = match &exchange.output_status { + AIAgentOutputStatus::Streaming { output } => { + output.as_ref().map(Shared::get_owned) + } + _ => continue, + }; + exchange.output_status = AIAgentOutputStatus::Finished { + finished_output: FinishedAIAgentOutput::Cancelled { output, reason }, + }; + exchange.finish_time = Some(Local::now()); + + let is_hidden = self.is_exchange_hidden(exchange_id); + ctx.emit(BlocklistAIHistoryEvent::UpdatedStreamingExchange { + exchange_id, + terminal_view_id, + conversation_id: self.id, + is_hidden, + }); + } + } + + self.write_updated_conversation_state(ctx); + + if !reason.is_follow_up_for_same_conversation() { + self.update_status(ConversationStatus::Cancelled, terminal_view_id, ctx); + } + } + pub fn mark_request_cancelled_due_to_revert( &mut self, terminal_view_id: EntityId, @@ -3002,7 +3055,8 @@ impl AIConversation { ); } - let new_task = Task::new_optimistic_cli_agent_subtask(block_id.clone()); + let parent_task_id = Some(self.task_store.root_task_id().to_string()); + let new_task = Task::new_optimistic_cli_agent_subtask(block_id.clone(), parent_task_id); let new_task_id = new_task.id().clone(); self.optimistic_cli_subagent_subtask_id = Some(new_task_id.clone()); self.task_store.insert(new_task); diff --git a/app/src/ai/agent/task.rs b/app/src/ai/agent/task.rs index 6c8f4f5f..53f27679 100644 --- a/app/src/ai/agent/task.rs +++ b/app/src/ai/agent/task.rs @@ -138,6 +138,7 @@ mod optimistic { #[derive(Debug, Clone)] pub(super) enum Task { Root, + #[allow(dead_code)] // Used in the server-mode path; Bedrock direct creates Server tasks directly CLIAgent(CLIAgentSubtask), } @@ -184,12 +185,42 @@ impl Task { } } - pub(super) fn new_optimistic_cli_agent_subtask(block_id: BlockId) -> Self { + pub(super) fn new_optimistic_cli_agent_subtask(block_id: BlockId, parent_task_id: Option) -> Self { + let task_id = Uuid::new_v4().to_string(); Self { - id: TaskId::new(Uuid::new_v4().to_string()), - data: TaskImpl::Optimistic(optimistic::Task::CLIAgent(optimistic::CLIAgentSubtask { - block_id, - })), + id: TaskId::new(task_id.clone()), + // Use a Server task with a source and CLI subagent_params so that: + // 1. add_messages can immediately append response messages without + // needing a CreateTask upgrade from the server (Bedrock direct path + // has no server to emit CreateTask). + // 2. cli_subagent_block_id() returns the correct block ID so the + // CLISubagentController can set up the monitoring view. + // 3. is_cli_subagent() returns true so the task is filtered from the + // main blocklist (responses only show in the CLI subagent panel). + data: TaskImpl::Server(ServerTask { + source: api::Task { + id: task_id.clone(), + description: String::new(), + dependencies: parent_task_id.map(|parent_id| api::task::Dependencies { + parent_task_id: parent_id, + }), + messages: vec![], + summary: String::new(), + server_data: String::new(), + }, + subagent_params: Some(SubagentParams { + tool_call_id: String::new(), + call: api::message::tool_call::Subagent { + task_id: task_id, + payload: String::new(), + metadata: Some(Metadata::Cli( + api::message::tool_call::subagent::CliSubagent { + command_id: block_id.as_str().to_owned(), + }, + )), + }, + }), + }), exchanges: vec![], } } diff --git a/app/src/ai/agent/task_store_tests.rs b/app/src/ai/agent/task_store_tests.rs index 507ffdc1..46fed311 100644 --- a/app/src/ai/agent/task_store_tests.rs +++ b/app/src/ai/agent/task_store_tests.rs @@ -44,7 +44,7 @@ fn create_test_task_with_exchanges(exchange_count: usize) -> Task { fn create_test_subtask_with_exchanges(exchange_count: usize) -> Task { use crate::terminal::model::block::BlockId; - let mut task = Task::new_optimistic_cli_agent_subtask(BlockId::new()); + let mut task = Task::new_optimistic_cli_agent_subtask(BlockId::new(), None); for _ in 0..exchange_count { task.append_exchange(create_test_exchange()); } @@ -486,7 +486,7 @@ fn test_linearization_nested_subtasks() { // Create child subtask with a call to grandchild use crate::terminal::model::block::BlockId; - let mut child_subtask = Task::new_optimistic_cli_agent_subtask(BlockId::new()); + let mut child_subtask = Task::new_optimistic_cli_agent_subtask(BlockId::new(), None); let child_id = child_subtask.id().clone(); let child_exchange1 = create_test_exchange(); diff --git a/app/src/ai/bedrock/convert_request.rs b/app/src/ai/bedrock/convert_request.rs index 35293d6f..da2ef41b 100644 --- a/app/src/ai/bedrock/convert_request.rs +++ b/app/src/ai/bedrock/convert_request.rs @@ -45,6 +45,37 @@ pub fn extract_new_input_messages(request: &api::Request) -> Vec { + if let Some(user_query) = &cli_query.user_query { + if !user_query.query.is_empty() { + // Include running command context as part of the user message + let query_text = if let Some(running_cmd) = &cli_query.running_command { + let mut context = format!( + "[Running command: {}]\n", + running_cmd.command + ); + if let Some(snapshot) = &running_cmd.snapshot { + if !snapshot.output.is_empty() { + context.push_str(&format!( + "[Terminal output:\n{}\n]\n", + snapshot.output + )); + } + } + context.push_str(&user_query.query); + context + } else { + user_query.query.clone() + }; + results.push(ConversationMessage { + role: MessageRole::User, + content: MessageContent::Text(query_text), + }); + } + } + } _ => {} } } @@ -262,6 +293,46 @@ fn extract_input_messages(request: &api::Request) -> Vec { }); } } + Some(api::request::input::user_inputs::user_input::Input::CliAgentUserQuery( + cli_query, + )) => { + if let Some(user_query) = &cli_query.user_query { + if !user_query.query.is_empty() { + let query_text = if let Some(running_cmd) = &cli_query.running_command { + let mut context = format!( + "[Running command: {}]\n", + running_cmd.command + ); + if let Some(snapshot) = &running_cmd.snapshot { + if !snapshot.output.is_empty() { + context.push_str(&format!( + "[Terminal output:\n{}\n]\n", + snapshot.output + )); + } + } + context.push_str(&user_query.query); + context + } else { + user_query.query.clone() + }; + results.push(api::Message { + id: uuid::Uuid::new_v4().to_string(), + task_id: task_id.clone(), + request_id: String::new(), + timestamp: None, + server_message_data: String::new(), + citations: vec![], + message: Some(api::message::Message::UserQuery( + api::message::UserQuery { + query: query_text, + ..Default::default() + }, + )), + }); + } + } + } _ => {} } } diff --git a/app/src/ai/bedrock/request_translator.rs b/app/src/ai/bedrock/request_translator.rs index ddc0dc4b..a167289f 100644 --- a/app/src/ai/bedrock/request_translator.rs +++ b/app/src/ai/bedrock/request_translator.rs @@ -74,6 +74,36 @@ pub fn extract_new_input_messages(request: &api::Request) -> Vec { + if let Some(user_query) = &cli_query.user_query { + if !user_query.query.is_empty() { + let query_text = if let Some(running_cmd) = &cli_query.running_command { + let mut context = format!( + "[Running command: {}]\n", + running_cmd.command + ); + if let Some(snapshot) = &running_cmd.snapshot { + if !snapshot.output.is_empty() { + context.push_str(&format!( + "[Terminal output:\n{}\n]\n", + snapshot.output + )); + } + } + context.push_str(&user_query.query); + context + } else { + user_query.query.clone() + }; + user_queries.push(ConversationMessage { + role: MessageRole::User, + content: MessageContent::Text(query_text), + }); + } + } + } _ => {} } } @@ -220,12 +250,22 @@ pub fn extract_user_query_text(request: &api::Request) -> Option { match input_type { api::request::input::Type::UserInputs(user_inputs) => { for user_input in &user_inputs.inputs { - if let Some(api::request::input::user_inputs::user_input::Input::UserQuery(query)) = - &user_input.input - { - if !query.query.is_empty() { - return Some(query.query.clone()); + match &user_input.input { + Some(api::request::input::user_inputs::user_input::Input::UserQuery(query)) => { + if !query.query.is_empty() { + return Some(query.query.clone()); + } } + Some(api::request::input::user_inputs::user_input::Input::CliAgentUserQuery( + cli_query, + )) => { + if let Some(user_query) = &cli_query.user_query { + if !user_query.query.is_empty() { + return Some(user_query.query.clone()); + } + } + } + _ => {} } } None @@ -348,6 +388,46 @@ fn extract_input_messages(request: &api::Request) -> Vec { }); } } + Some(api::request::input::user_inputs::user_input::Input::CliAgentUserQuery( + cli_query, + )) => { + if let Some(user_query) = &cli_query.user_query { + if !user_query.query.is_empty() { + let query_text = if let Some(running_cmd) = &cli_query.running_command { + let mut context = format!( + "[Running command: {}]\n", + running_cmd.command + ); + if let Some(snapshot) = &running_cmd.snapshot { + if !snapshot.output.is_empty() { + context.push_str(&format!( + "[Terminal output:\n{}\n]\n", + snapshot.output + )); + } + } + context.push_str(&user_query.query); + context + } else { + user_query.query.clone() + }; + results.push(api::Message { + id: uuid::Uuid::new_v4().to_string(), + task_id: task_id.clone(), + request_id: String::new(), + timestamp: None, + server_message_data: String::new(), + citations: vec![], + message: Some(api::message::Message::UserQuery( + api::message::UserQuery { + query: query_text, + ..Default::default() + }, + )), + }); + } + } + } _ => {} } } diff --git a/app/src/ai/blocklist/controller.rs b/app/src/ai/blocklist/controller.rs index 945c3bd5..816980a8 100644 --- a/app/src/ai/blocklist/controller.rs +++ b/app/src/ai/blocklist/controller.rs @@ -2213,6 +2213,27 @@ impl BlocklistAIController { request_params.bedrock_message_history = bedrock_history; request_params.bedrock_tool_result_archive = bedrock_tool_result_archive; request_params.bedrock_progressive_summary = bedrock_progressive_summary; + + // For the Bedrock path, when this is the first request in a new conversation + // (no tasks established yet), use the conversation's root task ID so the + // CreateTask response action can correctly upgrade the optimistic root task. + // + // However, for CLI subagent requests (long-running command interactions), the + // input_messages key is the subagent task ID, and we must keep that so response + // messages (AddMessagesToTask) are routed to the correct task/exchange. Overriding + // with the root task ID would cause the output to be added to a hidden root-task + // exchange instead of the visible subagent exchange. + { + let history_model = BlocklistAIHistoryModel::as_ref(ctx); + if let Some(conversation) = history_model.conversation(&conversation_id) { + let has_optimistic_cli_subagent = + conversation.has_active_subagent(); + if !has_optimistic_cli_subagent { + request_params.root_task_id = + Some(conversation.get_root_task_id().to_string()); + } + } + } let server_conversation_token_for_identifiers = conversation_data.server_conversation_token.clone(); @@ -2394,11 +2415,38 @@ impl BlocklistAIController { .in_flight_response_streams .try_cancel_streams_for_conversation(conversation_id, reason, ctx) { - // Otherwise, cancel pending actions and update the input state. + // No in-flight streams to cancel. Cancel pending actions and update the input state. self.action_model.update(ctx, |action_model, ctx| { action_model.cancel_all_pending_actions(conversation_id, Some(reason), ctx); }); self.set_input_mode_for_cancellation(ctx); + + // Force-cancel all streaming exchanges and set the conversation status to + // Cancelled. This handles the case where a query gets stuck (e.g. a subagent + // hangs or the stream ended unexpectedly without proper cleanup) and repeated + // Ctrl+C presses cannot resolve it. Without this, the conversation remains + // InProgress indefinitely, hiding the input box and blocking user interaction. + if !reason.is_follow_up_for_same_conversation() { + let history_model = BlocklistAIHistoryModel::handle(ctx); + if history_model + .as_ref(ctx) + .conversation(&conversation_id) + .is_some_and(|c| c.status().is_in_progress()) + { + let terminal_view_id = self.terminal_view_id; + history_model.update(ctx, |history_model, ctx| { + if let Some(conversation) = + history_model.conversation_mut(&conversation_id) + { + conversation.force_cancel_all_streaming_exchanges( + terminal_view_id, + reason, + ctx, + ); + } + }); + } + } } } diff --git a/app/src/lib.rs b/app/src/lib.rs index 017f044d..35568c9b 100644 --- a/app/src/lib.rs +++ b/app/src/lib.rs @@ -1866,6 +1866,11 @@ fn app_callbacks(is_integration_test: bool) -> galaxyui::platform::AppCallbacks ); })), on_will_terminate: Some(Box::new(move |ctx| { + // Persist the final app state before tearing down the writer. + // This ensures the latest session (tabs, CWD, conversations) is saved + // even if the termination bypassed individual window-close events. + ctx.dispatch_global_action("workspace:save_app", &()); + NotebookManager::handle(ctx).update(ctx, |manager, ctx| { // Notebooks are only saved periodically, so ensure that any pending changes have // been sent to the writer thread before terminating. diff --git a/app/src/pane_group/mod.rs b/app/src/pane_group/mod.rs index dd03be37..47d9cbe6 100644 --- a/app/src/pane_group/mod.rs +++ b/app/src/pane_group/mod.rs @@ -1552,10 +1552,14 @@ impl PaneGroup { }); let raw_cwd = terminal_snapshot.cwd.clone(); - let startup_directory = terminal_snapshot - .cwd - .map(PathBuf::from) - .filter(|path| path.is_dir()); + // Trust the persisted CWD without re-checking is_dir(). The path + // was validated at save time. If the directory no longer exists at + // restore time (e.g. deleted, network mount unavailable), the shell + // bootstrap script's `cd` will fail gracefully and fall back to HOME. + // Previously, the is_dir() check here would discard valid paths that + // were temporarily unavailable (e.g. network mounts not yet mounted + // at startup), causing sessions to always restore to ~/. + let startup_directory = terminal_snapshot.cwd.map(PathBuf::from); log::info!( "[session-restore] pane=terminal raw_cwd={raw_cwd:?} \ diff --git a/app/src/persistence/sqlite.rs b/app/src/persistence/sqlite.rs index 57fb8094..dac073ea 100644 --- a/app/src/persistence/sqlite.rs +++ b/app/src/persistence/sqlite.rs @@ -2755,7 +2755,17 @@ fn read_sqlite_data( let saved_tabs: Vec<_> = tabs_for_window .into_iter() .filter_map(|tab| { - let root = read_root_node(conn, tab.id).ok()?; + let root = match read_root_node(conn, tab.id) { + Ok(node) => node, + Err(err) => { + log::warn!( + "[session-restore] Failed to read root node for tab {}: {err}. \ + This tab will not be restored.", + tab.id, + ); + return None; + } + }; let panel = db_panels.get(&tab.id); let left_panel = panel diff --git a/app/src/terminal/input.rs b/app/src/terminal/input.rs index 38e00213..a9e26f55 100644 --- a/app/src/terminal/input.rs +++ b/app/src/terminal/input.rs @@ -12372,8 +12372,19 @@ impl Input { // If the agent view is inactive but the current input is detected as AI, submitting // this query triggers entering the agent view. + // + // Exception: when the agent is "tagged in" for a long-running command, we skip entering + // the agent view (which would fail because `can_start_new_conversation` returns false + // while a command is running). Instead, we fall through to the normal query submission + // path below, which correctly detects the running command and creates a CLI subagent task. if FeatureFlag::AgentView.is_enabled() && !self.agent_view_controller.as_ref(ctx).is_active() + && !self + .model + .lock() + .block_list() + .active_block() + .is_agent_tagged_in() { let prompt = self.editor.as_ref(ctx).buffer_text(ctx); let prompt = prompt.trim().to_owned(); diff --git a/app/src/terminal/view.rs b/app/src/terminal/view.rs index 4cba597f..1cd62049 100644 --- a/app/src/terminal/view.rs +++ b/app/src/terminal/view.rs @@ -6960,6 +6960,11 @@ impl TerminalView { self.agent_view_controller.update(ctx, |controller, ctx| { controller.clear_pending_exit_confirmation(ctx); }); + // Also cancel any in-progress conversation so that Ctrl+C while + // composing a message (or after submitting when the buffer hasn't + // cleared yet) properly stops the agent query and returns the + // terminal to a ready state. + self.cancel_active_conversation_via_status_bar(ctx); return; }