use crate::{AgentMessage, AgentMessageContent, UserMessage, UserMessageContent}; use acp_thread::UserMessageId; use agent_client_protocol as acp; use agent_settings::AgentProfileId; use anyhow::{Result, anyhow}; use chrono::{DateTime, Utc}; use collections::{HashMap, IndexMap}; use futures::{FutureExt, future::Shared}; use gpui::{BackgroundExecutor, Global, Task}; use indoc::indoc; use parking_lot::Mutex; use serde::{Deserialize, Serialize}; use sqlez::{ bindable::{Bind, Column}, connection::Connection, statement::Statement, }; use std::sync::Arc; use ui::{App, SharedString}; use zed_env_vars::ZED_STATELESS; pub type DbMessage = crate::Message; pub type DbSummary = crate::legacy_thread::DetailedSummaryState; pub type DbLanguageModel = crate::legacy_thread::SerializedLanguageModel; /// Metadata about the git worktree associated with an agent thread. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AgentGitWorktreeInfo { /// The branch name in the git worktree. pub branch: String, /// Absolute path to the git worktree on disk. pub worktree_path: std::path::PathBuf, /// The base branch/commit the worktree was created from. pub base_ref: String, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct DbThreadMetadata { pub id: acp::SessionId, pub parent_session_id: Option, #[serde(alias = "summary")] pub title: SharedString, pub updated_at: DateTime, /// Denormalized from `DbThread::git_worktree_info.branch` for efficient /// listing without decompressing thread data. The blob is the source of /// truth; this column is populated on save for query convenience. pub worktree_branch: Option, } #[derive(Debug, Serialize, Deserialize)] pub struct DbThread { pub title: SharedString, pub messages: Vec, pub updated_at: DateTime, #[serde(default)] pub detailed_summary: Option, #[serde(default)] pub initial_project_snapshot: Option>, #[serde(default)] pub cumulative_token_usage: language_model::TokenUsage, #[serde(default)] pub request_token_usage: HashMap, #[serde(default)] pub model: Option, #[serde(default)] pub profile: Option, #[serde(default)] pub imported: bool, #[serde(default)] pub subagent_context: Option, #[serde(default)] pub git_worktree_info: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct SharedThread { pub title: SharedString, pub messages: Vec, pub updated_at: DateTime, #[serde(default)] pub model: Option, pub version: String, } impl SharedThread { pub const VERSION: &'static str = "1.0.0"; pub fn from_db_thread(thread: &DbThread) -> Self { Self { title: thread.title.clone(), messages: thread.messages.clone(), updated_at: thread.updated_at, model: thread.model.clone(), version: Self::VERSION.to_string(), } } pub fn to_db_thread(self) -> DbThread { DbThread { title: format!("🔗 {}", self.title).into(), messages: self.messages, updated_at: self.updated_at, detailed_summary: None, initial_project_snapshot: None, cumulative_token_usage: Default::default(), request_token_usage: Default::default(), model: self.model, profile: None, imported: true, subagent_context: None, git_worktree_info: None, } } pub fn to_bytes(&self) -> Result> { const COMPRESSION_LEVEL: i32 = 3; let json = serde_json::to_vec(self)?; let compressed = zstd::encode_all(json.as_slice(), COMPRESSION_LEVEL)?; Ok(compressed) } pub fn from_bytes(data: &[u8]) -> Result { let decompressed = zstd::decode_all(data)?; Ok(serde_json::from_slice(&decompressed)?) } } impl DbThread { pub const VERSION: &'static str = "0.3.0"; pub fn from_json(json: &[u8]) -> Result { let saved_thread_json = serde_json::from_slice::(json)?; match saved_thread_json.get("version") { Some(serde_json::Value::String(version)) => match version.as_str() { Self::VERSION => Ok(serde_json::from_value(saved_thread_json)?), _ => Self::upgrade_from_agent_1(crate::legacy_thread::SerializedThread::from_json( json, )?), }, _ => { Self::upgrade_from_agent_1(crate::legacy_thread::SerializedThread::from_json(json)?) } } } fn upgrade_from_agent_1(thread: crate::legacy_thread::SerializedThread) -> Result { let mut messages = Vec::new(); let mut request_token_usage = HashMap::default(); let mut last_user_message_id = None; for (ix, msg) in thread.messages.into_iter().enumerate() { let message = match msg.role { language_model::Role::User => { let mut content = Vec::new(); // Convert segments to content for segment in msg.segments { match segment { crate::legacy_thread::SerializedMessageSegment::Text { text } => { content.push(UserMessageContent::Text(text)); } crate::legacy_thread::SerializedMessageSegment::Thinking { text, .. } => { // User messages don't have thinking segments, but handle gracefully content.push(UserMessageContent::Text(text)); } crate::legacy_thread::SerializedMessageSegment::RedactedThinking { .. } => { // User messages don't have redacted thinking, skip. } } } // If no content was added, add context as text if available if content.is_empty() && !msg.context.is_empty() { content.push(UserMessageContent::Text(msg.context)); } let id = UserMessageId::new(); last_user_message_id = Some(id.clone()); crate::Message::User(UserMessage { // MessageId from old format can't be meaningfully converted, so generate a new one id, content, }) } language_model::Role::Assistant => { let mut content = Vec::new(); // Convert segments to content for segment in msg.segments { match segment { crate::legacy_thread::SerializedMessageSegment::Text { text } => { content.push(AgentMessageContent::Text(text)); } crate::legacy_thread::SerializedMessageSegment::Thinking { text, signature, } => { content.push(AgentMessageContent::Thinking { text, signature }); } crate::legacy_thread::SerializedMessageSegment::RedactedThinking { data, } => { content.push(AgentMessageContent::RedactedThinking(data)); } } } // Convert tool uses let mut tool_names_by_id = HashMap::default(); for tool_use in msg.tool_uses { tool_names_by_id.insert(tool_use.id.clone(), tool_use.name.clone()); content.push(AgentMessageContent::ToolUse( language_model::LanguageModelToolUse { id: tool_use.id, name: tool_use.name.into(), raw_input: serde_json::to_string(&tool_use.input) .unwrap_or_default(), input: tool_use.input, is_input_complete: true, thought_signature: None, }, )); } // Convert tool results let mut tool_results = IndexMap::default(); for tool_result in msg.tool_results { let name = tool_names_by_id .remove(&tool_result.tool_use_id) .unwrap_or_else(|| SharedString::from("unknown")); tool_results.insert( tool_result.tool_use_id.clone(), language_model::LanguageModelToolResult { tool_use_id: tool_result.tool_use_id, tool_name: name.into(), is_error: tool_result.is_error, content: tool_result.content, output: tool_result.output, }, ); } if let Some(last_user_message_id) = &last_user_message_id && let Some(token_usage) = thread.request_token_usage.get(ix).copied() { request_token_usage.insert(last_user_message_id.clone(), token_usage); } crate::Message::Agent(AgentMessage { content, tool_results, reasoning_details: None, }) } language_model::Role::System => { // Skip system messages as they're not supported in the new format continue; } }; messages.push(message); } Ok(Self { title: thread.summary, messages, updated_at: thread.updated_at, detailed_summary: match thread.detailed_summary_state { crate::legacy_thread::DetailedSummaryState::NotGenerated | crate::legacy_thread::DetailedSummaryState::Generating => None, crate::legacy_thread::DetailedSummaryState::Generated { text, .. } => Some(text), }, initial_project_snapshot: thread.initial_project_snapshot, cumulative_token_usage: thread.cumulative_token_usage, request_token_usage, model: thread.model, profile: thread.profile, imported: false, subagent_context: None, git_worktree_info: None, }) } } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub enum DataType { #[serde(rename = "json")] Json, #[serde(rename = "zstd")] Zstd, } impl Bind for DataType { fn bind(&self, statement: &Statement, start_index: i32) -> Result { let value = match self { DataType::Json => "json", DataType::Zstd => "zstd", }; value.bind(statement, start_index) } } impl Column for DataType { fn column(statement: &mut Statement, start_index: i32) -> Result<(Self, i32)> { let (value, next_index) = String::column(statement, start_index)?; let data_type = match value.as_str() { "json" => DataType::Json, "zstd" => DataType::Zstd, _ => anyhow::bail!("Unknown data type: {}", value), }; Ok((data_type, next_index)) } } pub(crate) struct ThreadsDatabase { executor: BackgroundExecutor, connection: Arc>, } struct GlobalThreadsDatabase(Shared, Arc>>>); impl Global for GlobalThreadsDatabase {} impl ThreadsDatabase { pub fn connect(cx: &mut App) -> Shared, Arc>>> { if cx.has_global::() { return cx.global::().0.clone(); } let executor = cx.background_executor().clone(); let task = executor .spawn({ let executor = executor.clone(); async move { match ThreadsDatabase::new(executor) { Ok(db) => Ok(Arc::new(db)), Err(err) => Err(Arc::new(err)), } } }) .shared(); cx.set_global(GlobalThreadsDatabase(task.clone())); task } pub fn new(executor: BackgroundExecutor) -> Result { let connection = if *ZED_STATELESS { Connection::open_memory(Some("THREAD_FALLBACK_DB")) } else if cfg!(any(feature = "test-support", test)) { // rust stores the name of the test on the current thread. // We use this to automatically create a database that will // be shared within the test (for the test_retrieve_old_thread) // but not with concurrent tests. let thread = std::thread::current(); let test_name = thread.name(); Connection::open_memory(Some(&format!( "THREAD_FALLBACK_{}", test_name.unwrap_or_default() ))) } else { let threads_dir = paths::data_dir().join("threads"); std::fs::create_dir_all(&threads_dir)?; let sqlite_path = threads_dir.join("threads.db"); Connection::open_file(&sqlite_path.to_string_lossy()) }; connection.exec(indoc! {" CREATE TABLE IF NOT EXISTS threads ( id TEXT PRIMARY KEY, summary TEXT NOT NULL, updated_at TEXT NOT NULL, data_type TEXT NOT NULL, data BLOB NOT NULL ) "})?() .map_err(|e| anyhow!("Failed to create threads table: {}", e))?; if let Ok(mut s) = connection.exec(indoc! {" ALTER TABLE threads ADD COLUMN parent_id TEXT "}) { s().ok(); } if let Ok(mut s) = connection.exec(indoc! {" ALTER TABLE threads ADD COLUMN worktree_branch TEXT "}) { s().ok(); } let db = Self { executor, connection: Arc::new(Mutex::new(connection)), }; Ok(db) } fn save_thread_sync( connection: &Arc>, id: acp::SessionId, thread: DbThread, ) -> Result<()> { const COMPRESSION_LEVEL: i32 = 3; #[derive(Serialize)] struct SerializedThread { #[serde(flatten)] thread: DbThread, version: &'static str, } let title = thread.title.to_string(); let updated_at = thread.updated_at.to_rfc3339(); let parent_id = thread .subagent_context .as_ref() .map(|ctx| ctx.parent_thread_id.0.clone()); let worktree_branch = thread .git_worktree_info .as_ref() .map(|info| info.branch.clone()); let json_data = serde_json::to_string(&SerializedThread { thread, version: DbThread::VERSION, })?; let connection = connection.lock(); let compressed = zstd::encode_all(json_data.as_bytes(), COMPRESSION_LEVEL)?; let data_type = DataType::Zstd; let data = compressed; let mut insert = connection.exec_bound::<(Arc, Option>, Option, String, String, DataType, Vec)>(indoc! {" INSERT OR REPLACE INTO threads (id, parent_id, worktree_branch, summary, updated_at, data_type, data) VALUES (?, ?, ?, ?, ?, ?, ?) "})?; insert(( id.0, parent_id, worktree_branch, title, updated_at, data_type, data, ))?; Ok(()) } pub fn list_threads(&self) -> Task>> { let connection = self.connection.clone(); self.executor.spawn(async move { let connection = connection.lock(); let mut select = connection .select_bound::<(), (Arc, Option>, Option, String, String)>(indoc! {" SELECT id, parent_id, worktree_branch, summary, updated_at FROM threads ORDER BY updated_at DESC "})?; let rows = select(())?; let mut threads = Vec::new(); for (id, parent_id, worktree_branch, summary, updated_at) in rows { threads.push(DbThreadMetadata { id: acp::SessionId::new(id), parent_session_id: parent_id.map(acp::SessionId::new), title: summary.into(), updated_at: DateTime::parse_from_rfc3339(&updated_at)?.with_timezone(&Utc), worktree_branch, }); } Ok(threads) }) } pub fn load_thread(&self, id: acp::SessionId) -> Task>> { let connection = self.connection.clone(); self.executor.spawn(async move { let connection = connection.lock(); let mut select = connection.select_bound::, (DataType, Vec)>(indoc! {" SELECT data_type, data FROM threads WHERE id = ? LIMIT 1 "})?; let rows = select(id.0)?; if let Some((data_type, data)) = rows.into_iter().next() { let json_data = match data_type { DataType::Zstd => { let decompressed = zstd::decode_all(&data[..])?; String::from_utf8(decompressed)? } DataType::Json => String::from_utf8(data)?, }; let thread = DbThread::from_json(json_data.as_bytes())?; Ok(Some(thread)) } else { Ok(None) } }) } pub fn save_thread(&self, id: acp::SessionId, thread: DbThread) -> Task> { let connection = self.connection.clone(); self.executor .spawn(async move { Self::save_thread_sync(&connection, id, thread) }) } pub fn delete_thread(&self, id: acp::SessionId) -> Task> { let connection = self.connection.clone(); self.executor.spawn(async move { let connection = connection.lock(); let mut delete = connection.exec_bound::>(indoc! {" DELETE FROM threads WHERE id = ? "})?; delete(id.0)?; Ok(()) }) } pub fn delete_threads(&self) -> Task> { let connection = self.connection.clone(); self.executor.spawn(async move { let connection = connection.lock(); let mut delete = connection.exec_bound::<()>(indoc! {" DELETE FROM threads "})?; delete(())?; Ok(()) }) } } #[cfg(test)] mod tests { use super::*; use chrono::{DateTime, TimeZone, Utc}; use collections::HashMap; use gpui::TestAppContext; use std::sync::Arc; #[test] fn test_shared_thread_roundtrip() { let original = SharedThread { title: "Test Thread".into(), messages: vec![], updated_at: Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(), model: None, version: SharedThread::VERSION.to_string(), }; let bytes = original.to_bytes().expect("Failed to serialize"); let restored = SharedThread::from_bytes(&bytes).expect("Failed to deserialize"); assert_eq!(restored.title, original.title); assert_eq!(restored.version, original.version); assert_eq!(restored.updated_at, original.updated_at); } #[test] fn test_imported_flag_defaults_to_false() { // Simulate deserializing a thread without the imported field (backwards compatibility). let json = r#"{ "title": "Old Thread", "messages": [], "updated_at": "2024-01-01T00:00:00Z" }"#; let db_thread: DbThread = serde_json::from_str(json).expect("Failed to deserialize"); assert!( !db_thread.imported, "Legacy threads without imported field should default to false" ); } fn session_id(value: &str) -> acp::SessionId { acp::SessionId::new(Arc::::from(value)) } fn make_thread(title: &str, updated_at: DateTime) -> DbThread { DbThread { title: title.to_string().into(), messages: Vec::new(), updated_at, detailed_summary: None, initial_project_snapshot: None, cumulative_token_usage: Default::default(), request_token_usage: HashMap::default(), model: None, profile: None, imported: false, subagent_context: None, git_worktree_info: None, } } #[gpui::test] async fn test_list_threads_orders_by_updated_at(cx: &mut TestAppContext) { let database = ThreadsDatabase::new(cx.executor()).unwrap(); let older_id = session_id("thread-a"); let newer_id = session_id("thread-b"); let older_thread = make_thread( "Thread A", Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(), ); let newer_thread = make_thread( "Thread B", Utc.with_ymd_and_hms(2024, 1, 2, 0, 0, 0).unwrap(), ); database .save_thread(older_id.clone(), older_thread) .await .unwrap(); database .save_thread(newer_id.clone(), newer_thread) .await .unwrap(); let entries = database.list_threads().await.unwrap(); assert_eq!(entries.len(), 2); assert_eq!(entries[0].id, newer_id); assert_eq!(entries[1].id, older_id); } #[gpui::test] async fn test_save_thread_replaces_metadata(cx: &mut TestAppContext) { let database = ThreadsDatabase::new(cx.executor()).unwrap(); let thread_id = session_id("thread-a"); let original_thread = make_thread( "Thread A", Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(), ); let updated_thread = make_thread( "Thread B", Utc.with_ymd_and_hms(2024, 1, 2, 0, 0, 0).unwrap(), ); database .save_thread(thread_id.clone(), original_thread) .await .unwrap(); database .save_thread(thread_id.clone(), updated_thread) .await .unwrap(); let entries = database.list_threads().await.unwrap(); assert_eq!(entries.len(), 1); assert_eq!(entries[0].id, thread_id); assert_eq!(entries[0].title.as_ref(), "Thread B"); assert_eq!( entries[0].updated_at, Utc.with_ymd_and_hms(2024, 1, 2, 0, 0, 0).unwrap() ); } #[test] fn test_subagent_context_defaults_to_none() { let json = r#"{ "title": "Old Thread", "messages": [], "updated_at": "2024-01-01T00:00:00Z" }"#; let db_thread: DbThread = serde_json::from_str(json).expect("Failed to deserialize"); assert!( db_thread.subagent_context.is_none(), "Legacy threads without subagent_context should default to None" ); } #[gpui::test] async fn test_subagent_context_roundtrips_through_save_load(cx: &mut TestAppContext) { let database = ThreadsDatabase::new(cx.executor()).unwrap(); let parent_id = session_id("parent-thread"); let child_id = session_id("child-thread"); let mut child_thread = make_thread( "Subagent Thread", Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(), ); child_thread.subagent_context = Some(crate::SubagentContext { parent_thread_id: parent_id.clone(), depth: 2, }); database .save_thread(child_id.clone(), child_thread) .await .unwrap(); let loaded = database .load_thread(child_id) .await .unwrap() .expect("thread should exist"); let context = loaded .subagent_context .expect("subagent_context should be restored"); assert_eq!(context.parent_thread_id, parent_id); assert_eq!(context.depth, 2); } #[gpui::test] async fn test_non_subagent_thread_has_no_subagent_context(cx: &mut TestAppContext) { let database = ThreadsDatabase::new(cx.executor()).unwrap(); let thread_id = session_id("regular-thread"); let thread = make_thread( "Regular Thread", Utc.with_ymd_and_hms(2024, 1, 1, 0, 0, 0).unwrap(), ); database .save_thread(thread_id.clone(), thread) .await .unwrap(); let loaded = database .load_thread(thread_id) .await .unwrap() .expect("thread should exist"); assert!( loaded.subagent_context.is_none(), "Regular threads should have no subagent_context" ); } #[gpui::test] async fn test_git_worktree_info_roundtrip(cx: &mut TestAppContext) { let database = ThreadsDatabase::new(cx.executor()).unwrap(); let thread_id = session_id("worktree-thread"); let mut thread = make_thread( "Worktree Thread", Utc.with_ymd_and_hms(2024, 6, 15, 12, 0, 0).unwrap(), ); thread.git_worktree_info = Some(AgentGitWorktreeInfo { branch: "zed/agent/a4Xiu".to_string(), worktree_path: std::path::PathBuf::from("/repo/worktrees/zed/agent/a4Xiu"), base_ref: "main".to_string(), }); database .save_thread(thread_id.clone(), thread) .await .unwrap(); let loaded = database .load_thread(thread_id) .await .unwrap() .expect("thread should exist"); let info = loaded .git_worktree_info .expect("git_worktree_info should be restored"); assert_eq!(info.branch, "zed/agent/a4Xiu"); assert_eq!( info.worktree_path, std::path::PathBuf::from("/repo/worktrees/zed/agent/a4Xiu") ); assert_eq!(info.base_ref, "main"); } #[gpui::test] async fn test_session_list_includes_worktree_meta(cx: &mut TestAppContext) { let database = ThreadsDatabase::new(cx.executor()).unwrap(); // Save a thread with worktree info let worktree_id = session_id("wt-thread"); let mut worktree_thread = make_thread( "With Worktree", Utc.with_ymd_and_hms(2024, 6, 15, 12, 0, 0).unwrap(), ); worktree_thread.git_worktree_info = Some(AgentGitWorktreeInfo { branch: "zed/agent/bR9kz".to_string(), worktree_path: std::path::PathBuf::from("/repo/worktrees/zed/agent/bR9kz"), base_ref: "develop".to_string(), }); database .save_thread(worktree_id.clone(), worktree_thread) .await .unwrap(); // Save a thread without worktree info let plain_id = session_id("plain-thread"); let plain_thread = make_thread( "Without Worktree", Utc.with_ymd_and_hms(2024, 6, 15, 11, 0, 0).unwrap(), ); database .save_thread(plain_id.clone(), plain_thread) .await .unwrap(); // List threads and verify worktree_branch is populated correctly let threads = database.list_threads().await.unwrap(); assert_eq!(threads.len(), 2); let wt_entry = threads .iter() .find(|t| t.id == worktree_id) .expect("should find worktree thread"); assert_eq!(wt_entry.worktree_branch.as_deref(), Some("zed/agent/bR9kz")); let plain_entry = threads .iter() .find(|t| t.id == plain_id) .expect("should find plain thread"); assert!( plain_entry.worktree_branch.is_none(), "plain thread should have no worktree_branch" ); } }