diff --git a/Cargo.lock b/Cargo.lock index 96b3d6a1..5d6642ea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11612,6 +11612,7 @@ dependencies = [ "chrono", "diesel", "diesel_migrations", + "galaxy_agent_core", "serde", "serde_json", "warp_multi_agent_api", diff --git a/app/src/ai/agent/api/convert_conversation.rs b/app/src/ai/agent/api/convert_conversation.rs index 0132c15f..4b6f9abb 100644 --- a/app/src/ai/agent/api/convert_conversation.rs +++ b/app/src/ai/agent/api/convert_conversation.rs @@ -73,6 +73,7 @@ pub fn convert_conversation_data_to_ai_conversation( let agent_conversation_data = match restoration_mode { RestorationMode::Fork => AgentConversationData { agent_backend: Default::default(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: Default::default(), @@ -99,6 +100,7 @@ pub fn convert_conversation_data_to_ai_conversation( }, RestorationMode::Continue => AgentConversationData { agent_backend: Default::default(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: Default::default(), diff --git a/app/src/ai/agent/conversation.rs b/app/src/ai/agent/conversation.rs index f2c053a5..3d49e451 100644 --- a/app/src/ai/agent/conversation.rs +++ b/app/src/ai/agent/conversation.rs @@ -17,7 +17,7 @@ use galaxy_core::ui::theme::GalaxyTheme; use galaxyui::color::ColorU; use galaxyui::{AppContext, EntityId, ModelContext, SingletonEntity}; use itertools::Itertools as _; -use persistence::model::LocalAgentGroup; +use persistence::model::{LocalAgentGroup, ProviderConversationContext}; use serde::{Deserialize, Serialize}; use uuid::Uuid; use vec1::{Size0Error, Vec1}; @@ -69,7 +69,6 @@ use crate::persistence::model::{ }; use crate::persistence::ModelEvent; use crate::server::ids::ServerId; -use crate::terminal::general_settings::GeneralSettings; use crate::terminal::model::block::{ AgentInteractionMetadata, AgentViewVisibility, BlockId, SerializedAIMetadata, SerializedBlock, }; @@ -453,11 +452,27 @@ impl AIConversation { tasks: Vec, conversation_data: Option, ) -> Result { - let bedrock_message_history: Vec = tasks - .iter() - .flat_map(|task| task.messages.iter()) - .filter_map(crate::ai::provider::request_translator::convert_proto_message) - .collect(); + let provider_context = conversation_data + .as_ref() + .and_then(|data| data.provider_context.clone()); + let (bedrock_message_history, tool_result_archive) = match provider_context { + Some(context) => (context.messages, context.tool_result_archive), + None => { + // Legacy rows have only UI task messages. Preserve that fallback, but + // do not use it for new rows, whose provider context may be compacted. + let messages: Vec<_> = tasks + .iter() + .flat_map(|task| task.messages.iter()) + .filter_map(crate::ai::provider::request_translator::convert_proto_message) + .collect(); + // Legacy UI messages do not share provider-message indices. Do not + // truncate them using a provider compaction count: retain everything + // recoverable and rebuild recall from the available tool messages. + let mut archive = Vec::new(); + galaxy_agent_core::archive_tool_results(&mut archive, messages.clone()); + (messages, archive) + } + }; let (task_store, todo_lists, status) = if tasks.is_empty() { // Bypass `derive_status_from_root_task`: it would return `Success` @@ -575,7 +590,6 @@ impl AIConversation { last_event_sequence, pinned, progressive_summary, - messages_summarized_up_to, ) = if let Some(data) = conversation_data { let server_conversation_token = data .server_conversation_token @@ -626,7 +640,6 @@ impl AIConversation { data.last_event_sequence, data.pinned, data.progressive_summary.clone(), - data.messages_summarized_up_to, ) } else { ( @@ -647,7 +660,6 @@ impl AIConversation { None, false, None, - 0, ) }; @@ -715,9 +727,10 @@ impl AIConversation { orchestration_configs: HashMap::new(), pinned, bedrock_message_history, - tool_result_archive: Vec::new(), + tool_result_archive, progressive_summary, - messages_summarized_up_to, + // The previous response stream no longer exists; its drain count cannot apply. + messages_summarized_up_to: 0, current_context_tokens: 0, latest_model_call_usage: ProviderUsage::default(), has_pending_progressive_summary: false, @@ -841,6 +854,19 @@ impl AIConversation { &mut self.bedrock_message_history } + pub(crate) fn provider_context(&self) -> ProviderConversationContext { + ProviderConversationContext { + messages: self.bedrock_message_history.clone(), + tool_result_archive: self.tool_result_archive.clone(), + } + } + + pub(crate) fn set_provider_context(&mut self, context: ProviderConversationContext) { + self.bedrock_message_history = context.messages; + self.tool_result_archive = context.tool_result_archive; + self.messages_summarized_up_to = 0; + } + pub fn tool_result_archive(&self) -> &[crate::ai::provider::convert::ConversationMessage] { &self.tool_result_archive } @@ -4016,10 +4042,8 @@ impl AIConversation { return; } - // Check if session restoration is enabled before writing any state. - if !*GeneralSettings::as_ref(ctx).restore_session - || !AppExecutionMode::as_ref(ctx).can_save_session() - { + // Conversation history is durable even when automatic session reopening is disabled. + if !AppExecutionMode::as_ref(ctx).can_save_session() { return; } @@ -4064,6 +4088,7 @@ impl AIConversation { .filter_map(|task| task.source_for_persistence()) .collect(), conversation_data: AgentConversationData { + provider_context: Some(self.provider_context()), agent_backend: self.agent_backend.clone(), active_provider_run_json: self.active_provider_run_json.clone(), local_agent_groups: self.local_agent_groups.clone(), @@ -4097,14 +4122,9 @@ impl AIConversation { messages_summarized_up_to: self.messages_summarized_up_to, }, }; - ctx.spawn( - async move { - if let Err(e) = sqlite_sender.send(event) { - log::warn!("Failed to send updated AI tasks to sqlite writer thread: {e:?}"); - } - }, - |_, _, _| {}, - ); + if let Err(e) = sqlite_sender.send(event) { + log::error!("Failed to send updated AI tasks to sqlite writer thread: {e:?}"); + } } pub fn rollback_transaction(&mut self, response_stream_id: &ResponseStreamId) { diff --git a/app/src/ai/agent/conversation_tests.rs b/app/src/ai/agent/conversation_tests.rs index 04b1151c..fd3ac96d 100644 --- a/app/src/ai/agent/conversation_tests.rs +++ b/app/src/ai/agent/conversation_tests.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; +use galaxy_agent_core::{ContentPart, ConversationMessage, MessageContent, MessageRole}; use galaxy_core::features::FeatureFlag; use warp_multi_agent_api as api; @@ -8,7 +9,9 @@ use super::{ ConversationStatus, RestoreConversationError, }; use crate::ai::artifacts::Artifact; -use crate::persistence::model::{AcpConversationData, AgentBackend, AgentConversationData}; +use crate::persistence::model::{ + AcpConversationData, AgentBackend, AgentConversationData, ProviderConversationContext, +}; fn restored_conversation(conversation_data: Option) -> AIConversation { AIConversation::new_restored( @@ -706,3 +709,62 @@ fn fetched_memories_dedupes_keeping_first_position_and_latest_data() { ] ); } + +#[test] +fn restored_provider_context_preserves_compacted_history_and_tool_archive() { + let context = ProviderConversationContext { + messages: vec![ConversationMessage { + role: MessageRole::Assistant, + content: MessageContent::MultiPart(vec![ + ContentPart::Reasoning { + text: "retained reasoning".into(), + signature: Some("signature".into()), + }, + ContentPart::Text("retained response".into()), + ]), + }], + tool_result_archive: vec![ConversationMessage { + role: MessageRole::User, + content: MessageContent::ToolResult { + tool_use_id: "old-tool".into(), + content: "original output".into(), + is_error: false, + }, + }], + }; + let data = AgentConversationData { + provider_context: Some(context.clone()), + progressive_summary: Some("Older history".into()), + ..Default::default() + }; + let data = serde_json::from_str(&serde_json::to_string(&data).unwrap()).unwrap(); + let restored = restored_conversation(Some(data)); + assert_eq!(restored.provider_context(), context); + assert_eq!(restored.progressive_summary(), Some("Older history")); + assert!(restored.active_provider_run_json().is_none()); +} + +#[test] +fn explicit_empty_provider_context_does_not_reconstruct_old_task_messages() { + let data = AgentConversationData { + provider_context: Some(ProviderConversationContext { + messages: Vec::new(), + tool_result_archive: Vec::new(), + }), + ..Default::default() + }; + let task = api::Task { + id: "root-task".into(), + messages: vec![user_query_message("query", "request", "old query")], + dependencies: None, + description: String::new(), + summary: String::new(), + server_data: String::new(), + }; + let legacy = + AIConversation::new_restored(AIConversationId::new(), vec![task.clone()], None).unwrap(); + assert!(!legacy.bedrock_message_history().is_empty()); + let restored = + AIConversation::new_restored(AIConversationId::new(), vec![task], Some(data)).unwrap(); + assert!(restored.bedrock_message_history().is_empty()); +} diff --git a/app/src/ai/blocklist/controller.rs b/app/src/ai/blocklist/controller.rs index a70c489a..b6b4c37f 100644 --- a/app/src/ai/blocklist/controller.rs +++ b/app/src/ai/blocklist/controller.rs @@ -33,6 +33,7 @@ use input_context::{input_context_for_request, parse_context_attachments}; use itertools::Itertools; use parking_lot::FairMutex; use pending_response_streams::PendingResponseStreams; +use persistence::model::ProviderConversationContext; use session_sharing_protocol::common::ParticipantId; use settings::Setting; pub use slash_command::*; @@ -6519,9 +6520,16 @@ impl BlocklistAIController { Err(error) => return Err(error), }; snapshot.queued_follow_ups = queued_follow_ups; + let context = ProviderConversationContext { + messages: snapshot.run.transcript()[snapshot.persistence_offset..].to_vec(), + tool_result_archive: snapshot.run.tool_result_archive().to_vec(), + }; let json = serde_json::to_string(&snapshot) .map_err(|error| format!("failed to serialize active provider run: {error}"))?; BlocklistAIHistoryModel::handle(ctx).update(ctx, |history_model, ctx| { + if let Some(conversation) = history_model.conversation_mut(&conversation_id) { + conversation.set_provider_context(context); + } history_model .persist_active_provider_run_json(conversation_id, Some(json), ctx) .map_err(|error| format!("failed to persist active provider run: {error:?}")) @@ -8290,6 +8298,20 @@ impl BlocklistAIController { let offset = run.persistence_offset.min(transcript.len()); *messages_sent = transcript[offset..].to_vec(); } + // Retain the exact context before terminal projection clears the active checkpoint. + // UI task messages are a lossy representation of tool/reasoning messages. + let transcript = run.coordinator.run().transcript(); + let offset = run.persistence_offset.min(transcript.len()); + let context = ProviderConversationContext { + messages: transcript[offset..].to_vec(), + tool_result_archive: run.coordinator.run().tool_result_archive().to_vec(), + }; + BlocklistAIHistoryModel::handle(ctx).update(ctx, |history, ctx| { + if let Some(conversation) = history.conversation_mut(&conversation_id) { + conversation.set_provider_context(context); + conversation.write_updated_conversation_state(ctx); + } + }); let aggregate_usage = run.coordinator.run().usage().clone(); let events = match run.projector.finish(&outcome, &aggregate_usage) { Ok(events) => events, diff --git a/app/src/ai/blocklist/history_model.rs b/app/src/ai/blocklist/history_model.rs index 96e4a704..715091ee 100644 --- a/app/src/ai/blocklist/history_model.rs +++ b/app/src/ai/blocklist/history_model.rs @@ -1703,6 +1703,7 @@ impl BlocklistAIHistoryModel { let conversation_data = AgentConversationData { agent_backend: source_conversation.agent_backend().for_fork(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: source_conversation.tool_titles().clone(), @@ -1870,6 +1871,7 @@ impl BlocklistAIHistoryModel { // be recomputed based on the retained exchanges in a follow-up. let conversation_data = AgentConversationData { agent_backend: conversation.agent_backend().for_fork(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: conversation.tool_titles().clone(), @@ -2898,6 +2900,7 @@ fn merged_remote_child_placeholder_conversation_data( cloud_conversation: &AIConversation, ) -> AgentConversationData { AgentConversationData { + provider_context: None, // Placeholder authoritative. agent_backend: placeholder.agent_backend().clone(), diff --git a/app/src/ai/blocklist/history_model/conversation_loader.rs b/app/src/ai/blocklist/history_model/conversation_loader.rs index b6a26555..eb1c8ef7 100644 --- a/app/src/ai/blocklist/history_model/conversation_loader.rs +++ b/app/src/ai/blocklist/history_model/conversation_loader.rs @@ -83,7 +83,14 @@ pub fn convert_persisted_conversation_to_ai_conversation_with_metadata( } }; - let conversation_data = serde_json::from_str::(&conversation_data).ok(); + let conversation_data = match serde_json::from_str::(&conversation_data) + { + Ok(data) => Some(data), + Err(error) => { + log::error!("Failed to restore conversation {conversation_id} metadata: {error}"); + return None; + } + }; // Local-DB restore: an empty `agent_tasks` row is the normal shape of a // child conversation persisted before its first server response, so diff --git a/app/src/app_state.rs b/app/src/app_state.rs index e3cb96ce..64ba9525 100644 --- a/app/src/app_state.rs +++ b/app/src/app_state.rs @@ -23,7 +23,7 @@ use crate::workspace::tab_group::TabGroupId; use crate::workspace::view::left_panel::ToolPanelView; use crate::workspace::WorkspaceRegistry; -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, Default, PartialEq)] pub struct AppState { pub windows: Vec, pub active_window_index: Option, diff --git a/app/src/lib.rs b/app/src/lib.rs index a3384413..c99a2693 100644 --- a/app/src/lib.rs +++ b/app/src/lib.rs @@ -2829,6 +2829,24 @@ fn launch(ctx: &mut galaxyui::AppContext, app_state: Option, launch_mo timer.mark_interval_end("WINDOWS_CREATED"); }); + #[cfg(feature = "local_fs")] + if matches!(launch_mode, LaunchMode::App { .. }) + && PersistenceWriter::as_ref(ctx).sender().is_none() + { + let windows: Vec<_> = ctx.window_ids().collect(); + for window_id in windows { + ToastStack::handle(ctx).update(ctx, |toasts, ctx| { + toasts.add_persistent_toast( + DismissibleToast::error( + "Galaxy could not restore all local data. Saving is disabled to protect existing data. Check the logs and restart after resolving the problem.".to_string(), + ), + window_id, + ctx, + ); + }); + } + } + // TODO(ben): We should skip this for LaunchMode::Test. #[cfg(any(target_os = "macos", target_os = "windows"))] { diff --git a/app/src/pane_group/mod_tests.rs b/app/src/pane_group/mod_tests.rs index b7958f2a..b750364b 100644 --- a/app/src/pane_group/mod_tests.rs +++ b/app/src/pane_group/mod_tests.rs @@ -388,6 +388,7 @@ fn persisted_remote_child_conversation( conversation_id: conversation_id.to_string(), conversation_data: serde_json::to_string(&AgentConversationData { agent_backend: Default::default(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: Default::default(), @@ -3546,3 +3547,35 @@ fn test_auto_hiding_completed_child_preserves_parent_focus_and_reopen() { }); }); } + +#[test] +fn local_transcript_snapshot_retains_conversation_links() { + App::test((), |mut app| async move { + initialize_app(&mut app); + let pane_group = mock_pane_group(&mut app, Default::default()); + pane_group.update(&mut app, |panes, ctx| { + let pane_id = get_newly_created_pane_id(panes, &[]); + let terminal = panes.terminal_view_from_pane_id(pane_id, ctx).unwrap(); + let conversation_id = + BlocklistAIHistoryModel::handle(ctx).update(ctx, |history, ctx| { + history.start_new_conversation(terminal.id(), false, false, false, ctx) + }); + terminal + .as_ref(ctx) + .model + .lock() + .set_conversation_transcript_viewer_status(Some( + ConversationTranscriptViewerStatus::ViewingLocalConversation, + )); + let PaneNodeSnapshot::Leaf(leaf) = panes.snapshot(ctx) else { + panic!("expected a terminal leaf"); + }; + let LeafContents::Terminal(snapshot) = leaf.contents else { + panic!("expected a terminal snapshot"); + }; + assert!(snapshot + .conversation_ids_to_restore + .contains(&conversation_id)); + }); + }); +} diff --git a/app/src/pane_group/pane/terminal_pane.rs b/app/src/pane_group/pane/terminal_pane.rs index cc3234a8..f67fd634 100644 --- a/app/src/pane_group/pane/terminal_pane.rs +++ b/app/src/pane_group/pane/terminal_pane.rs @@ -49,7 +49,6 @@ use crate::server::server_api::ai::{SpawnAgentRequest, UserQueryMode}; use crate::server::server_api::ServerApiProvider; use crate::session_management::SessionNavigationData; use crate::terminal::cli_agent_sessions::CLIAgentSessionsModel; -use crate::terminal::general_settings::GeneralSettings; use crate::terminal::shared_session::manager::{Manager, ManagerEvent}; use crate::terminal::shared_session::role_change_modal::RoleChangeOpenSource; #[cfg(not(target_family = "wasm"))] @@ -493,6 +492,13 @@ impl PaneContent for TerminalPane { // Capture the current input_config from the AI input model let current_input_config = view.input_config(app.as_ref()); + let transcript_task_id = { + let model = view.model.lock(); + model + .is_conversation_transcript_viewer() + .then(|| model.ambient_agent_task_id()) + .flatten() + }; if view.model.lock().shared_session_status().is_viewer() { // We save and restore ambient agent sessions @@ -530,33 +536,13 @@ impl PaneContent for TerminalPane { uuid: self.uuid.clone(), task_id: Some(task_id), }) - } else if view.model.lock().is_conversation_transcript_viewer() { - // Conversation transcript viewers (opened from the conversation list) - // can be restored via the ambient agent task if one exists. - let task_id = view.model.lock().ambient_agent_task_id(); - if task_id.is_some() { - log::trace!("[session-snapshot] pane=transcript/ambient task_id={task_id:?}"); - LeafContents::AmbientAgent(AmbientAgentPaneSnapshot { - uuid: self.uuid.clone(), - task_id, - }) - } else { - let cwd = view.pwd_if_local(app); - log::trace!("[session-snapshot] pane=transcript cwd={cwd:?} is_active={is_active}"); - LeafContents::Terminal(TerminalPaneSnapshot { - uuid: self.uuid.clone(), - cwd, - is_active, - is_read_only: false, - shell_launch_data: None, - input_config: None, - llm_model_override: None, - active_profile_id: None, - conversation_ids_to_restore: vec![], - active_conversation_id: None, - }) - } + } else if let Some(task_id) = transcript_task_id { + LeafContents::AmbientAgent(AmbientAgentPaneSnapshot { + uuid: self.uuid.clone(), + task_id: Some(task_id), + }) } else { + // Local transcript viewers retain their conversation links just like live panes. let llm_model_override = LLMPreferences::as_ref(app).get_base_llm_override(self.terminal_view(app).id()); @@ -1038,9 +1024,7 @@ fn handle_terminal_view_event( Event::BlockCompleted { block, is_local } => { match group.terminal_session_by_id(pane_id) { Some(pane) => { - if *GeneralSettings::as_ref(ctx).restore_session - && AppExecutionMode::as_ref(ctx).can_save_session() - { + if AppExecutionMode::as_ref(ctx).can_save_session() { if let Some(sender) = &group.model_event_sender { let block_completed_event = ModelEvent::SaveBlock(BlockCompleted { pane_id: pane.session_uuid(), @@ -2249,10 +2233,8 @@ fn handle_ai_history_event( is_hidden, .. } => { - // Check if session restoration is enabled. - if !*GeneralSettings::as_ref(ctx).restore_session - || !AppExecutionMode::as_ref(ctx).can_save_session() - { + // Save query history independently of automatic session reopening. + if !AppExecutionMode::as_ref(ctx).can_save_session() { return; } diff --git a/app/src/persistence/README.md b/app/src/persistence/README.md index a9c49479..8e409bbf 100644 --- a/app/src/persistence/README.md +++ b/app/src/persistence/README.md @@ -1,3 +1,22 @@ +# Saving and restoring local data + +On macOS, Galaxy stores its local database at `~/.galaxy/galaxy.sqlite`. +`general.restore_session` controls automatic reopening of the previous session; +conversation history and session snapshots continue to be saved when it is off. +Terminal scrollback retention does not limit persisted provider conversation context. + +New conversation records store the provider message history and tool-result archive +alongside the progressive summary. Older records fall back to reconstructing the +available task messages; provider details omitted by those older saves cannot be +recovered exactly. + +Startup reads use a consistent SQLite snapshot and recover independent data domains. +If a domain or saved tab/conversation cannot be read, Galaxy logs the failure, shows a persistent startup warning, and +leaves database writing disabled for that launch so partial in-memory state cannot +overwrite the original records. Repair the underlying problem and restart to resume +saving. Snapshot deduplication happens only after a successful database commit; +transactional snapshot and conversation saves retry transient database contention. + # How to perform migrations ## How do migrations work? @@ -27,20 +46,20 @@ This will create a new folder with an up.sql and down.sql. ## Step 3: Run the migration + generate the schema ``` cd -diesel migration run --database-url="/Users/$USER/.galaxy-local/galaxy.sqlite" +diesel migration run --database-url="/Users/$USER/.galaxy/galaxy.sqlite" ``` This will run the migration on the same warp that runs when you run the app locally. This automatically generates or updates the `crates/persistence/src/schema.rs`. We do not make manual edits to `schema.rs`. You can also print the schema from a database that already has the migration with: ``` -diesel print-schema --database-url="/Users/$USER/.galaxy-local/galaxy.sqlite" +diesel print-schema --database-url="/Users/$USER/.galaxy/galaxy.sqlite" ``` ## Reverting/redo-ing migrations As you are writing features and changing branches, you'll want to undo migrations to fix your database and make it compatible with older code. Redo-ing can also be helpful as you are iterating on your schema. ``` -diesel migration revert --database-url="/Users/$USER/.galaxy-local/galaxy.sqlite" -diesel migration redo --database-url="/Users/$USER/.galaxy-local/galaxy.sqlite" +diesel migration revert --database-url="/Users/$USER/.galaxy/galaxy.sqlite" +diesel migration redo --database-url="/Users/$USER/.galaxy/galaxy.sqlite" ``` # Schema style diff --git a/app/src/persistence/agent.rs b/app/src/persistence/agent.rs index e0e56170..ee8f8117 100644 --- a/app/src/persistence/agent.rs +++ b/app/src/persistence/agent.rs @@ -204,6 +204,7 @@ pub(super) fn select_conversations_to_evict( pub(super) fn read_agent_conversations( conn: &mut SqliteConnection, + errors: &mut Vec, ) -> Result, diesel::result::Error> { let mut conversations_by_id = HashMap::::from_iter( agent_conversations::table @@ -242,7 +243,27 @@ pub(super) fn read_agent_conversations( } } - conversations_by_id.retain(|c_id, _| !invalid_conversation_ids.contains(c_id)); + for conversation_id in &invalid_conversation_ids { + errors.push(format!( + "Failed to restore conversation {conversation_id}: invalid task data" + )); + } + conversations_by_id.retain(|id, _| !invalid_conversation_ids.contains(id)); + // Keep healthy conversations available, but mark the restore incomplete so + // invalid metadata can never be replaced with empty defaults on disk. + conversations_by_id.retain(|id, conversation| { + match serde_json::from_str::( + &conversation.conversation.conversation_data, + ) { + Ok(_) => true, + Err(error) => { + let message = format!("Failed to restore conversation {id} metadata: {error}"); + log::error!("{message}"); + errors.push(message); + false + } + } + }); Ok(conversations_by_id.into_values().collect()) } diff --git a/app/src/persistence/mod.rs b/app/src/persistence/mod.rs index f2c72f4f..addc02ec 100644 --- a/app/src/persistence/mod.rs +++ b/app/src/persistence/mod.rs @@ -150,6 +150,7 @@ impl SingletonEntity for PersistenceWriter {} /// /// This data belongs to the local installation rather than an inherited Warp /// account, so logging out of a legacy account must not clear it. +#[derive(Default)] pub struct PersistedData { /// Session restoration data pub app_state: AppState, diff --git a/app/src/persistence/sqlite.rs b/app/src/persistence/sqlite.rs index 80f2d2bf..2a1dd203 100644 --- a/app/src/persistence/sqlite.rs +++ b/app/src/persistence/sqlite.rs @@ -136,7 +136,10 @@ pub fn initialize( let database_path = database_file_path_for_scope(&scope); match init_db(&scope) { Ok(mut conn) => { - let persisted_data = read_persisted_data(&mut conn, ctx); + let (persisted_data, restore_complete) = read_persisted_data(&mut conn, ctx); + if !restore_complete { + return (Some(persisted_data), None); + } let writer_handles = match start_writer(conn, database_path.clone()) { Ok(writer_handles) => Some(writer_handles), @@ -149,7 +152,7 @@ pub fn initialize( None } }; - (persisted_data, writer_handles) + (Some(persisted_data), writer_handles) } Err(err) => { send_telemetry_from_app_ctx!( @@ -165,16 +168,14 @@ pub fn initialize( fn read_persisted_data( conn: &mut SqliteConnection, ctx: &mut AppContext, -) -> Option> { +) -> (Box, bool) { let user_uid = AuthStateProvider::as_ref(ctx).get().user_id(); - match read_sqlite_data(conn, user_uid) { - Ok(app_state) => Some(Box::new(app_state)), - Err(err) => { - send_telemetry_from_app_ctx!(TelemetryEvent::DatabaseReadError(err.to_string()), ctx); - report_error!(anyhow::Error::new(err).context("Failed to read persisted data")); - None - } + let (data, errors) = read_sqlite_data_with_recovery(conn, user_uid); + if !errors.is_empty() { + send_telemetry_from_app_ctx!(TelemetryEvent::DatabaseReadError(errors.join("; ")), ctx); + report_error!(anyhow!("Incomplete SQLite restore; database writes are disabled for this launch to preserve existing data: {}", errors.join("; "))); } + (Box::new(data), errors.is_empty()) } /// Returns a read-only connection to the sqlite database. @@ -466,6 +467,7 @@ fn start_writer(conn: SqliteConnection, database_path: PathBuf) -> Result { @@ -490,6 +492,15 @@ fn start_writer(conn: SqliteConnection, database_path: PathBuf) -> Result { + if let Err(err) = save_snapshot_if_changed( + &mut current_conn, + app_state, + &mut last_saved_snapshot, + ) { + report_db_error("session snapshot", err, &database_path); + } + } event => { if let Err(err) = handle_model_event(event, &mut current_conn) { report_db_error("Model", err, &database_path); @@ -502,6 +513,43 @@ fn start_writer(conn: SqliteConnection, database_path: PathBuf) -> Result, +) -> Result<()> { + if last_saved.as_ref() == Some(&snapshot) { + return Ok(()); + } + retry_sqlite_write(|| save_app_state(conn, &snapshot))?; + *last_saved = Some(snapshot); + Ok(()) +} + +/// Retry transactional, idempotent saves on transient SQLite contention. Permanent +/// failures remain errors, and must never be recorded as committed state. +fn retry_sqlite_write(mut write: impl FnMut() -> Result) -> Result { + for attempt in 0..3 { + match write() { + Ok(value) => return Ok(value), + Err(error) => { + let busy = error.chain().any(|cause| { + cause.downcast_ref::().is_some_and(|error| { + matches!(error, Error::DatabaseError(_, info) + if info.message().contains("locked") || info.message().contains("busy")) + }) + }); + if !busy || attempt == 2 { + return Err(error); + } + thread::sleep(std::time::Duration::from_millis(100 * (attempt + 1))); + } + } + } + unreachable!("the last write attempt always returns") +} + /// Handles a single [`ModelEvent`] by dispatching to an event-specific function. /// Events which affect the SQLite writer event loop _must_ instead be handled by the event loop itself: /// * [`ModelEvent::Terminate`] @@ -636,13 +684,15 @@ fn handle_model_event(event: ModelEvent, connection: &mut SqliteConnection) -> a conversation_id, updated_tasks, conversation_data, - } => upsert_agent_conversation( - connection, - &conversation_id, - &updated_tasks, - conversation_data, - ) - .map_err(anyhow::Error::from), + } => retry_sqlite_write(|| { + upsert_agent_conversation( + connection, + &conversation_id, + &updated_tasks, + conversation_data.clone(), + ) + .map_err(anyhow::Error::from) + }), ModelEvent::DeleteMultiAgentConversations { conversation_ids } => { delete_agent_conversations(connection, conversation_ids) .map_err(anyhow::Error::from) @@ -2355,19 +2405,200 @@ fn box_persisted_generic_string_object( } } -/// This is not in a transaction. The interface for a transaction is a bit awkward, -/// and makes it invalid to write the logic recursively. It's ok it's not in a -/// transaction because we should be the only connection using the database. -/// -/// One notable exception is the case where there may be two warp apps running -/// in the same bundle. In this case, we may read some garbage, but all that will -/// happen is the user won't have session restoration. -/// -/// In the future, the awkwardness of the transaction interface is resolved in diesel 2.0.0. +/// Use a consistent SQLite snapshot even when another Galaxy process is writing. +/// A failed domain is reported independently; incomplete restoration disables writes. +fn read_sqlite_data_with_recovery( + conn: &mut SqliteConnection, + current_user_id: Option, +) -> (PersistedData, Vec) { + match conn.transaction::<_, Error, _>(|conn| Ok(read_sqlite_domains(conn, current_user_id))) { + Ok(result) => result, + Err(error) => ( + PersistedData::default(), + vec![format!("Failed to read SQLite snapshot: {error}")], + ), + } +} + +fn read_sqlite_domains( + conn: &mut SqliteConnection, + current_user_id: Option, +) -> (PersistedData, Vec) { + let mut errors = Vec::new(); + let layout = read_session_layout(conn, &mut errors); + let (windows, active_window_index) = recover_read("session layout", layout, &mut errors); + let cloud_objects = recover_read( + "local objects", + read_cloud_objects(conn, current_user_id), + &mut errors, + ); + let (workspaces, current_workspace_uid) = + recover_read("workspaces", read_workspaces(conn), &mut errors); + let commands = recover_read( + "commands", + (|| -> Result<_, Error> { + let commands = schema::commands::dsl::commands + // Ensure the commands come into memory sorted chronologically. + .order(schema::commands::columns::id.desc()) + .load_iter::(conn)? + .filter_map(|command| command.ok()) + .map(PersistedCommand::from) + .collect(); + + Ok(commands) + })(), + &mut errors, + ); + let user_profiles = recover_read( + "user_profiles", + (|| -> Result<_, Error> { + let user_profiles = schema::user_profiles::dsl::user_profiles + .load_iter::(conn)? + .filter_map(|user_profile| user_profile.ok()) + .map(user_profile_from_persistence) + .collect(); + + Ok(user_profiles) + })(), + &mut errors, + ); + let object_actions = recover_read( + "object_actions", + (|| -> Result<_, Error> { + let object_actions: Vec = schema::object_actions::dsl::object_actions + .load_iter::(conn)? + .filter_map(|object_action| object_action.ok()) // parse into PersistedObjectAction + .filter_map(|action| object_action_from_persisted(action).ok()) + .collect(); + + Ok(object_actions) + })(), + &mut errors, + ); + let server_experiments = recover_read( + "server_experiments", + (|| -> Result<_, Error> { + let server_experiments = schema::server_experiments::dsl::server_experiments + .load_iter::(conn)? + .filter_map(|server_experiment| server_experiment.ok()) + .filter_map(|server_experiment| { + ServerExperiment::from_string(server_experiment.experiment).ok() + }) + .collect(); + + Ok(server_experiments) + })(), + &mut errors, + ); + let restored_blocks = recover_read( + "restored_blocks", + get_all_restored_blocks(conn), + &mut errors, + ); + let running_mcp_servers = recover_read( + "running_mcp_servers", + load_active_mcp_servers(conn), + &mut errors, + ); + let time_of_next_force_object_refresh = recover_read( + "time_of_next_force_object_refresh", + read_time_of_next_force_object_refresh(conn), + &mut errors, + ); + let ai_queries = recover_read("ai_queries", read_ai_queries(conn), &mut errors); + let codebase_indices = recover_read( + "codebase_indices", + get_all_codebase_index_metadata(conn), + &mut errors, + ); + let workspace_language_servers = recover_read( + "workspace_language_servers", + get_all_workspace_language_servers_by_workspace(conn), + &mut errors, + ); + let conversations = read_agent_conversations(conn, &mut errors); + let multi_agent_conversations = + recover_read("multi_agent_conversations", conversations, &mut errors); + let projects = recover_read("projects", get_all_projects(conn), &mut errors); + let project_rules = recover_read("project_rules", get_all_project_rules(conn), &mut errors); + let ignored_suggestions = recover_read( + "ignored_suggestions", + get_all_ignored_suggestions(conn), + &mut errors, + ); + let mcp_server_installations = recover_read( + "mcp_server_installations", + get_all_mcp_server_installations(conn), + &mut errors, + ); + let mcp_servers_to_restore = recover_read( + "mcp_servers_to_restore", + get_mcp_servers_to_restore(conn), + &mut errors, + ); + let app_state = AppState { + windows, + active_window_index, + block_lists: Arc::new(restored_blocks), + running_mcp_servers, + }; + ( + PersistedData { + app_state, + cloud_objects, + workspaces, + current_workspace_uid, + command_history: commands, + user_profiles, + time_of_next_force_object_refresh, + object_actions, + experiments: server_experiments, + ai_queries, + codebase_indices, + workspace_language_servers, + multi_agent_conversations, + projects, + project_rules, + ignored_suggestions, + mcp_server_installations, + mcp_servers_to_restore, + }, + errors, + ) +} + +fn recover_read( + domain: &str, + result: Result, + errors: &mut Vec, +) -> T { + match result { + Ok(value) => value, + Err(error) => { + let message = format!("Failed to restore {domain}: {error}"); + log::error!("{message}"); + errors.push(message); + T::default() + } + } +} + +#[cfg(test)] fn read_sqlite_data( conn: &mut SqliteConnection, current_user_id: Option, -) -> Result { +) -> Result { + let (data, errors) = read_sqlite_data_with_recovery(conn, current_user_id); + if !errors.is_empty() { + bail!("{}", errors.join("; ")); + } + Ok(data) +} + +fn read_session_layout( + conn: &mut SqliteConnection, + errors: &mut Vec, +) -> Result<(Vec, Option), Error> { use schema::windows::dsl::*; let active_window_id = schema::app::dsl::app @@ -2427,7 +2658,18 @@ 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(root) => root, + Err(error) => { + let message = format!( + "Failed to restore session layout tab {}: {error}", + tab.id + ); + log::error!("{message}"); + errors.push(message); + return None; + } + }; let panel = db_panels.get(&tab.id); let left_panel = panel @@ -2560,6 +2802,13 @@ fn read_sqlite_data( ) .collect(); + Ok((saved_windows, active_window_index)) +} + +fn read_cloud_objects( + conn: &mut SqliteConnection, + current_user_id: Option, +) -> Result>, Error> { let read_context = load_cloud_object_read_context(conn, current_user_id)?; let mut cloud_objects: Vec> = Vec::new(); cloud_objects.extend( @@ -2583,6 +2832,12 @@ fn read_sqlite_data( .map(box_persisted_generic_string_object), ); + Ok(cloud_objects) +} + +fn read_workspaces( + conn: &mut SqliteConnection, +) -> Result<(Vec, Option), Error> { let db_teams: Vec = schema::teams::dsl::teams.load(conn)?; let team_member_rows: Vec = @@ -2671,79 +2926,7 @@ fn read_sqlite_data( .optional()? .map(|uid| uid.into()); - let commands = schema::commands::dsl::commands - // Ensure the commands come into memory sorted chronologically. - .order(schema::commands::columns::id.desc()) - .load_iter::(conn)? - .filter_map(|command| command.ok()) - .map(PersistedCommand::from) - .collect(); - - let user_profiles = schema::user_profiles::dsl::user_profiles - .load_iter::(conn)? - .filter_map(|user_profile| user_profile.ok()) - .map(user_profile_from_persistence) - .collect(); - - let object_actions: Vec = schema::object_actions::dsl::object_actions - .load_iter::(conn)? - .filter_map(|object_action| object_action.ok()) // parse into PersistedObjectAction - .filter_map(|action| object_action_from_persisted(action).ok()) - .collect(); - - let server_experiments = schema::server_experiments::dsl::server_experiments - .load_iter::(conn)? - .filter_map(|server_experiment| server_experiment.ok()) - .filter_map(|server_experiment| { - ServerExperiment::from_string(server_experiment.experiment).ok() - }) - .collect(); - - let restored_blocks = get_all_restored_blocks(conn)?; - - // Load active MCP servers from database - let running_mcp_servers = load_active_mcp_servers(conn)?; - - let app_state = AppState { - windows: saved_windows, - active_window_index, - block_lists: Arc::new(restored_blocks), - running_mcp_servers, - }; - - let time_of_next_force_object_refresh = read_time_of_next_force_object_refresh(conn)?; - - let ai_queries = read_ai_queries(conn)?; - - let codebase_indices = get_all_codebase_index_metadata(conn)?; - let workspace_language_servers = get_all_workspace_language_servers_by_workspace(conn)?; - let multi_agent_conversations = read_agent_conversations(conn)?; - let projects = get_all_projects(conn)?; - let project_rules = get_all_project_rules(conn)?; - let ignored_suggestions = get_all_ignored_suggestions(conn)?; - let mcp_server_installations = get_all_mcp_server_installations(conn)?; - let mcp_servers_to_restore = get_mcp_servers_to_restore(conn)?; - - Ok(PersistedData { - app_state, - cloud_objects, - workspaces, - current_workspace_uid, - command_history: commands, - user_profiles, - time_of_next_force_object_refresh, - object_actions, - experiments: server_experiments, - ai_queries, - codebase_indices, - workspace_language_servers, - multi_agent_conversations, - projects, - project_rules, - ignored_suggestions, - mcp_server_installations, - mcp_servers_to_restore, - }) + Ok((workspaces, current_workspace_uid)) } impl From for model::NewCommand { diff --git a/app/src/persistence/sqlite_tests.rs b/app/src/persistence/sqlite_tests.rs index bd3f0b14..17bdc2f2 100644 --- a/app/src/persistence/sqlite_tests.rs +++ b/app/src/persistence/sqlite_tests.rs @@ -5,6 +5,7 @@ use ai::workspace::WorkspaceMetadata; use chrono::Utc; use cloud_object_persistence::to_cloud_object_permissions; use diesel::connection::SimpleConnection; +use galaxy_agent_core::{ConversationMessage, MessageContent, MessageRole}; use galaxy_core::features::FeatureFlag; use galaxy_graphql::scalars::time::ServerTimestamp; use pathfinder_geometry::rect::RectF; @@ -15,6 +16,7 @@ use super::{ encode_path, get_all_codebase_index_metadata, read_sqlite_data, save_app_state, save_codebase_index_metadata, setup_database, start_writer, GALAXY_SQLITE_FILE_NAME, }; +use crate::ai::agent::conversation::{AIConversation, AIConversationId}; use crate::ai::facts::{AIFact, AIMemory, CloudAIFact}; use crate::app_state::{ AppState, CodePaneSnapShot, CodePaneTabSnapshot, LeafContents, LeafSnapshot, PaneNodeSnapshot, @@ -23,7 +25,10 @@ use crate::app_state::{ use crate::cloud_object::{CloudObjectPermissions, Owner}; use crate::code::editor_management::CodeSource; use crate::notebooks::{CloudNotebook, CloudNotebookModel}; -use crate::persistence::model::ObjectPermissions; +use crate::persistence::agent::upsert_agent_conversation; +use crate::persistence::model::{ + AgentConversationData, ObjectPermissions, ProviderConversationContext, +}; use crate::persistence::{BlockCompleted, ModelEvent, PersistenceScope}; use crate::server::ids::{ClientId, SyncId}; use crate::tab::SelectedTabColor; @@ -1125,3 +1130,144 @@ fn test_sqlite_drops_too_small_bounds_on_read() { "tiny persisted bounds must be discarded on read so users recover from a corrupt DB" ); } + +#[test] +fn failed_snapshot_is_not_cached_and_can_be_retried() { + let dir = tempfile::tempdir().unwrap(); + let mut conn = setup_database(&dir.path().join("galaxy.sqlite")).unwrap(); + let snapshot = AppState { + windows: vec![test_terminal_window_snapshot(false)], + active_window_index: Some(0), + block_lists: Default::default(), + running_mcp_servers: Vec::new(), + }; + conn.batch_execute("CREATE TRIGGER reject_window BEFORE INSERT ON windows BEGIN SELECT RAISE(ABORT, 'injected write failure'); END;").unwrap(); + let mut last_saved = None; + assert!(super::save_snapshot_if_changed(&mut conn, snapshot.clone(), &mut last_saved).is_err()); + assert!(last_saved.is_none()); + conn.batch_execute("DROP TRIGGER reject_window;").unwrap(); + super::save_snapshot_if_changed(&mut conn, snapshot.clone(), &mut last_saved).unwrap(); + assert_eq!(last_saved, Some(snapshot)); + assert_eq!( + read_sqlite_data(&mut conn, None) + .unwrap() + .app_state + .windows + .len(), + 1 + ); +} + +#[test] +fn sqlite_busy_writes_retry_but_permanent_failures_do_not() { + let mut attempts = 0; + super::retry_sqlite_write(|| { + attempts += 1; + if attempts < 3 { + return Err(diesel::result::Error::DatabaseError( + diesel::result::DatabaseErrorKind::Unknown, + Box::new("database is locked".to_owned()), + ) + .into()); + } + Ok(()) + }) + .unwrap(); + assert_eq!(attempts, 3); + attempts = 0; + assert!(super::retry_sqlite_write(|| -> anyhow::Result<()> { + attempts += 1; + anyhow::bail!("permanent failure") + }) + .is_err()); + assert_eq!(attempts, 1); +} + +#[test] +fn incomplete_restore_keeps_independent_domains_and_reports_failure() { + let dir = tempfile::tempdir().unwrap(); + let mut conn = setup_database(&dir.path().join("galaxy.sqlite")).unwrap(); + let snapshot = AppState { + windows: vec![test_terminal_window_snapshot(false)], + active_window_index: Some(0), + block_lists: Default::default(), + running_mcp_servers: Vec::new(), + }; + save_app_state(&mut conn, &snapshot).unwrap(); + conn.batch_execute("DROP TABLE ignored_suggestions;") + .unwrap(); + let (restored, errors) = super::read_sqlite_data_with_recovery(&mut conn, None); + assert_eq!(restored.app_state.windows.len(), 1); + assert_eq!(errors.len(), 1); + assert!(errors[0].contains("ignored_suggestions")); +} + +#[test] +fn corrupt_tab_is_reported_instead_of_silently_replacing_saved_layout() { + let dir = tempfile::tempdir().unwrap(); + let mut conn = setup_database(&dir.path().join("galaxy.sqlite")).unwrap(); + let snapshot = AppState { + windows: vec![test_terminal_window_snapshot(false)], + active_window_index: Some(0), + block_lists: Default::default(), + running_mcp_servers: Vec::new(), + }; + save_app_state(&mut conn, &snapshot).unwrap(); + conn.batch_execute("DELETE FROM terminal_panes;").unwrap(); + let (_, errors) = super::read_sqlite_data_with_recovery(&mut conn, None); + assert!(!errors.is_empty()); + assert!(errors.iter().any(|error| error.contains("session layout"))); + // Reading alone must leave the recoverable tab record in place. + conn.batch_execute("CREATE TRIGGER require_saved_tab BEFORE DELETE ON tabs BEGIN SELECT RAISE(ABORT, 'must retain tab'); END;").unwrap(); + assert!(save_app_state(&mut conn, &snapshot).is_err()); +} + +#[test] +fn provider_context_round_trips_through_sqlite_without_terminal_blocks() { + let dir = tempfile::tempdir().unwrap(); + let mut conn = setup_database(&dir.path().join("galaxy.sqlite")).unwrap(); + let id = AIConversationId::new(); + let context = ProviderConversationContext { + messages: vec![ConversationMessage { + role: MessageRole::Assistant, + content: MessageContent::Text("Most recent response".into()), + }], + tool_result_archive: vec![ConversationMessage { + role: MessageRole::User, + content: MessageContent::ToolResult { + tool_use_id: "old-call".into(), + content: "archived output".into(), + is_error: false, + }, + }], + }; + upsert_agent_conversation( + &mut conn, + &id.to_string(), + &[], + AgentConversationData { + provider_context: Some(context.clone()), + progressive_summary: Some("Earlier conversation".into()), + ..Default::default() + }, + ) + .unwrap(); + conn.batch_execute("INSERT INTO agent_conversations (conversation_id, conversation_data) VALUES ('corrupt', 'not-json');").unwrap(); + let (healthy, errors) = super::read_sqlite_data_with_recovery(&mut conn, None); + assert_eq!(healthy.multi_agent_conversations.len(), 1); + assert_eq!(errors.len(), 1); + assert!(errors[0].contains("corrupt")); + conn.batch_execute("DELETE FROM agent_conversations WHERE conversation_id = 'corrupt';") + .unwrap(); + let mut data = read_sqlite_data(&mut conn, None).unwrap(); + assert!(data.app_state.block_lists.is_empty()); + let saved = data.multi_agent_conversations.pop().unwrap(); + let restored = AIConversation::new_restored_synthesizing_on_empty( + id, + saved.tasks, + Some(serde_json::from_str(&saved.conversation.conversation_data).unwrap()), + ) + .unwrap(); + assert_eq!(restored.provider_context(), context); + assert_eq!(restored.progressive_summary(), Some("Earlier conversation")); +} diff --git a/app/src/terminal/view/load_ai_conversation.rs b/app/src/terminal/view/load_ai_conversation.rs index 470619e2..0917f543 100644 --- a/app/src/terminal/view/load_ai_conversation.rs +++ b/app/src/terminal/view/load_ai_conversation.rs @@ -943,6 +943,7 @@ impl TerminalView { let conversation_data = AgentConversationData { agent_backend: Default::default(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: Default::default(), diff --git a/app/src/workspace/global_actions.rs b/app/src/workspace/global_actions.rs index 69d523af..cfcc5a8a 100644 --- a/app/src/workspace/global_actions.rs +++ b/app/src/workspace/global_actions.rs @@ -10,13 +10,12 @@ use galaxyui::{AppContext, Entity, ModelContext, SingletonEntity, TypedActionVie use crate::ai::agent::conversation::AIConversationId; use crate::ai::agent::AIAgentExchangeId; -use crate::app_state::{get_app_state, AppState}; +use crate::app_state::get_app_state; use crate::network::NetworkStatus; use crate::persistence::ModelEvent; use crate::root_view::OpenPath; use crate::server::server_api::ServerApiProvider; use crate::terminal::alt_screen_reporting::AltScreenReporting; -use crate::terminal::general_settings::GeneralSettings; use crate::undo_close::UndoCloseStack; use crate::workspace::cross_window_tab_drag::CrossWindowTabDrag; use crate::workspace::{Workspace, WorkspaceAction}; @@ -25,7 +24,6 @@ use crate::{auth, GlobalResourceHandlesProvider}; const SESSION_SAVE_DEBOUNCE: Duration = Duration::from_millis(250); pub(crate) struct SessionSaveCoordinator { - last_enqueued_state: Option, pending_save: bool, timer_generation: u64, timer: Option, @@ -34,7 +32,6 @@ pub(crate) struct SessionSaveCoordinator { impl SessionSaveCoordinator { pub(crate) fn new() -> Self { Self { - last_enqueued_state: None, pending_save: false, timer_generation: 0, timer: None, @@ -47,7 +44,7 @@ impl SessionSaveCoordinator { } if self.timer.is_none() { - self.save_if_changed(ctx); + self.enqueue_snapshot(ctx); } else { self.pending_save = true; } @@ -62,7 +59,7 @@ impl SessionSaveCoordinator { self.pending_save = false; if session_save_is_enabled(ctx) { - self.save_if_changed(ctx); + self.enqueue_snapshot(ctx); } } @@ -80,13 +77,13 @@ impl SessionSaveCoordinator { } coordinator.timer = None; if std::mem::take(&mut coordinator.pending_save) { - coordinator.save_if_changed(ctx); + coordinator.enqueue_snapshot(ctx); } }, )); } - fn save_if_changed(&mut self, ctx: &mut ModelContext) { + fn enqueue_snapshot(&mut self, ctx: &mut ModelContext) { let Some(model_event_sender) = GlobalResourceHandlesProvider::as_ref(ctx) .get() .model_event_sender @@ -95,14 +92,9 @@ impl SessionSaveCoordinator { return; }; let app_state = get_app_state(ctx); - if self.last_enqueued_state.as_ref() == Some(&app_state) { - return; - } - if let Err(err) = model_event_sender.send(ModelEvent::Snapshot(app_state.clone())) { + if let Err(err) = model_event_sender.send(ModelEvent::Snapshot(app_state)) { log::error!("Error trying to send model event {err:?}"); - return; } - self.last_enqueued_state = Some(app_state); } } @@ -113,9 +105,7 @@ impl Entity for SessionSaveCoordinator { impl SingletonEntity for SessionSaveCoordinator {} fn session_save_is_enabled(ctx: &AppContext) -> bool { - AppExecutionMode::as_ref(ctx).can_save_session() - && *GeneralSettings::as_ref(ctx).restore_session - && !CrossWindowTabDrag::as_ref(ctx).is_active() + AppExecutionMode::as_ref(ctx).can_save_session() && !CrossWindowTabDrag::as_ref(ctx).is_active() } /// Specifies where a forked conversation should be opened. diff --git a/crates/persistence/Cargo.toml b/crates/persistence/Cargo.toml index 0d0a92fa..6a115f57 100644 --- a/crates/persistence/Cargo.toml +++ b/crates/persistence/Cargo.toml @@ -9,6 +9,7 @@ license.workspace = true local_fs = [] [dependencies] +galaxy_agent_core.workspace = true chrono.workspace = true diesel = { workspace = true, features = [ "chrono", diff --git a/crates/persistence/src/model.rs b/crates/persistence/src/model.rs index df6942c6..e554712d 100644 --- a/crates/persistence/src/model.rs +++ b/crates/persistence/src/model.rs @@ -1086,8 +1086,19 @@ pub struct LocalAgentGroup { pub notified: bool, } +/// Exact provider history, independent of terminal scrollback and UI task projections. +/// `None` on AgentConversationData denotes a legacy row requiring reconstruction; +/// an explicitly empty history must stay empty after restoration. +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)] +pub struct ProviderConversationContext { + pub messages: Vec, + pub tool_result_archive: Vec, +} + #[derive(Debug, Serialize, Deserialize, Clone, Default)] pub struct AgentConversationData { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub provider_context: Option, /// Display-only tool intent, scoped to this conversation and keyed by call ID. #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub tool_titles: HashMap, diff --git a/crates/persistence/src/model_tests.rs b/crates/persistence/src/model_tests.rs index f763c8cd..4bff7a38 100644 --- a/crates/persistence/src/model_tests.rs +++ b/crates/persistence/src/model_tests.rs @@ -137,6 +137,7 @@ fn is_restorable_accepts_empty_and_single_task_conversations() { fn agent_conversation_data_roundtrips_last_event_sequence() { let data = AgentConversationData { agent_backend: Default::default(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: Default::default(), @@ -242,6 +243,7 @@ fn agent_conversation_data_omits_default_provider_backend() { fn agent_conversation_data_roundtrips_remote_child_marker() { let data = AgentConversationData { agent_backend: Default::default(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: Default::default(), @@ -272,6 +274,7 @@ fn agent_conversation_data_roundtrips_remote_child_marker() { fn agent_conversation_data_roundtrips_optimistic_root_marker() { let data = AgentConversationData { agent_backend: Default::default(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: Default::default(), @@ -314,6 +317,7 @@ fn agent_conversation_data_deserializes_legacy_payload_without_last_event_sequen fn agent_conversation_data_skips_serializing_none_last_event_sequence() { let data = AgentConversationData { agent_backend: Default::default(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: Default::default(), @@ -346,6 +350,7 @@ fn agent_conversation_data_skips_serializing_none_last_event_sequence() { fn agent_conversation_data_roundtrips_pinned() { let data = AgentConversationData { agent_backend: Default::default(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: Default::default(), @@ -376,6 +381,7 @@ fn agent_conversation_data_roundtrips_pinned() { fn agent_conversation_data_skips_serializing_unpinned() { let data = AgentConversationData { agent_backend: Default::default(), + provider_context: None, active_provider_run_json: None, local_agent_groups: Vec::new(), tool_titles: Default::default(),