diff --git a/src-tauri/src/commands/conversations.rs b/src-tauri/src/commands/conversations.rs index 23ba28b1c..a25b0eb2a 100644 --- a/src-tauri/src/commands/conversations.rs +++ b/src-tauri/src/commands/conversations.rs @@ -1,5 +1,8 @@ use std::collections::{HashMap, HashSet}; +#[cfg(feature = "tauri-runtime")] +use tauri::Manager; + use crate::app_error::AppCommandError; use crate::db::entities::conversation; use crate::db::entities::folder::FolderKind; @@ -31,15 +34,69 @@ use crate::web::event_bridge::{ IMPORT_SCAN_PROGRESS_EVENT, TABS_CHANGED_EVENT, }; -pub async fn list_all_conversations_core( +#[derive(Default)] +pub(crate) struct ListAllConversationsOptions { + pub(crate) folder_ids: Option>, + pub(crate) agent_type: Option, + pub(crate) search: Option, + pub(crate) sort_by: Option, + pub(crate) status: Option, + pub(crate) include_children: bool, +} + +pub(crate) async fn list_all_conversations_core( conn: &sea_orm::DatabaseConnection, - folder_ids: Option>, - agent_type: Option, - search: Option, - sort_by: Option, - status: Option, - include_children: bool, + emitter: &EventEmitter, + chat_channel_manager: &crate::chat_channel::manager::ChatChannelManager, + options: ListAllConversationsOptions, +) -> Result, AppCommandError> { + let codex_titles = match tokio::task::spawn_blocking(|| { + CodexParser::new().load_thread_name_index() + }) + .await + { + Ok(titles) => titles, + Err(error) => { + tracing::warn!( + error = %error, + "conversation list: failed to load Codex session titles; continuing without refresh" + ); + HashMap::new() + } + }; + list_all_conversations_core_with_codex_titles( + conn, + emitter, + chat_channel_manager, + options, + &codex_titles, + ) + .await +} + +async fn list_all_conversations_core_with_codex_titles( + conn: &sea_orm::DatabaseConnection, + emitter: &EventEmitter, + chat_channel_manager: &crate::chat_channel::manager::ChatChannelManager, + options: ListAllConversationsOptions, + codex_titles: &HashMap, ) -> Result, AppCommandError> { + // Synchronize before `list_all` builds any folder/agent/search/status + // filters so a freshly generated Codex title is visible on this same call. + let refreshed_ids = conversation_service::refresh_codex_auto_titles(conn, codex_titles).await; + // Detached on purpose — see `notify_conversation_title_updates`. The list + // must not wait on Telegram. + drop( + notify_conversation_title_updates(conn, emitter, chat_channel_manager, refreshed_ids).await, + ); + let ListAllConversationsOptions { + folder_ids, + agent_type, + search, + sort_by, + status, + include_children, + } = options; conversation_service::list_all( conn, folder_ids, @@ -56,7 +113,7 @@ pub async fn list_all_conversations_core( #[cfg(feature = "tauri-runtime")] #[cfg_attr(feature = "tauri-runtime", tauri::command)] pub async fn list_all_conversations( - db: tauri::State<'_, AppDatabase>, + app: tauri::AppHandle, folder_ids: Option>, agent_type: Option, search: Option, @@ -64,14 +121,22 @@ pub async fn list_all_conversations( status: Option, include_children: Option, ) -> Result, AppCommandError> { + let emitter = EventEmitter::Tauri(app.clone()); + let db = app.state::(); + let chat_channel_manager = + app.state::(); list_all_conversations_core( &db.conn, - folder_ids, - agent_type, - search, - sort_by, - status, - include_children.unwrap_or(false), + &emitter, + &chat_channel_manager, + ListAllConversationsOptions { + folder_ids, + agent_type, + search, + sort_by, + status, + include_children: include_children.unwrap_or(false), + }, ) .await } @@ -390,6 +455,7 @@ fn compute_folders(all_conversations: &[ConversationSummary]) -> Vec pub async fn import_local_conversations_core( conn: &sea_orm::DatabaseConnection, emitter: &EventEmitter, + chat_channel_manager: &crate::chat_channel::manager::ChatChannelManager, folder_id: i32, ) -> Result { // Share IMPORT_GUARD with the batch importer: `(external_id, agent_type)` @@ -415,11 +481,13 @@ pub async fn import_local_conversations_core( .map_err(AppCommandError::from)?; // Broadcast a sidebar upsert for every title refreshed in place, so other - // windows and web clients converge live. The importing client refetches the - // list itself, which also covers the newly imported rows. - for id in updated_ids { - emit_conversation_upsert(emitter, conn, id).await; - } + // windows and web clients converge live, and propagate the new name to any + // bound chat thread — the same treatment the scan and list paths give a + // title discovered outside codeg. The importing client refetches the list + // itself, which also covers the newly imported rows. + drop( + notify_conversation_title_updates(conn, emitter, chat_channel_manager, updated_ids).await, + ); Ok(result) } @@ -429,9 +497,16 @@ pub async fn import_local_conversations_core( pub async fn import_local_conversations( app: tauri::AppHandle, db: tauri::State<'_, AppDatabase>, + chat_channel_manager: tauri::State<'_, crate::chat_channel::manager::ChatChannelManager>, folder_id: i32, ) -> Result { - import_local_conversations_core(&db.conn, &EventEmitter::Tauri(app), folder_id).await + import_local_conversations_core( + &db.conn, + &EventEmitter::Tauri(app), + &chat_channel_manager, + folder_id, + ) + .await } /// Serializes concurrent batch imports: `(external_id, agent_type)` has no DB @@ -617,9 +692,8 @@ fn build_scan_result( pub async fn scan_importable_sessions_core( conn: &sea_orm::DatabaseConnection, emitter: &EventEmitter, + chat_channel_manager: &crate::chat_channel::manager::ChatChannelManager, ) -> Result { - use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; - let progress_emitter = emitter.clone(); let summaries = import_service::collect_local_summaries(move |agent_type, done, total, session_count| { @@ -636,6 +710,20 @@ pub async fn scan_importable_sessions_core( }) .await; + scan_importable_sessions_from_summaries(conn, emitter, chat_channel_manager, summaries).await +} + +/// Reconcile summaries already collected by the filesystem scan. Keeping this +/// boundary separate makes the DB refresh and notification behavior testable +/// without reading the developer's real agent session directories. +async fn scan_importable_sessions_from_summaries( + conn: &sea_orm::DatabaseConnection, + emitter: &EventEmitter, + chat_channel_manager: &crate::chat_channel::manager::ChatChannelManager, + summaries: Vec<(AgentType, ConversationSummary)>, +) -> Result { + use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; + let conv_rows = conversation::Entity::find() .filter(conversation::Column::ExternalId.is_not_null()) .all(conn) @@ -655,10 +743,17 @@ pub async fn scan_importable_sessions_core( } // Refresh the already-imported rows in place before answering, then - // broadcast each one so open sidebars re-sort without a refetch. - for id in import_service::sync_imported_sessions(conn, &conv_rows, &summaries).await { - emit_conversation_upsert(emitter, conn, id).await; - } + // broadcast each one so open sidebars re-sort without a refetch. The + // chat-channel half runs detached so the scan never waits on Telegram. + drop( + notify_conversation_title_updates( + conn, + emitter, + chat_channel_manager, + import_service::sync_imported_sessions(conn, &conv_rows, &summaries).await, + ) + .await, + ); let folder_rows = load_folder_rows(conn).await?; Ok(build_scan_result(summaries, &imported_index, &folder_rows)) @@ -669,8 +764,9 @@ pub async fn scan_importable_sessions_core( pub async fn scan_importable_sessions( app: tauri::AppHandle, db: tauri::State<'_, AppDatabase>, + chat_channel_manager: tauri::State<'_, crate::chat_channel::manager::ChatChannelManager>, ) -> Result { - scan_importable_sessions_core(&db.conn, &EventEmitter::Tauri(app)).await + scan_importable_sessions_core(&db.conn, &EventEmitter::Tauri(app), &chat_channel_manager).await } /// Batch-import the selected sessions, creating (or reopening) each target @@ -1511,6 +1607,102 @@ pub(crate) async fn emit_conversation_upsert( } } +/// Push one conversation's CURRENT title to its bound chat threads, then +/// confirm it is still current — re-sending if it is not. +/// +/// The confirmation loop is what makes a detached sync safe. A provider edit is +/// a remote call that can land arbitrarily late, so two syncs for the same +/// conversation can reach Telegram out of order: a stalled auto-title edit +/// completing AFTER a manual rename would leave the thread (and the binding's +/// `display_title`) named after a title the user already replaced, and nothing +/// would ever retry. Re-reading before every attempt also means the title is +/// never a stale snapshot captured at spawn time. +/// +/// Deliberately NOT capped at N attempts. The loop exits only when the title it +/// just read equals the one it last sent, so an exit always leaves the provider +/// holding the current title; any fixed cap reintroduces exactly the bug this +/// closes (rename → stall → rename → … exhausts the cap and exits stale). It +/// cannot spin on its own: an iteration happens only when a NEW title was +/// observed, so it terminates as soon as renames stop, and each iteration is +/// rate-limited by one provider round-trip. Two concurrent syncs for the same +/// conversation converge on the same final value for the same reason. +/// +/// Serializing per conversation would be the other way to get ordering, but the +/// lock would also be taken by the INLINE rename path +/// (`sync_conversation_title_to_channels_core`), which would then block a user's +/// rename for up to Telegram's 60s timeout behind a stalled background sync — +/// reintroducing the hang this whole path exists to avoid. +async fn sync_conversation_title_until_current( + conn: &sea_orm::DatabaseConnection, + chat_channel_manager: &crate::chat_channel::manager::ChatChannelManager, + conversation_id: i32, +) { + let mut sent: Option = None; + loop { + let summary = match conversation_service::get_by_id(conn, conversation_id).await { + Ok(summary) => summary, + Err(e) => { + tracing::warn!( + "[conversations] chat-thread title sync stopped for {conversation_id} \ + (get_by_id failed): {e}" + ); + return; + } + }; + let Some(title) = summary.title else { return }; + if sent.as_deref() == Some(title.as_str()) { + return; + } + chat_channel_manager + .sync_conversation_title(conn, conversation_id, &title) + .await; + sent = Some(title); + } +} + +/// Broadcast and propagate title changes discovered outside codeg (for +/// example, Codex's session index or an import scan). Both operations are +/// best-effort: the database update has already committed, so notification +/// failures must not turn the originating list/scan request into an error. +/// +/// The two halves are deliberately NOT symmetric: +/// +/// * The sidebar upsert is emitted inline. It is DB-only and cheap, and the +/// response the caller is about to build must not disagree with what other +/// clients were just told. +/// * Chat-channel propagation is detached onto its own task, because it ends in +/// outbound HTTP (Telegram `editForumTopic`, a 60s per-request timeout, once +/// per bound thread). `list_all_conversations` is the sidebar's primary read +/// — also driven per-keystroke by the search and manage dialogs — so it must +/// never await a remote service. A slow or unreachable Telegram now costs a +/// late topic rename, not a hung conversation list. +/// +/// The detached half deliberately re-reads each title rather than carrying the +/// one this notification was raised for — see +/// [`sync_conversation_title_until_current`]. +/// +/// Returns the detached task's handle so tests can join it; production callers +/// drop it (the work is best-effort and already logged on failure). +async fn notify_conversation_title_updates( + conn: &sea_orm::DatabaseConnection, + emitter: &EventEmitter, + chat_channel_manager: &crate::chat_channel::manager::ChatChannelManager, + conversation_ids: Vec, +) -> tokio::task::JoinHandle<()> { + for conversation_id in &conversation_ids { + emit_conversation_upsert(emitter, conn, *conversation_id).await; + } + + let conn = conn.clone(); + let chat_channel_manager = chat_channel_manager.clone_ref(); + tokio::spawn(async move { + for conversation_id in conversation_ids { + sync_conversation_title_until_current(&conn, &chat_channel_manager, conversation_id) + .await; + } + }) +} + /// Emit a `conversation://changed` Deleted for `conversation_id` so every /// client removes the row. No re-fetch: the row is already soft-deleted. pub(crate) fn emit_conversation_deleted(emitter: &EventEmitter, conversation_id: i32) { @@ -2834,10 +3026,14 @@ mod tests { assert!(summary.git_branch.is_none()); // It surfaces in the default sidebar query (active-folder scope). - let rows = - list_all_conversations_core(&db.conn, None, None, None, None, None, false) - .await - .expect("list"); + let rows = list_all_conversations_core( + &db.conn, + &EventEmitter::Noop, + &crate::chat_channel::manager::ChatChannelManager::new(), + ListAllConversationsOptions::default(), + ) + .await + .expect("list"); assert!(rows.iter().any(|c| c.id == result.conversation_id)); } @@ -3225,12 +3421,454 @@ mod tests { #[tokio::test] async fn list_all_conversations_core_empty_db_returns_empty() { let db = fresh_in_memory_db().await; - let rows = list_all_conversations_core(&db.conn, None, None, None, None, None, false) - .await - .expect("list"); + let rows = list_all_conversations_core( + &db.conn, + &EventEmitter::Noop, + &crate::chat_channel::manager::ChatChannelManager::new(), + ListAllConversationsOptions::default(), + ) + .await + .expect("list"); assert!(rows.is_empty(), "fresh db must have zero conversations"); } + #[tokio::test] + async fn list_all_conversations_core_syncs_codex_index_title_before_search() { + let db = fresh_in_memory_db().await; + let folder_id = seed_folder(&db, "/tmp/codeg-list-codex-index").await; + let row = conversation_service::create( + &db.conn, + folder_id, + AgentType::Codex, + Some("Makefile 文件的作用".into()), + None, + ) + .await + .expect("create conversation"); + conversation_service::update_external_id( + &db.conn, + row.id, + "01a00496-1418-7273-a06f-dc4fae5cfa64".into(), + ) + .await + .expect("set external id"); + let before = conversation_service::get_by_id(&db.conn, row.id) + .await + .expect("get before"); + let titles = HashMap::from([( + "01a00496-1418-7273-a06f-dc4fae5cfa64".to_string(), + "解释 Makefile 文件作用".to_string(), + )]); + let (broadcaster, emitter) = sync_test_emitter(); + let mut events = broadcaster.subscribe(); + let (chat_channel_manager, title_edits) = title_sync_test_manager(&db, row.id).await; + + let rows = list_all_conversations_core_with_codex_titles( + &db.conn, + &emitter, + &chat_channel_manager, + ListAllConversationsOptions { + agent_type: Some(AgentType::Codex), + search: Some("解释 Makefile".into()), + ..Default::default() + }, + &titles, + ) + .await + .expect("list"); + + assert_eq!( + rows.len(), + 1, + "the new title must satisfy this same call's search" + ); + assert_eq!(rows[0].id, row.id); + assert_eq!(rows[0].title.as_deref(), Some("解释 Makefile 文件作用")); + let stored = conversation_service::get_by_id(&db.conn, row.id) + .await + .expect("get stored"); + assert_eq!(stored.title.as_deref(), Some("解释 Makefile 文件作用")); + assert_eq!(stored.updated_at, before.updated_at); + let event = events.try_recv().expect("title refresh must broadcast"); + assert_eq!(event.channel, CONVERSATION_CHANGED_EVENT); + assert_eq!(event.payload["summary"]["id"], row.id); + assert_eq!(event.payload["summary"]["title"], "解释 Makefile 文件作用"); + assert_eq!( + title_edits.titles.lock().await.as_slice(), + [format!("#{} 解释 Makefile 文件作用", row.id)], + "the same external title refresh must propagate to a bound chat thread" + ); + } + + #[tokio::test] + async fn list_all_conversations_core_preserves_locked_codex_title() { + let db = fresh_in_memory_db().await; + let folder_id = seed_folder(&db, "/tmp/codeg-list-codex-locked").await; + let row = conversation_service::create( + &db.conn, + folder_id, + AgentType::Codex, + Some("initial".into()), + None, + ) + .await + .expect("create conversation"); + conversation_service::update_external_id(&db.conn, row.id, "locked-session".into()) + .await + .expect("set external id"); + conversation_service::update_title(&db.conn, row.id, "我的手动标题".into()) + .await + .expect("manual rename"); + let titles = HashMap::from([("locked-session".to_string(), "Codex 自动标题".to_string())]); + + let rows = list_all_conversations_core_with_codex_titles( + &db.conn, + &EventEmitter::Noop, + &crate::chat_channel::manager::ChatChannelManager::new(), + ListAllConversationsOptions::default(), + &titles, + ) + .await + .expect("list"); + + let listed = rows + .iter() + .find(|item| item.id == row.id) + .expect("listed row"); + assert_eq!(listed.title.as_deref(), Some("我的手动标题")); + assert!(listed.title_locked); + let stored = conversation_service::get_by_id(&db.conn, row.id) + .await + .expect("get stored"); + assert_eq!(stored.title.as_deref(), Some("我的手动标题")); + assert!(stored.title_locked); + } + + #[tokio::test] + async fn list_all_conversations_core_keeps_title_when_codex_index_is_missing() { + let db = fresh_in_memory_db().await; + let folder_id = seed_folder(&db, "/tmp/codeg-list-codex-no-index").await; + let row = conversation_service::create( + &db.conn, + folder_id, + AgentType::Codex, + Some("数据库原标题".into()), + None, + ) + .await + .expect("create conversation"); + conversation_service::update_external_id(&db.conn, row.id, "missing-index-session".into()) + .await + .expect("set external id"); + let temp_dir = tempfile::tempdir().expect("tempdir"); + let titles = CodexParser::with_base_dir(temp_dir.path().join("missing-sessions")) + .load_thread_name_index(); + assert!( + titles.is_empty(), + "a missing index must produce no title updates" + ); + + let rows = list_all_conversations_core_with_codex_titles( + &db.conn, + &EventEmitter::Noop, + &crate::chat_channel::manager::ChatChannelManager::new(), + ListAllConversationsOptions::default(), + &titles, + ) + .await + .expect("list"); + + let listed = rows + .iter() + .find(|item| item.id == row.id) + .expect("listed row"); + assert_eq!(listed.title.as_deref(), Some("数据库原标题")); + let stored = conversation_service::get_by_id(&db.conn, row.id) + .await + .expect("get stored"); + assert_eq!(stored.title.as_deref(), Some("数据库原标题")); + } + + #[tokio::test] + async fn list_all_conversations_core_returns_persisted_rows_when_title_sync_fails() { + use sea_orm::{ConnectionTrait, DbBackend, Statement}; + + let db = fresh_in_memory_db().await; + let folder_id = seed_folder(&db, "/tmp/codeg-list-codex-sync-failure").await; + let row = conversation_service::create( + &db.conn, + folder_id, + AgentType::Codex, + Some("persisted title".into()), + None, + ) + .await + .expect("create conversation"); + conversation_service::update_external_id(&db.conn, row.id, "failing-session".into()) + .await + .expect("set external id"); + db.conn + .execute(Statement::from_string( + DbBackend::Sqlite, + format!( + r#"CREATE TRIGGER fail_codex_title_sync + BEFORE UPDATE OF title ON conversation + WHEN OLD.id = {} + BEGIN + SELECT RAISE(FAIL, 'injected title sync failure'); + END"#, + row.id + ), + )) + .await + .expect("install title failure trigger"); + let titles = + HashMap::from([("failing-session".to_string(), "new Codex title".to_string())]); + + let rows = list_all_conversations_core_with_codex_titles( + &db.conn, + &EventEmitter::Noop, + &crate::chat_channel::manager::ChatChannelManager::new(), + ListAllConversationsOptions::default(), + &titles, + ) + .await + .expect("list must degrade to persisted rows"); + + let listed = rows + .iter() + .find(|item| item.id == row.id) + .expect("persisted row remains visible"); + assert_eq!(listed.title.as_deref(), Some("persisted title")); + } + + #[tokio::test] + async fn scan_importable_sessions_syncs_title_and_notifies_clients() { + let db = fresh_in_memory_db().await; + let folder_id = seed_folder(&db, "/tmp/codeg-scan-codex-title-sync").await; + let row = conversation_service::create( + &db.conn, + folder_id, + AgentType::Codex, + Some("first prompt".into()), + None, + ) + .await + .expect("create imported conversation"); + conversation_service::update_external_id(&db.conn, row.id, "scan-session".into()) + .await + .expect("set external id"); + let (broadcaster, emitter) = sync_test_emitter(); + let mut events = broadcaster.subscribe(); + let (chat_channel_manager, title_edits) = title_sync_test_manager(&db, row.id).await; + let mut summary = scan_summary( + "scan-session", + AgentType::Codex, + Some("/tmp/codeg-scan-codex-title-sync"), + at(0), + ); + summary.1.title = Some("Codex index title".into()); + + let result = scan_importable_sessions_from_summaries( + &db.conn, + &emitter, + &chat_channel_manager, + vec![summary], + ) + .await + .expect("scan summaries"); + + assert_eq!(result.total_sessions, 1); + assert_eq!(result.importable_count, 0); + assert_eq!( + result.folders[0].sessions[0].status, + ScanSessionStatus::Imported + ); + let stored = conversation_service::get_by_id(&db.conn, row.id) + .await + .expect("get refreshed conversation"); + assert_eq!(stored.title.as_deref(), Some("Codex index title")); + let event = events + .try_recv() + .expect("scan title refresh must broadcast"); + assert_eq!(event.channel, CONVERSATION_CHANGED_EVENT); + assert_eq!(event.payload["kind"], "upsert"); + assert_eq!(event.payload["summary"]["id"], row.id); + assert_eq!(event.payload["summary"]["title"], "Codex index title"); + assert_eq!( + title_edits.wait_for_edits(1).await.as_slice(), + [format!("#{} Codex index title", row.id)], + "scan-discovered titles must propagate to a bound chat thread" + ); + } + + /// The channel half of a title notification must not be on the caller's + /// critical path: `edit_thread_title` reaches Telegram with a 60s per-call + /// timeout, and `list_all_conversations` is the sidebar's primary read. + #[tokio::test] + async fn notify_conversation_title_updates_detaches_channel_sync_from_the_caller() { + let db = fresh_in_memory_db().await; + let folder_id = seed_folder(&db, "/tmp/codeg-notify-detached").await; + let row = conversation_service::create( + &db.conn, + folder_id, + AgentType::Codex, + Some("detached title".into()), + None, + ) + .await + .expect("create conversation"); + let (broadcaster, emitter) = sync_test_emitter(); + let mut events = broadcaster.subscribe(); + let (chat_channel_manager, title_edits) = + title_sync_test_manager_with(&db, row.id, TitleEditRecorder::blocked()).await; + + let handle = notify_conversation_title_updates( + &db.conn, + &emitter, + &chat_channel_manager, + vec![row.id], + ) + .await; + + // Returned while the backend is still parked: the sidebar upsert is + // already out, the outbound edit has not even been attempted. + let event = events.try_recv().expect("upsert must be emitted inline"); + assert_eq!(event.payload["summary"]["title"], "detached title"); + assert!( + title_edits.recorded().await.is_empty(), + "caller must not wait on the chat backend" + ); + + title_edits.unblock(1); + handle.await.expect("detached title sync task"); + assert_eq!( + title_edits.recorded().await.as_slice(), + [format!("#{} detached title", row.id)], + "the detached task still propagates the title" + ); + } + + /// A detached edit can land after a rename that happened while it was in + /// flight. The last value the provider (and the binding's `display_title`) + /// ends up with must be the conversation's CURRENT title, not the one the + /// stalled sync started with — nothing retries afterwards. + #[tokio::test] + async fn detached_title_sync_converges_on_a_rename_that_lands_mid_flight() { + let db = fresh_in_memory_db().await; + let folder_id = seed_folder(&db, "/tmp/codeg-notify-late-rename").await; + let row = conversation_service::create( + &db.conn, + folder_id, + AgentType::Codex, + Some("auto title".into()), + None, + ) + .await + .expect("create conversation"); + let (_broadcaster, emitter) = sync_test_emitter(); + let (chat_channel_manager, title_edits) = + title_sync_test_manager_with(&db, row.id, TitleEditRecorder::blocked()).await; + + let handle = notify_conversation_title_updates( + &db.conn, + &emitter, + &chat_channel_manager, + vec![row.id], + ) + .await; + + // The auto-title edit is parked mid-flight; the user renames underneath it. + conversation_service::update_title(&db.conn, row.id, "manual rename".into()) + .await + .expect("manual rename"); + + title_edits.unblock(8); + handle.await.expect("detached title sync task"); + + let recorded = title_edits.recorded().await; + assert_eq!( + recorded.last().map(String::as_str), + Some(format!("#{} manual rename", row.id).as_str()), + "the provider must end on the newest title, not the stalled one: {recorded:?}" + ); + let bindings = + crate::db::service::thread_binding_service::list_by_conversation(&db.conn, row.id) + .await + .expect("list bindings"); + assert_eq!( + bindings[0].display_title.as_deref(), + Some(format!("#{} manual rename", row.id).as_str()), + "the persisted display title must match what the provider was last told" + ); + } + + /// The convergence loop must not be defeated by a RUN of renames that each + /// land mid-flight. Any fixed retry cap exits stale on a long enough run — + /// this drives more consecutive mid-flight renames than any such cap. + #[tokio::test] + async fn detached_title_sync_converges_after_a_run_of_mid_flight_renames() { + let db = fresh_in_memory_db().await; + let folder_id = seed_folder(&db, "/tmp/codeg-notify-rename-run").await; + let row = conversation_service::create( + &db.conn, + folder_id, + AgentType::Codex, + Some("title 1".into()), + None, + ) + .await + .expect("create conversation"); + let (_broadcaster, emitter) = sync_test_emitter(); + // Each provider edit is overtaken by the next rename before it returns. + let (chat_channel_manager, title_edits) = title_sync_test_manager_with( + &db, + row.id, + TitleEditRecorder::renaming_mid_edit( + &db, + row.id, + &["title 2", "title 3", "title 4", "title 5", "title 6"], + ), + ) + .await; + + notify_conversation_title_updates( + &db.conn, + &emitter, + &chat_channel_manager, + vec![row.id], + ) + .await + .await + .expect("detached title sync task"); + + let recorded = title_edits.recorded().await; + let current = conversation_service::get_by_id(&db.conn, row.id) + .await + .expect("read conversation") + .title + .expect("conversation has a title"); + // The invariant, stated against whatever the row actually ended on: the + // last thing the provider was told IS the conversation's current title. + assert_eq!( + recorded.last().map(String::as_str), + Some(format!("#{} {current}", row.id).as_str()), + "provider must end on the row's current title however long the run: {recorded:?}" + ); + assert_eq!( + current, "title 6", + "fixture must exhaust every queued rename, or it is not testing a long run" + ); + let bindings = + crate::db::service::thread_binding_service::list_by_conversation(&db.conn, row.id) + .await + .expect("list bindings"); + assert_eq!( + bindings[0].display_title.as_deref(), + Some(format!("#{} title 6", row.id).as_str()) + ); + } + #[tokio::test] async fn list_opened_tabs_core_empty_db_returns_empty() { let db = fresh_in_memory_db().await; @@ -3596,7 +4234,12 @@ mod tests { #[tokio::test] async fn import_local_conversations_core_missing_folder_errors() { let db = fresh_in_memory_db().await; - let err = import_local_conversations_core(&db.conn, &EventEmitter::Noop, 999_999) + let err = import_local_conversations_core( + &db.conn, + &EventEmitter::Noop, + &crate::chat_channel::manager::ChatChannelManager::new(), + 999_999, + ) .await .expect_err("missing folder must surface as error"); let msg = format!("{err:?}"); @@ -3651,9 +4294,14 @@ mod tests { .await .expect("delete"); // After soft delete the row should no longer show up in list_all. - let remaining = list_all_conversations_core(&db.conn, None, None, None, None, None, false) - .await - .expect("list"); + let remaining = list_all_conversations_core( + &db.conn, + &EventEmitter::Noop, + &crate::chat_channel::manager::ChatChannelManager::new(), + ListAllConversationsOptions::default(), + ) + .await + .expect("list"); assert!( remaining.iter().all(|c| c.id != conv_id), "soft-deleted conversation must not appear in list_all" @@ -3744,6 +4392,219 @@ mod tests { (broadcaster, emitter) } + /// Applies one queued rename per provider edit, so the rename provably + /// lands while that edit is still in flight — the exact interleaving a + /// detached sync has to survive. + #[derive(Clone)] + struct RenameDuringEdit { + conn: sea_orm::DatabaseConnection, + conversation_id: i32, + pending: std::sync::Arc>>, + } + + #[derive(Clone)] + struct TitleEditRecorder { + titles: std::sync::Arc>>, + /// Stands in for Telegram's latency. Open by default; `blocked()` parks + /// every `edit_thread_title` until the test hands out permits, which is + /// how the detached-channel-sync tests prove a caller did not wait. + gate: std::sync::Arc, + rename_during_edit: Option, + } + + impl Default for TitleEditRecorder { + fn default() -> Self { + Self { + titles: Default::default(), + gate: std::sync::Arc::new(tokio::sync::Semaphore::new( + tokio::sync::Semaphore::MAX_PERMITS, + )), + rename_during_edit: None, + } + } + } + + impl TitleEditRecorder { + fn blocked() -> Self { + Self { + gate: std::sync::Arc::new(tokio::sync::Semaphore::new(0)), + ..Default::default() + } + } + + /// Open gate, but every edit is overtaken by the next queued rename. + fn renaming_mid_edit( + db: &crate::db::AppDatabase, + conversation_id: i32, + renames: &[&str], + ) -> Self { + Self { + rename_during_edit: Some(RenameDuringEdit { + conn: db.conn.clone(), + conversation_id, + pending: std::sync::Arc::new(tokio::sync::Mutex::new( + renames.iter().map(|t| (*t).to_string()).collect(), + )), + }), + ..Default::default() + } + } + + fn unblock(&self, edits: usize) { + self.gate.add_permits(edits); + } + + async fn recorded(&self) -> Vec { + self.titles.lock().await.clone() + } + + /// Await a detached channel sync. Bounded so a wiring regression fails + /// the test instead of hanging it. + async fn wait_for_edits(&self, count: usize) -> Vec { + for _ in 0..500 { + let titles = self.recorded().await; + if titles.len() >= count { + return titles; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + panic!("timed out waiting for {count} chat-thread title edit(s)"); + } + } + + struct RecordingTitleBackend { + recorder: TitleEditRecorder, + } + + #[async_trait::async_trait] + impl crate::chat_channel::traits::ChatChannelBackend for RecordingTitleBackend { + fn channel_type(&self) -> crate::chat_channel::types::ChannelType { + crate::chat_channel::types::ChannelType::Telegram + } + + async fn start( + &self, + _command_tx: tokio::sync::mpsc::Sender, + ) -> Result<(), crate::chat_channel::error::ChatChannelError> { + Ok(()) + } + + async fn stop(&self) -> Result<(), crate::chat_channel::error::ChatChannelError> { + Ok(()) + } + + async fn status(&self) -> crate::chat_channel::types::ChannelConnectionStatus { + crate::chat_channel::types::ChannelConnectionStatus::Connected + } + + async fn send_message( + &self, + _text: &str, + ) -> Result< + crate::chat_channel::types::SentMessageId, + crate::chat_channel::error::ChatChannelError, + > { + Ok(crate::chat_channel::types::SentMessageId("sent".into())) + } + + async fn send_rich_message( + &self, + _message: &crate::chat_channel::types::RichMessage, + ) -> Result< + crate::chat_channel::types::SentMessageId, + crate::chat_channel::error::ChatChannelError, + > { + Ok(crate::chat_channel::types::SentMessageId("sent".into())) + } + + async fn edit_thread_title( + &self, + _target: &crate::chat_channel::types::ChannelMessageTarget, + title: &str, + ) -> Result<(), crate::chat_channel::error::ChatChannelError> { + self.recorder + .gate + .acquire() + .await + .expect("title edit gate closed") + .forget(); + self.recorder.titles.lock().await.push(title.to_string()); + if let Some(hook) = &self.recorder.rename_during_edit { + let next = hook.pending.lock().await.pop_front(); + if let Some(next) = next { + conversation_service::update_title(&hook.conn, hook.conversation_id, next) + .await + .expect("rename during in-flight edit"); + } + } + Ok(()) + } + + async fn test_connection( + &self, + ) -> Result<(), crate::chat_channel::error::ChatChannelError> { + Ok(()) + } + } + + async fn title_sync_test_manager( + db: &crate::db::AppDatabase, + conversation_id: i32, + ) -> ( + crate::chat_channel::manager::ChatChannelManager, + TitleEditRecorder, + ) { + title_sync_test_manager_with(db, conversation_id, TitleEditRecorder::default()).await + } + + async fn title_sync_test_manager_with( + db: &crate::db::AppDatabase, + conversation_id: i32, + recorder: TitleEditRecorder, + ) -> ( + crate::chat_channel::manager::ChatChannelManager, + TitleEditRecorder, + ) { + let channel = crate::db::service::chat_channel_service::create( + &db.conn, + "title sync test".into(), + "telegram".into(), + "{}".into(), + true, + false, + None, + ) + .await + .expect("create chat channel"); + let manager = crate::chat_channel::manager::ChatChannelManager::new(); + manager + .add_channel( + channel.id, + channel.name, + crate::chat_channel::types::ChannelType::Telegram, + Box::new(RecordingTitleBackend { + recorder: recorder.clone(), + }), + ) + .await + .expect("connect recording channel"); + let target = crate::chat_channel::types::ChannelMessageTarget::telegram_forum_topic( + channel.id, "chat-1", "topic-1", + ); + crate::db::service::thread_binding_service::upsert_for_target( + &db.conn, + &target, + "telegram", + conversation_id, + None, + "test-user", + Some("old topic title".into()), + ) + .await + .expect("bind conversation thread"); + (manager, recorder) + } + #[tokio::test] async fn emit_conversation_upsert_broadcasts_full_root_summary() { let db = fresh_in_memory_db().await; @@ -4413,7 +5274,12 @@ mod tests { let folder_id = seed_folder(&db, "/tmp/legacy-guard").await; let _held = IMPORT_GUARD.try_lock().expect("guard free in test"); - let err = import_local_conversations_core(&db.conn, &EventEmitter::Noop, folder_id) + let err = import_local_conversations_core( + &db.conn, + &EventEmitter::Noop, + &crate::chat_channel::manager::ChatChannelManager::new(), + folder_id, + ) .await .expect_err("legacy import must be rejected while an import is in progress"); let msg = format!("{err:?}").to_lowercase(); diff --git a/src-tauri/src/db/service/conversation_service.rs b/src-tauri/src/db/service/conversation_service.rs index cc00519f8..63f612729 100644 --- a/src-tauri/src/db/service/conversation_service.rs +++ b/src-tauri/src/db/service/conversation_service.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use chrono::Utc; use sea_orm::{ ActiveModelTrait, ActiveValue::NotSet, ColumnTrait, DatabaseConnection, EntityTrait, @@ -70,7 +72,10 @@ pub async fn create_with_delegation( } else { ConversationKind::Regular }; - create_inner(conn, folder_id, agent_type, title, git_branch, delegation, kind).await + create_inner( + conn, folder_id, agent_type, title, git_branch, delegation, kind, + ) + .await } async fn create_inner( @@ -279,6 +284,144 @@ pub async fn retitle_if_unchanged( Ok(res.rows_affected > 0) } +/// Conditionally adopt one title discovered from Codex's session index. +/// +/// Every field used to select the candidate is re-checked in the UPDATE, plus +/// the title observed by the candidate read. This prevents a delayed refresh +/// from writing an index title after the row was re-pointed, deleted, manually +/// renamed, moved into a deleted folder, or refreshed by another task. +/// +/// `kind` is deliberately absent: it is written once at insert and never +/// updated, so the candidate query's filter cannot go stale between the two. +async fn refresh_codex_auto_title_candidate( + conn: &DatabaseConnection, + candidate: &conversation::Model, + title: &str, +) -> Result { + use sea_orm::sea_query::Expr; + + let title = title.trim(); + let Some(external_id) = candidate.external_id.as_deref() else { + return Ok(false); + }; + if title.is_empty() || candidate.title.as_deref() == Some(title) { + return Ok(false); + } + + let old_title = match candidate.title.as_deref() { + Some(old_title) => conversation::Column::Title.eq(old_title), + None => conversation::Column::Title.is_null(), + }; + let res = conversation::Entity::update_many() + .col_expr(conversation::Column::Title, Expr::value(title)) + .filter(conversation::Column::Id.eq(candidate.id)) + .filter(conversation::Column::AgentType.eq(AgentType::Codex.as_wire().into_owned())) + .filter(conversation::Column::ExternalId.eq(external_id)) + .filter(conversation::Column::DeletedAt.is_null()) + .filter(conversation::Column::TitleLocked.eq(false)) + .filter( + conversation::Column::FolderId.in_subquery( + sea_orm::sea_query::Query::select() + .column(folder::Column::Id) + .from(folder::Entity) + .and_where(folder::Column::DeletedAt.is_null()) + .to_owned(), + ), + ) + .filter(old_title) + .exec(conn) + .await?; + Ok(res.rows_affected > 0) +} + +/// Refresh every live, unlocked Codex conversation whose external session id +/// has a title in `titles`. Candidate selection is limited to indexed session +/// ids and chunked below SQLite's bound-variable limit. Each UPDATE atomically +/// re-checks the complete candidate identity via +/// [`refresh_codex_auto_title_candidate`]. Converged rows issue no UPDATE and +/// title refreshes never bump `updated_at`. This is a best-effort reconciliation: +/// a failed chunk or row is logged and skipped while successful row ids are +/// retained for downstream notifications. +/// +/// Candidate scope mirrors [`list_all`]'s own visibility rules — a live folder +/// and `kind != 'loop'` — because every refreshed id is broadcast as a sidebar +/// upsert. Refreshing a row this query would never return would push a row the +/// list deliberately hides into every client's sidebar until its next refetch. +/// +/// Each candidate keeps its OWN autocommit UPDATE rather than sharing one +/// transaction per chunk: the guarantee callers rely on is that a row that +/// failed (or lost its CAS) never holds back the rows that succeeded, and a +/// chunk-wide transaction would roll those back together. +pub(crate) async fn refresh_codex_auto_titles( + conn: &DatabaseConnection, + titles: &HashMap, +) -> Vec { + if titles.is_empty() { + return Vec::new(); + } + + const SQLITE_TITLE_QUERY_CHUNK_SIZE: usize = 500; + let external_ids: Vec = titles + .iter() + .filter(|(_, title)| !title.trim().is_empty()) + .map(|(external_id, _)| external_id.clone()) + .collect(); + let mut refreshed = Vec::new(); + + for external_id_chunk in external_ids.chunks(SQLITE_TITLE_QUERY_CHUNK_SIZE) { + let candidates = match conversation::Entity::find() + .filter(conversation::Column::AgentType.eq(AgentType::Codex.as_wire().into_owned())) + .filter(conversation::Column::ExternalId.is_in(external_id_chunk.iter().cloned())) + .filter(conversation::Column::DeletedAt.is_null()) + .filter(conversation::Column::TitleLocked.eq(false)) + .filter(conversation::Column::Kind.ne(ConversationKind::Loop)) + .filter( + conversation::Column::FolderId.in_subquery( + sea_orm::sea_query::Query::select() + .column(folder::Column::Id) + .from(folder::Entity) + .and_where(folder::Column::DeletedAt.is_null()) + .to_owned(), + ), + ) + .order_by_asc(conversation::Column::Id) + .all(conn) + .await + { + Ok(candidates) => candidates, + Err(error) => { + tracing::warn!( + error = %error, + session_count = external_id_chunk.len(), + "failed to select Codex title refresh candidates; skipping chunk" + ); + continue; + } + }; + + for candidate in candidates { + let Some(external_id) = candidate.external_id.as_deref() else { + continue; + }; + let Some(title) = titles.get(external_id) else { + continue; + }; + match refresh_codex_auto_title_candidate(conn, &candidate, title).await { + Ok(true) => refreshed.push(candidate.id), + Ok(false) => {} + Err(error) => tracing::warn!( + error = %error, + conversation_id = candidate.id, + external_id, + "failed to refresh Codex title candidate; skipping row" + ), + } + } + } + + refreshed +} + /// Adopt an imported conversation's newest activity from its agent-side /// transcript: stamp `updated_at` with the session file's own last-activity /// time (never `now()` — the scan is not the activity) and re-sync @@ -781,9 +924,15 @@ mod tests { async fn list_children_orders_newest_first() { let db = fresh_in_memory_db().await; let folder = seed_folder(&db, "/tmp/codeg-list-children-order").await; - let parent = create(&db.conn, folder, AgentType::ClaudeCode, Some("P".into()), None) - .await - .expect("parent"); + let parent = create( + &db.conn, + folder, + AgentType::ClaudeCode, + Some("P".into()), + None, + ) + .await + .expect("parent"); // Two children created oldest → newest under the same parent. let first = create_with_delegation( &db.conn, @@ -882,7 +1031,9 @@ mod tests { let folder = seed_folder(&db, "/tmp/codeg-child-count-deleted").await; let (parent, child) = seed_parent_with_child(&db.conn, folder).await; - soft_delete(&db.conn, child).await.expect("soft delete child"); + soft_delete(&db.conn, child) + .await + .expect("soft delete child"); // A removed sub-session must not keep the parent's chevron alive: the // aggregate filters deleted_at IS NULL, matching list_children. @@ -900,9 +1051,15 @@ mod tests { async fn update_pin_sets_and_clears_without_bumping_updated_at() { let db = fresh_in_memory_db().await; let folder = seed_folder(&db, "/tmp/codeg-update-pin").await; - let conv = create(&db.conn, folder, AgentType::ClaudeCode, Some("c".into()), None) - .await - .expect("create"); + let conv = create( + &db.conn, + folder, + AgentType::ClaudeCode, + Some("c".into()), + None, + ) + .await + .expect("create"); // Freshly created rows are unpinned, and the summary projection carries // the field through (conv_to_summary mapping). @@ -917,7 +1074,10 @@ mod tests { // preference, not activity). update_pin(&db.conn, conv.id, true).await.expect("pin"); let pinned = get_by_id(&db.conn, conv.id).await.expect("get pinned"); - assert!(pinned.pinned_at.is_some(), "pinned_at must be set after pin"); + assert!( + pinned.pinned_at.is_some(), + "pinned_at must be set after pin" + ); assert_eq!( pinned.updated_at, updated_at_before, "pinning must not bump updated_at" @@ -955,11 +1115,20 @@ mod tests { async fn create_leaves_title_unlocked() { let db = fresh_in_memory_db().await; let folder = seed_folder(&db, "/tmp/codeg-title-unlocked").await; - let row = create(&db.conn, folder, AgentType::ClaudeCode, Some("hi".into()), None) - .await - .expect("create"); + let row = create( + &db.conn, + folder, + AgentType::ClaudeCode, + Some("hi".into()), + None, + ) + .await + .expect("create"); let summary = get_by_id(&db.conn, row.id).await.expect("get"); - assert!(!summary.title_locked, "new conversation must start unlocked"); + assert!( + !summary.title_locked, + "new conversation must start unlocked" + ); } #[tokio::test] @@ -1202,9 +1371,15 @@ mod tests { async fn retitle_if_unchanged_follows_the_owner_rename() { let db = fresh_in_memory_db().await; let folder = seed_folder(&db, "/tmp/codeg-retitle-follow").await; - let row = create(&db.conn, folder, AgentType::ClaudeCode, Some("old".into()), None) - .await - .expect("create"); + let row = create( + &db.conn, + folder, + AgentType::ClaudeCode, + Some("old".into()), + None, + ) + .await + .expect("create"); let before = row.updated_at; lock_title(&db.conn, row.id).await.expect("lock"); @@ -1229,9 +1404,15 @@ mod tests { async fn retitle_if_unchanged_refuses_after_a_manual_rename() { let db = fresh_in_memory_db().await; let folder = seed_folder(&db, "/tmp/codeg-retitle-refuse").await; - let row = create(&db.conn, folder, AgentType::ClaudeCode, Some("old".into()), None) - .await - .expect("create"); + let row = create( + &db.conn, + folder, + AgentType::ClaudeCode, + Some("old".into()), + None, + ) + .await + .expect("create"); update_title(&db.conn, row.id, "User pick".into()) .await .expect("rename"); @@ -1248,9 +1429,15 @@ mod tests { async fn retitle_if_unchanged_skips_empty_and_identical() { let db = fresh_in_memory_db().await; let folder = seed_folder(&db, "/tmp/codeg-retitle-skip").await; - let row = create(&db.conn, folder, AgentType::ClaudeCode, Some("same".into()), None) - .await - .expect("create"); + let row = create( + &db.conn, + folder, + AgentType::ClaudeCode, + Some("same".into()), + None, + ) + .await + .expect("create"); assert!( !retitle_if_unchanged(&db.conn, row.id, "same", "same") @@ -1268,6 +1455,445 @@ mod tests { assert_eq!(summary.title.as_deref(), Some("same")); } + #[tokio::test] + async fn refresh_codex_auto_titles_converges_without_bumping_updated_at() { + let db = fresh_in_memory_db().await; + let folder = seed_folder(&db, "/tmp/codeg-codex-index-title").await; + let row = create( + &db.conn, + folder, + AgentType::Codex, + Some("Makefile 文件的作用".into()), + None, + ) + .await + .expect("create"); + update_external_id(&db.conn, row.id, "codex-session-1".into()) + .await + .expect("set external id"); + let before = get_by_id(&db.conn, row.id).await.expect("get before"); + let titles = HashMap::from([( + "codex-session-1".to_string(), + "解释 Makefile 文件作用".to_string(), + )]); + + assert_eq!( + refresh_codex_auto_titles(&db.conn, &titles).await, + vec![row.id] + ); + let refreshed = get_by_id(&db.conn, row.id).await.expect("get refreshed"); + assert_eq!(refreshed.title.as_deref(), Some("解释 Makefile 文件作用")); + assert_eq!( + refreshed.updated_at, before.updated_at, + "index title refresh must not count as conversation activity" + ); + + assert_eq!( + refresh_codex_auto_titles(&db.conn, &titles).await, + Vec::::new(), + "a converged title map must issue no UPDATE" + ); + } + + #[tokio::test] + async fn refresh_codex_auto_titles_keeps_partial_successes_after_row_failure() { + use sea_orm::{ConnectionTrait, DbBackend, Statement}; + + let db = fresh_in_memory_db().await; + let folder = seed_folder(&db, "/tmp/codeg-codex-index-partial-failure").await; + let successful = create( + &db.conn, + folder, + AgentType::Codex, + Some("old success".into()), + None, + ) + .await + .expect("create successful candidate"); + update_external_id(&db.conn, successful.id, "session-success".into()) + .await + .expect("set successful external id"); + let failing = create( + &db.conn, + folder, + AgentType::Codex, + Some("old failure".into()), + None, + ) + .await + .expect("create failing candidate"); + update_external_id(&db.conn, failing.id, "session-failure".into()) + .await + .expect("set failing external id"); + db.conn + .execute(Statement::from_string( + DbBackend::Sqlite, + format!( + r#"CREATE TRIGGER fail_second_codex_title_sync + BEFORE UPDATE OF title ON conversation + WHEN OLD.id = {} + BEGIN + SELECT RAISE(FAIL, 'injected partial title sync failure'); + END"#, + failing.id + ), + )) + .await + .expect("install title failure trigger"); + let titles = HashMap::from([ + ("session-success".to_string(), "new success".to_string()), + ("session-failure".to_string(), "new failure".to_string()), + ]); + + let refreshed = refresh_codex_auto_titles(&db.conn, &titles).await; + + assert_eq!(refreshed, vec![successful.id]); + assert_eq!( + get_by_id(&db.conn, successful.id) + .await + .expect("get successful row") + .title + .as_deref(), + Some("new success") + ); + assert_eq!( + get_by_id(&db.conn, failing.id) + .await + .expect("get failing row") + .title + .as_deref(), + Some("old failure") + ); + } + + #[tokio::test] + async fn refresh_codex_auto_title_candidate_rechecks_external_id_at_write_time() { + let db = fresh_in_memory_db().await; + let folder = seed_folder(&db, "/tmp/codeg-codex-index-race").await; + let row = create( + &db.conn, + folder, + AgentType::Codex, + Some("old title".into()), + None, + ) + .await + .expect("create"); + update_external_id(&db.conn, row.id, "session-before".into()) + .await + .expect("set initial external id"); + let stale_candidate = conversation::Entity::find_by_id(row.id) + .one(&db.conn) + .await + .expect("query candidate") + .expect("candidate exists"); + + update_external_id(&db.conn, row.id, "session-after".into()) + .await + .expect("re-point conversation"); + let wrote = refresh_codex_auto_title_candidate( + &db.conn, + &stale_candidate, + "title for session-before", + ) + .await + .expect("conditional refresh"); + + assert!(!wrote, "a stale candidate must not update a re-pointed row"); + let current = get_by_id(&db.conn, row.id).await.expect("read current row"); + assert_eq!(current.external_id.as_deref(), Some("session-after")); + assert_eq!(current.title.as_deref(), Some("old title")); + } + + #[tokio::test] + async fn refresh_codex_auto_title_candidate_rechecks_original_title_at_write_time() { + let db = fresh_in_memory_db().await; + let folder = seed_folder(&db, "/tmp/codeg-codex-index-title-race").await; + let row = create( + &db.conn, + folder, + AgentType::Codex, + Some("candidate title".into()), + None, + ) + .await + .expect("create"); + update_external_id(&db.conn, row.id, "session-title-race".into()) + .await + .expect("set external id"); + let stale_candidate = conversation::Entity::find_by_id(row.id) + .one(&db.conn) + .await + .expect("query candidate") + .expect("candidate exists"); + + refresh_auto_title(&db.conn, row.id, "newer automatic title".into()) + .await + .expect("apply concurrent automatic title"); + let wrote = refresh_codex_auto_title_candidate( + &db.conn, + &stale_candidate, + "stale session index title", + ) + .await + .expect("conditional refresh"); + + assert!(!wrote, "a stale candidate must not overwrite a newer title"); + let current = get_by_id(&db.conn, row.id).await.expect("read current row"); + assert_eq!(current.title.as_deref(), Some("newer automatic title")); + } + + #[tokio::test] + async fn refresh_codex_auto_title_candidate_rechecks_deleted_at_at_write_time() { + let db = fresh_in_memory_db().await; + let folder = seed_folder(&db, "/tmp/codeg-codex-index-delete-race").await; + let row = create( + &db.conn, + folder, + AgentType::Codex, + Some("candidate title".into()), + None, + ) + .await + .expect("create"); + update_external_id(&db.conn, row.id, "session-delete-race".into()) + .await + .expect("set external id"); + let stale_candidate = conversation::Entity::find_by_id(row.id) + .one(&db.conn) + .await + .expect("query candidate") + .expect("candidate exists"); + + soft_delete(&db.conn, row.id) + .await + .expect("concurrently delete candidate"); + let wrote = refresh_codex_auto_title_candidate( + &db.conn, + &stale_candidate, + "stale session index title", + ) + .await + .expect("conditional refresh"); + + assert!(!wrote, "a stale candidate must not update a deleted row"); + let current = conversation::Entity::find_by_id(row.id) + .one(&db.conn) + .await + .expect("read current row") + .expect("soft-deleted row remains persisted"); + assert!(current.deleted_at.is_some()); + assert_eq!(current.title.as_deref(), Some("candidate title")); + } + + #[tokio::test] + async fn refresh_codex_auto_title_candidate_rechecks_title_lock_at_write_time() { + let db = fresh_in_memory_db().await; + let folder = seed_folder(&db, "/tmp/codeg-codex-index-lock-race").await; + let row = create( + &db.conn, + folder, + AgentType::Codex, + Some("same manual title".into()), + None, + ) + .await + .expect("create"); + update_external_id(&db.conn, row.id, "session-lock-race".into()) + .await + .expect("set external id"); + let stale_candidate = conversation::Entity::find_by_id(row.id) + .one(&db.conn) + .await + .expect("query candidate") + .expect("candidate exists"); + + update_title(&db.conn, row.id, "same manual title".into()) + .await + .expect("lock title without changing its value"); + let wrote = refresh_codex_auto_title_candidate( + &db.conn, + &stale_candidate, + "stale session index title", + ) + .await + .expect("conditional refresh"); + + assert!( + !wrote, + "a stale candidate must not overwrite a locked title" + ); + let current = get_by_id(&db.conn, row.id).await.expect("read current row"); + assert!(current.title_locked); + assert_eq!(current.title.as_deref(), Some("same manual title")); + } + + #[tokio::test] + async fn refresh_codex_auto_title_candidate_rechecks_folder_deletion_at_write_time() { + let db = fresh_in_memory_db().await; + let folder = seed_folder(&db, "/tmp/codeg-codex-index-folder-race").await; + let row = create( + &db.conn, + folder, + AgentType::Codex, + Some("old title".into()), + None, + ) + .await + .expect("create"); + update_external_id(&db.conn, row.id, "session-folder-race".into()) + .await + .expect("set external id"); + let stale_candidate = conversation::Entity::find_by_id(row.id) + .one(&db.conn) + .await + .expect("query candidate") + .expect("candidate exists"); + + crate::db::service::folder_service::soft_delete_folder(&db.conn, folder) + .await + .expect("soft delete folder"); + let wrote = + refresh_codex_auto_title_candidate(&db.conn, &stale_candidate, "index title").await; + + assert!( + !wrote.expect("conditional refresh"), + "a row whose folder was deleted mid-refresh must not be rewritten (and so must not be broadcast)" + ); + let current = get_by_id(&db.conn, row.id).await.expect("read current row"); + assert_eq!(current.title.as_deref(), Some("old title")); + } + + #[tokio::test] + async fn refresh_codex_auto_title_candidate_adopts_a_title_over_null() { + let db = fresh_in_memory_db().await; + let folder = seed_folder(&db, "/tmp/codeg-codex-index-null-title").await; + let row = create(&db.conn, folder, AgentType::Codex, None, None) + .await + .expect("create"); + update_external_id(&db.conn, row.id, "session-null-title".into()) + .await + .expect("set external id"); + let candidate = conversation::Entity::find_by_id(row.id) + .one(&db.conn) + .await + .expect("query candidate") + .expect("candidate exists"); + assert!(candidate.title.is_none(), "fixture must start titleless"); + + assert!( + refresh_codex_auto_title_candidate(&db.conn, &candidate, "first Codex title") + .await + .expect("conditional refresh"), + "the IS NULL branch of the observed-title CAS must still write" + ); + assert_eq!( + get_by_id(&db.conn, row.id) + .await + .expect("read current row") + .title + .as_deref(), + Some("first Codex title") + ); + + // ...and once a title exists, the same stale (title = NULL) candidate + // must no longer match. + assert!( + !refresh_codex_auto_title_candidate(&db.conn, &candidate, "second Codex title") + .await + .expect("conditional refresh"), + "a stale titleless candidate must not clobber the title it just wrote" + ); + } + + #[tokio::test] + async fn refresh_codex_auto_titles_skips_rows_the_sidebar_list_hides() { + let db = fresh_in_memory_db().await; + let live_folder = seed_folder(&db, "/tmp/codeg-codex-index-visible").await; + let dead_folder = seed_folder(&db, "/tmp/codeg-codex-index-hidden").await; + + let visible = create( + &db.conn, + live_folder, + AgentType::Codex, + Some("old visible".into()), + None, + ) + .await + .expect("create visible"); + update_external_id(&db.conn, visible.id, "session-visible".into()) + .await + .expect("set visible external id"); + + let in_dead_folder = create( + &db.conn, + dead_folder, + AgentType::Codex, + Some("old hidden".into()), + None, + ) + .await + .expect("create hidden"); + update_external_id(&db.conn, in_dead_folder.id, "session-hidden".into()) + .await + .expect("set hidden external id"); + crate::db::service::folder_service::soft_delete_folder(&db.conn, dead_folder) + .await + .expect("soft delete folder"); + + let loop_row = create( + &db.conn, + live_folder, + AgentType::Codex, + Some("old loop".into()), + None, + ) + .await + .expect("create loop row"); + update_external_id(&db.conn, loop_row.id, "session-loop".into()) + .await + .expect("set loop external id"); + // No public write path mints kind='loop' yet, so flip it directly. + let mut active: conversation::ActiveModel = conversation::Entity::find_by_id(loop_row.id) + .one(&db.conn) + .await + .expect("query loop row") + .expect("loop row exists") + .into(); + active.kind = Set(ConversationKind::Loop); + active.update(&db.conn).await.expect("flip kind"); + + let titles = HashMap::from([ + ("session-visible".to_string(), "new visible".to_string()), + ("session-hidden".to_string(), "new hidden".to_string()), + ("session-loop".to_string(), "new loop".to_string()), + ]); + + let refreshed = refresh_codex_auto_titles(&db.conn, &titles).await; + + assert_eq!( + refreshed, + vec![visible.id], + "only rows `list_all` would return may be refreshed — every refreshed id is broadcast as a sidebar upsert" + ); + assert_eq!( + get_by_id(&db.conn, in_dead_folder.id) + .await + .expect("read hidden row") + .title + .as_deref(), + Some("old hidden") + ); + assert_eq!( + get_by_id(&db.conn, loop_row.id) + .await + .expect("read loop row") + .title + .as_deref(), + Some("old loop") + ); + } + #[tokio::test] async fn refresh_external_activity_moves_forward_only() { let db = fresh_in_memory_db().await; diff --git a/src-tauri/src/parsers/codex.rs b/src-tauri/src/parsers/codex.rs index 8fe892bb7..2aef185e3 100644 --- a/src-tauri/src/parsers/codex.rs +++ b/src-tauri/src/parsers/codex.rs @@ -42,6 +42,43 @@ impl CodexParser { Self { base_dir } } + /// Load Codex's append-only session title index. The transcript remains the + /// fallback source, so a missing/unreadable index or a malformed line is + /// deliberately ignored. Later non-empty records for the same session win. + pub(crate) fn load_thread_name_index(&self) -> HashMap { + let mut titles = HashMap::new(); + let Some(home_dir) = self.base_dir.parent() else { + return titles; + }; + let Ok(file) = fs::File::open(home_dir.join("session_index.jsonl")) else { + return titles; + }; + + for line in BufReader::new(file).lines() { + let line = match line { + Ok(line) => line, + Err(_) => break, + }; + if line.trim().is_empty() { + continue; + } + let Ok(value) = serde_json::from_str::(&line) else { + continue; + }; + let session_id = value.get("id").and_then(serde_json::Value::as_str); + let thread_name = value + .get("thread_name") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|name| !name.is_empty()); + if let (Some(id), Some(name)) = (session_id, thread_name) { + titles.insert(id.to_string(), truncate_str(name, 100)); + } + } + + titles + } + fn parse_jsonl_summary( &self, path: &PathBuf, @@ -292,6 +329,10 @@ impl AgentParser for CodexParser { return Ok(conversations); } + // Apply this outside `summary_cache`: changing only session_index.jsonl + // must refresh a title even when the rollout itself is unchanged. + let indexed_titles = self.load_thread_name_index(); + for entry in WalkDir::new(&self.base_dir) .into_iter() .filter_map(|e| e.ok()) @@ -308,7 +349,12 @@ impl AgentParser for CodexParser { match super::summary_cache::get_or_parse(AgentType::Codex, &path, || { self.parse_jsonl_summary(&path) }) { - Ok(Some(summary)) => conversations.push(summary), + Ok(Some(mut summary)) => { + if let Some(title) = indexed_titles.get(&summary.id) { + summary.title = Some(title.clone()); + } + conversations.push(summary); + } _ => continue, } } @@ -335,7 +381,11 @@ impl AgentParser for CodexParser { } let fname = path.file_name().unwrap_or_default().to_string_lossy(); if fname.contains(conversation_id) { - return self.parse_conversation_detail(&path, conversation_id); + let mut detail = self.parse_conversation_detail(&path, conversation_id)?; + if let Some(title) = self.load_thread_name_index().get(conversation_id) { + detail.summary.title = Some(title.clone()); + } + return Ok(detail); } } @@ -4171,6 +4221,7 @@ mod tests { use super::COLLAB_OP_KEY; use super::should_skip_duplicate_user_message; use super::strip_blocked_resource_mentions; + use super::AgentParser; use super::CodexParser; use super::CODEX_SCRIPT_TOOL_NAME; use crate::models::{ @@ -4182,6 +4233,188 @@ mod tests { use std::path::PathBuf; use std::time::{SystemTime, UNIX_EPOCH}; + fn write_index_title_fixture( + conversation_id: &str, + ) -> (tempfile::TempDir, CodexParser, PathBuf) { + let temp_dir = tempfile::tempdir().expect("create temp dir"); + let codex_home = temp_dir.path().join(".codex"); + let sessions_dir = codex_home.join("sessions"); + let rollout_dir = sessions_dir.join("2026").join("08").join("15"); + fs::create_dir_all(&rollout_dir).expect("create rollout dir"); + + let rollout_path = rollout_dir.join(format!( + "rollout-2026-08-15T16-00-00-{conversation_id}.jsonl" + )); + let lines = [ + serde_json::json!({ + "timestamp": "2026-08-15T08:00:00Z", + "type": "session_meta", + "payload": {"id": conversation_id, "cwd": "/tmp/Temp"} + }) + .to_string(), + serde_json::json!({ + "timestamp": "2026-08-15T08:00:01Z", + "type": "event_msg", + "payload": {"type": "user_message", "message": "Makefile 文件的作用"} + }) + .to_string(), + ]; + fs::write(&rollout_path, format!("{}\n", lines.join("\n"))).expect("write rollout"); + + let index_path = codex_home.join("session_index.jsonl"); + let parser = CodexParser::with_base_dir(sessions_dir); + (temp_dir, parser, index_path) + } + + #[test] + fn session_index_title_wins_for_list_and_detail() { + let conversation_id = "01a00496-1418-7273-a06f-dc4fae5cfa64"; + let (_temp_dir, parser, index_path) = write_index_title_fixture(conversation_id); + let index_lines = [ + serde_json::json!({ + "id": conversation_id, + "thread_name": "旧标题", + "updated_at": "2026-08-15T08:01:00Z" + }) + .to_string(), + "{malformed json".to_string(), + serde_json::json!({ + "id": conversation_id, + "thread_name": " 解释 Makefile 文件作用 ", + "updated_at": "2026-08-15T08:02:00Z" + }) + .to_string(), + serde_json::json!({ + "id": conversation_id, + "thread_name": " ", + "updated_at": "2026-08-15T08:03:00Z" + }) + .to_string(), + ]; + fs::write(&index_path, format!("{}\n", index_lines.join("\n"))).expect("write index"); + + let summaries = parser.list_conversations().expect("list conversations"); + assert_eq!(summaries.len(), 1); + assert_eq!( + summaries[0].title.as_deref(), + Some("解释 Makefile 文件作用") + ); + + let detail = parser + .get_conversation(conversation_id) + .expect("get conversation"); + assert_eq!( + detail.summary.title.as_deref(), + Some("解释 Makefile 文件作用") + ); + } + + #[test] + fn session_index_title_wins_over_rollout_thread_name_update() { + let conversation_id = "index-vs-rollout-title"; + let (_temp_dir, parser, index_path) = write_index_title_fixture(conversation_id); + let rollout_path = parser + .base_dir + .join("2026") + .join("08") + .join("15") + .join(format!( + "rollout-2026-08-15T16-00-00-{conversation_id}.jsonl" + )); + let mut rollout = fs::read_to_string(&rollout_path).expect("read rollout"); + rollout.push_str( + &serde_json::json!({ + "timestamp": "2026-08-15T08:00:02Z", + "type": "event_msg", + "payload": { + "type": "thread_name_updated", + "thread_name": "rollout thread title" + } + }) + .to_string(), + ); + rollout.push('\n'); + fs::write(&rollout_path, rollout).expect("append rollout title"); + fs::write( + &index_path, + format!( + "{}\n", + serde_json::json!({ + "id": conversation_id, + "thread_name": "session index title", + "updated_at": "2026-08-15T08:01:00Z" + }) + ), + ) + .expect("write index title"); + + let summaries = parser.list_conversations().expect("list conversations"); + assert_eq!(summaries[0].title.as_deref(), Some("session index title")); + let detail = parser + .get_conversation(conversation_id) + .expect("get conversation"); + assert_eq!(detail.summary.title.as_deref(), Some("session index title")); + } + + #[test] + fn session_index_title_refreshes_after_summary_cache_hit() { + let conversation_id = "cache-title-session"; + let (_temp_dir, parser, index_path) = write_index_title_fixture(conversation_id); + fs::write( + &index_path, + format!( + "{}\n", + serde_json::json!({"id": conversation_id, "thread_name": "索引标题甲"}) + ), + ) + .expect("write first index title"); + + let first = parser.list_conversations().expect("first list"); + assert_eq!(first[0].title.as_deref(), Some("索引标题甲")); + + // The rollout file is unchanged, so its summary is served from cache. + // The independently-read index must still replace the cached title. + fs::write( + &index_path, + format!( + "{}\n", + serde_json::json!({"id": conversation_id, "thread_name": "索引标题乙"}) + ), + ) + .expect("update index title"); + let second = parser.list_conversations().expect("second list"); + assert_eq!(second[0].title.as_deref(), Some("索引标题乙")); + } + + #[test] + fn unavailable_session_index_preserves_rollout_title_fallback() { + let conversation_id = "index-fallback-session"; + let (_temp_dir, parser, index_path) = write_index_title_fixture(conversation_id); + + let missing = parser.list_conversations().expect("list without index"); + assert_eq!(missing[0].title.as_deref(), Some("Makefile 文件的作用")); + + fs::write( + &index_path, + format!( + "not json\n{}\n", + serde_json::json!({"id": conversation_id, "thread_name": " "}) + ), + ) + .expect("write unusable index records"); + let malformed = parser.list_conversations().expect("list malformed index"); + assert_eq!(malformed[0].title.as_deref(), Some("Makefile 文件的作用")); + + fs::remove_file(&index_path).expect("remove index file"); + fs::create_dir(&index_path).expect("make index path unreadable as a file"); + let unreadable = parser.list_conversations().expect("list unreadable index"); + assert_eq!(unreadable[0].title.as_deref(), Some("Makefile 文件的作用")); + let detail = parser + .get_conversation(conversation_id) + .expect("detail unreadable index"); + assert_eq!(detail.summary.title.as_deref(), Some("Makefile 文件的作用")); + } + #[test] fn skips_agents_instructions_title_candidate() { let input = diff --git a/src-tauri/src/web/handlers/conversations.rs b/src-tauri/src/web/handlers/conversations.rs index 0912e3b66..a46a20fc1 100644 --- a/src-tauri/src/web/handlers/conversations.rs +++ b/src-tauri/src/web/handlers/conversations.rs @@ -26,12 +26,16 @@ pub async fn list_all_conversations( Ok(Json( conv_commands::list_all_conversations_core( &state.db.conn, - params.folder_ids, - params.agent_type, - params.search, - params.sort_by, - params.status, - params.include_children.unwrap_or(false), + &state.emitter, + &state.chat_channel_manager, + conv_commands::ListAllConversationsOptions { + folder_ids: params.folder_ids, + agent_type: params.agent_type, + search: params.search, + sort_by: params.sort_by, + status: params.status, + include_children: params.include_children.unwrap_or(false), + }, ) .await?, )) @@ -202,6 +206,7 @@ pub async fn import_local_conversations( conv_commands::import_local_conversations_core( &state.db.conn, &state.emitter, + &state.chat_channel_manager, params.folder_id, ) .await?, @@ -212,7 +217,12 @@ pub async fn scan_importable_sessions( Extension(state): Extension>, ) -> Result, AppCommandError> { Ok(Json( - conv_commands::scan_importable_sessions_core(&state.db.conn, &state.emitter).await?, + conv_commands::scan_importable_sessions_core( + &state.db.conn, + &state.emitter, + &state.chat_channel_manager, + ) + .await?, )) } diff --git a/src-tauri/src/web/mod.rs b/src-tauri/src/web/mod.rs index d48070368..ec2682db9 100644 --- a/src-tauri/src/web/mod.rs +++ b/src-tauri/src/web/mod.rs @@ -785,7 +785,16 @@ pub(crate) async fn do_start_web_server_tauri( &app.path().app_data_dir().unwrap_or_default(), ), web_server_state: WebServerState::new(), // placeholder; not used by handlers - chat_channel_manager: crate::app_state::default_chat_channel_manager(), + // Reuse the manager Tauri registered and started, NOT a fresh one: a + // fresh `ChatChannelManager` has an empty channel registry, so every + // handler that renames a bound chat thread (`update_conversation_title`, + // the import/scan/list title refreshes) would fail with `NotFound` and + // silently never retry — the DB has already converged by then. Falls + // back to a standalone manager only if the state is somehow absent. + chat_channel_manager: app + .try_state::() + .map(|state| state.inner().clone_ref()) + .unwrap_or_else(crate::app_state::default_chat_channel_manager), workspace_transfer: app .try_state::>() .map(|state| state.inner().clone()) diff --git a/src/components/settings/channel-events-tab.test.tsx b/src/components/settings/channel-events-tab.test.tsx index bbc92b79b..10bd6b604 100644 --- a/src/components/settings/channel-events-tab.test.tsx +++ b/src/components/settings/channel-events-tab.test.tsx @@ -56,11 +56,11 @@ describe("ChannelEventsTab event filter (opt-in user_prompt_sent)", () => { it("defaults user_prompt_sent OFF under a null filter while other events stay ON", async () => { mockGetFilter.mockResolvedValue(null) renderTab() - await waitFor(() => expect(mockGetFilter).toHaveBeenCalled()) + const userMessageSwitch = await screen.findByRole("switch", { + name: "User Message", + }) - expect( - screen.getByRole("switch", { name: "User Message" }) - ).not.toBeChecked() + expect(userMessageSwitch).not.toBeChecked() expect(screen.getByRole("switch", { name: "Turn Complete" })).toBeChecked() expect( screen.getByRole("switch", { name: "Permission Request" }) @@ -70,9 +70,11 @@ describe("ChannelEventsTab event filter (opt-in user_prompt_sent)", () => { it("enabling user_prompt_sent persists an explicit list including it (never null)", async () => { mockGetFilter.mockResolvedValue(null) renderTab() - await waitFor(() => expect(mockGetFilter).toHaveBeenCalled()) + const userMessageSwitch = await screen.findByRole("switch", { + name: "User Message", + }) - fireEvent.click(screen.getByRole("switch", { name: "User Message" })) + fireEvent.click(userMessageSwitch) await waitFor(() => expect(mockSetFilter).toHaveBeenCalled()) const calls = mockSetFilter.mock.calls @@ -174,9 +176,11 @@ describe("ChannelEventsTab webhooks", () => { it("adds a webhook through the dialog and persists it as enabled", async () => { renderTab() - await waitFor(() => expect(mockGetWebhooks).toHaveBeenCalled()) + const addWebhook = await screen.findByRole("button", { + name: "Add Webhook", + }) - fireEvent.click(screen.getByRole("button", { name: "Add Webhook" })) + fireEvent.click(addWebhook) fireEvent.change( screen.getByPlaceholderText("https://example.com/webhook"), { target: { value: "https://hook.test/in" } } @@ -266,9 +270,11 @@ describe("ChannelEventsTab webhooks", () => { it("rejects an invalid url in the dialog without persisting", async () => { renderTab() - await waitFor(() => expect(mockGetWebhooks).toHaveBeenCalled()) + const addWebhook = await screen.findByRole("button", { + name: "Add Webhook", + }) - fireEvent.click(screen.getByRole("button", { name: "Add Webhook" })) + fireEvent.click(addWebhook) fireEvent.change( screen.getByPlaceholderText("https://example.com/webhook"), { target: { value: "not-a-url" } } @@ -321,9 +327,11 @@ describe("ChannelEventsTab webhooks", () => { }) ) renderTab() - await waitFor(() => expect(mockGetWebhooks).toHaveBeenCalled()) + const addWebhook = await screen.findByRole("button", { + name: "Add Webhook", + }) - fireEvent.click(screen.getByRole("button", { name: "Add Webhook" })) + fireEvent.click(addWebhook) const input = screen.getByPlaceholderText("https://example.com/webhook") fireEvent.change(input, { target: { value: "https://b.test/h" } }) fireEvent.click(screen.getByRole("button", { name: "Save" }))