diff --git a/src-tauri/src/acp/connection.rs b/src-tauri/src/acp/connection.rs index 2075243314..24d6cc2b55 100644 --- a/src-tauri/src/acp/connection.rs +++ b/src-tauri/src/acp/connection.rs @@ -1992,7 +1992,7 @@ fn current_config_option_values( .collect() } -fn current_model_id_from_opts(opts: &[SessionConfigOptionInfo]) -> Option { +pub(crate) fn current_model_id_from_opts(opts: &[SessionConfigOptionInfo]) -> Option { opts.iter() .find(|o| o.category.as_deref() == Some("model")) .and_then(|o| { diff --git a/src-tauri/src/acp/lifecycle.rs b/src-tauri/src/acp/lifecycle.rs index 8ecbf26086..5818cfa1ae 100644 --- a/src-tauri/src/acp/lifecycle.rs +++ b/src-tauri/src/acp/lifecycle.rs @@ -333,16 +333,25 @@ pub(crate) async fn handle_event( else { return Ok(()); }; - let (conversation_id, last_text, claimed) = { + let (conversation_id, agent_type, external_id, last_text, claimed, selector_model) = { let mut snap = state_arc.write().await; // Re-entrant for this worker: a retry of this same envelope // claims again and goes on. Only a CLI run's fallback that // already wrote this turn's status makes it skip. let claimed = snap.claim_turn_status(envelope.seq, TurnStatusWriter::Lifecycle); + // Same snapshot as the ids above: the fallback for a store + // that names no model. + let selector_model = snap + .config_options + .as_deref() + .and_then(crate::acp::connection::current_model_id_from_opts); ( snap.conversation_id, + snap.agent_type, + snap.external_id.clone(), snap.last_assistant_text.clone(), claimed, + selector_model, ) }; // No conversation row bound (defensive — should never happen in @@ -351,9 +360,23 @@ pub(crate) async fn handle_event( let Some(cid) = conversation_id else { return Ok(()); }; + // The turn status goes out FIRST: it carries only the id and the + // status. The model chip is a separate upsert after that write. if claimed { write_turn_status(db_conn, &state_arc, &emitter, cid, stop_reason).await?; } + // Cursor does not put the model on the conversation row. Copy the + // model this turn recorded in the store onto an empty chip, or the + // selector's when the store names none. + crate::commands::conversations::seed_cursor_model_after_turn( + db_conn, + &emitter, + cid, + agent_type, + external_id.as_deref(), + selector_model.as_deref(), + ) + .await; // If this conversation was spawned by a delegation, resolve the // pending broker call. The broker maps the outcome onto the @@ -5207,4 +5230,192 @@ mod tests { drop(bus); let _ = dispatcher.await; } + + fn cursor_model_selector(current: &str) -> crate::acp::types::SessionConfigOptionInfo { + crate::acp::types::SessionConfigOptionInfo { + id: "model".into(), + name: "Model".into(), + description: None, + category: Some("model".into()), + kind: crate::acp::types::SessionConfigKindInfo::Select( + crate::acp::types::SessionConfigSelectInfo { + current_value: current.into(), + options: Vec::new(), + groups: Vec::new(), + }, + ), + recommended_value: None, + } + } + + /// Points the Cursor agent at `config_dir` so the turn-end store lookup + /// never reads the real `~/.cursor`. + async fn point_cursor_at(db: &crate::db::AppDatabase, config_dir: &std::path::Path) { + use crate::db::service::agent_setting_service; + agent_setting_service::ensure_defaults( + &db.conn, + &[agent_setting_service::AgentDefaultInput { + agent_type: AgentType::Cursor, + registry_id: AgentType::Cursor.to_string(), + default_sort_order: 0, + }], + ) + .await + .expect("seed agent setting"); + agent_setting_service::update( + &db.conn, + AgentType::Cursor, + agent_setting_service::AgentSettingsUpdate { + enabled: true, + env_json: Some( + serde_json::json!({ "CURSOR_CONFIG_DIR": config_dir.to_str().unwrap() }) + .to_string(), + ), + model_provider_id: None, + }, + ) + .await + .expect("save env_json"); + let loaded = agent_setting_service::load_agent_env(&db.conn, AgentType::Cursor).await; + assert_eq!( + loaded.var("CURSOR_CONFIG_DIR").as_deref(), + config_dir.to_str(), + "the lookup must stay inside the temp dir" + ); + } + + fn cursor_turn_complete(connection_id: &str, session_id: &str, seq: u64) -> EventEnvelope { + EventEnvelope { + seq, + connection_id: connection_id.to_string(), + payload: AcpEvent::TurnComplete { + session_id: session_id.into(), + stop_reason: "end_turn".into(), + agent_type: "cursor".into(), + }, + } + } + + async fn bound_cursor_row( + db: &crate::db::AppDatabase, + folder_id: i32, + mgr: &ConnectionManager, + connection_id: &str, + session_id: &str, + ) -> (i32, Arc>) { + let conv = conversation_service::create(&db.conn, folder_id, AgentType::Cursor, None, None) + .await + .unwrap(); + conversation_service::bind_external_id(&db.conn, conv.id, session_id, &[]) + .await + .unwrap(); + let conn = fake_connection_with_state(connection_id, Some(conv.id)); + let state = conn.state.clone(); + { + let mut snap = state.write().await; + snap.agent_type = AgentType::Cursor; + snap.external_id = Some(session_id.into()); + snap.config_options = None; + } + mgr.connections + .lock() + .await + .insert(connection_id.to_string(), conn); + (conv.id, state) + } + + async fn row_model(db: &crate::db::AppDatabase, conv_id: i32) -> Option { + conversation_service::get_by_id(&db.conn, conv_id) + .await + .unwrap() + .model + .filter(|model| !model.trim().is_empty()) + } + + #[tokio::test] + async fn handle_event_seeds_an_empty_cursor_model_from_the_store_then_the_selector() { + let db = test_helpers::fresh_in_memory_db().await; + let config_dir = tempfile::tempdir().unwrap(); + point_cursor_at(&db, config_dir.path()).await; + let folder_id = test_helpers::seed_folder(&db, "/tmp/cursor-model-seed").await; + let mgr = ConnectionManager::new(); + + // Row A: its store records the variant that ran. + let session_a = "cursor-sess-store"; + let (conv_a, state_a) = bound_cursor_row(&db, folder_id, &mgr, "c1", session_a).await; + + handle_event( + &db.conn, + &mgr, + &cursor_turn_complete("c1", session_a, 1), + None, + ) + .await + .unwrap(); + assert_eq!( + row_model(&db, conv_a).await, + None, + "no store and no selector means no model is invented" + ); + assert_eq!( + read_row_status(&db, conv_a).await, + ConversationStatus::PendingReview, + "turn status is written before the model seed" + ); + + crate::parsers::cursor::write_model_session_for_test( + config_dir.path(), + session_a, + Some("grok-4.7-high"), + ); + // The parameterized picker names only the base model. + state_a.write().await.config_options = Some(vec![cursor_model_selector("grok-4.7")]); + handle_event( + &db.conn, + &mgr, + &cursor_turn_complete("c1", session_a, 2), + None, + ) + .await + .unwrap(); + assert_eq!( + row_model(&db, conv_a).await.as_deref(), + Some("grok-4.7-high"), + "the store's variant wins over the selector's base id, matching the backfill" + ); + + crate::parsers::cursor::write_model_session_for_test( + config_dir.path(), + session_a, + Some("grok-4.7-xhigh-fast"), + ); + handle_event( + &db.conn, + &mgr, + &cursor_turn_complete("c1", session_a, 3), + None, + ) + .await + .unwrap(); + assert_eq!( + row_model(&db, conv_a).await.as_deref(), + Some("grok-4.7-high"), + "a later model does not overwrite a chip that is already filled" + ); + + // Row B: its store names no model, so the selector fills the chip. + let session_b = "cursor-sess-selector"; + let (conv_b, state_b) = bound_cursor_row(&db, folder_id, &mgr, "c2", session_b).await; + crate::parsers::cursor::write_model_session_for_test(config_dir.path(), session_b, None); + state_b.write().await.config_options = Some(vec![cursor_model_selector("grok-4.7")]); + handle_event( + &db.conn, + &mgr, + &cursor_turn_complete("c2", session_b, 1), + None, + ) + .await + .unwrap(); + assert_eq!(row_model(&db, conv_b).await.as_deref(), Some("grok-4.7")); + } } diff --git a/src-tauri/src/commands/conversations.rs b/src-tauri/src/commands/conversations.rs index a21907626b..4c31cc343d 100644 --- a/src-tauri/src/commands/conversations.rs +++ b/src-tauri/src/commands/conversations.rs @@ -18,6 +18,7 @@ use crate::models::*; // index reader and not part of the `AgentParser` trait. Every history read goes // through `build_agent_parser`. use crate::parsers::codex::CodexParser; +use crate::parsers::cursor::CursorParser; use crate::parsers::{ build_agent_parser, folder_name_from_path, normalize_path_for_matching, path_eq_for_matching, AgentEnvs, AgentParser, ParseError, @@ -59,6 +60,13 @@ pub(crate) async fn list_all_conversations_core( HashMap::new() } }; + // Cursor's conversation row is created before the store names a model, and + // current stores have no `lastUsedModel`. Fill a blank chip from the + // assistant message's `modelName`, in the background, off the request + // path. Search hits this endpoint after every 300 ms debounce, so an + // unresolved store must never delay unrelated list results. Once a row + // has a model the query finds nothing and no store is opened again. + schedule_cursor_model_backfill(conn, emitter); list_all_conversations_core_with_codex_titles( conn, emitter, @@ -69,6 +77,215 @@ pub(crate) async fn list_all_conversations_core( .await } +/// Shared state for the background Cursor model backfill. `running` makes the +/// scheduled pass single-flight (a 300 ms-debounced search must not stack +/// walks). `failed_at_mtime` is the negative cache: a session whose store +/// gave no model is not walked again until `store.db` or `store.db-wal` +/// changes — without it every list request would re-read the same +/// unresolvable history. A lookup whose mtime could not be read is not +/// cached, so a transient stat failure is retried next time. +#[derive(Default)] +struct CursorModelBackfillState { + running: bool, + failed_at_mtime: HashMap>, +} + +fn cursor_model_backfill_state() -> &'static std::sync::Mutex { + static STATE: std::sync::OnceLock> = + std::sync::OnceLock::new(); + STATE.get_or_init(|| std::sync::Mutex::new(CursorModelBackfillState::default())) +} + +fn schedule_cursor_model_backfill(conn: &sea_orm::DatabaseConnection, emitter: &EventEmitter) { + { + let mut state = cursor_model_backfill_state() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if state.running { + return; + } + state.running = true; + } + // Clears `running` when the pass ends however it ends: an `Err`, a panic + // that unwinds the task, or the task being dropped. A flag left set would + // stop every later backfill for the life of the process. + struct RunningGuard; + impl Drop for RunningGuard { + fn drop(&mut self) { + cursor_model_backfill_state() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .running = false; + } + } + let guard = RunningGuard; + let conn = conn.clone(); + let emitter = emitter.clone(); + tokio::spawn(async move { + let _guard = guard; + if let Err(error) = backfill_cursor_models(&conn, &emitter).await { + tracing::warn!( + error = %error, + "conversation list: failed to backfill Cursor session models; continuing" + ); + } + }); +} + +async fn backfill_cursor_models( + conn: &sea_orm::DatabaseConnection, + emitter: &EventEmitter, +) -> Result<(), AppCommandError> { + let missing = conversation_service::cursor_sessions_missing_model(conn) + .await + .map_err(AppCommandError::from)?; + if missing.is_empty() { + return Ok(()); + } + let cursor_env = agent_setting_service::load_agent_env(conn, AgentType::Cursor).await; + let resolved = tokio::task::spawn_blocking(move || { + let parser = CursorParser::for_env(&cursor_env); + let mut found = Vec::new(); + let mut failed = Vec::new(); + for (id, external_id) in missing { + let mtime = parser.store_mtime_for_session(&external_id); + { + let state = cursor_model_backfill_state() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if state.failed_at_mtime.get(&external_id) == Some(&mtime) { + // Store unchanged since the last unsuccessful walk. + continue; + } + } + match parser.model_for_session(&external_id) { + Some(model) => found.push((id, external_id, model)), + None if mtime.is_some() => failed.push((external_id, mtime)), + None => {} + } + } + (found, failed) + }) + .await + .map_err(|error| { + AppCommandError::task_execution_failed("Failed to read Cursor session models") + .with_detail(error.to_string()) + })?; + let (found, failed) = resolved; + { + let mut state = cursor_model_backfill_state() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + for (external_id, mtime) in failed { + state.failed_at_mtime.insert(external_id, mtime); + } + } + for (id, external_id, model) in found { + // Match the session id in the same UPDATE: if the row was rebound to + // another session during the store walk, this write must no-op rather + // than pin the old session's model on the replacement. + if conversation_service::seed_model_if_empty_for_session(conn, id, &external_id, &model) + .await + .map_err(AppCommandError::from)? + { + { + let mut state = cursor_model_backfill_state() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + state.failed_at_mtime.remove(&external_id); + } + emit_conversation_upsert(emitter, conn, id).await; + } + } + Ok(()) +} + +/// After a Cursor turn, copy the session's model onto the conversation row +/// when the row still has none, and broadcast the sidebar chip. +/// +/// The store comes first, read with the same parser as history and the +/// background backfill, so a live row and a backfilled row carry the same +/// id. Cursor's selector names only the base model (`grok-4.7`) in the +/// parameterized picker codeg asks for, while each reply records the variant +/// that ran (`grok-4.7-high`). Cursor writes the reply's blobs before it +/// answers the prompt, so the store already has this turn. That lookup reads +/// the newest message refs only, never the blob graph. The selector is the +/// fallback for a store that names no model. +/// +/// A no-op for every other agent, and for a Cursor row that already has a +/// model: filling an empty column does not follow a later switch. +pub(crate) async fn seed_cursor_model_after_turn( + conn: &sea_orm::DatabaseConnection, + emitter: &EventEmitter, + conversation_id: i32, + agent_type: AgentType, + external_id: Option<&str>, + selector_model: Option<&str>, +) { + if agent_type != AgentType::Cursor { + return; + } + let Some(external_id) = external_id.map(str::trim).filter(|id| !id.is_empty()) else { + return; + }; + // Every later turn of a session lands here too; once the chip is filled + // there is nothing to look up. + match conversation_service::get_by_id(conn, conversation_id).await { + Ok(summary) + if summary + .model + .as_deref() + .is_some_and(|model| !model.trim().is_empty()) => + { + return; + } + Ok(_) => {} + Err(_) => return, + } + let cursor_env = agent_setting_service::load_agent_env(conn, AgentType::Cursor).await; + let session_id = external_id.to_string(); + let store_model = match tokio::task::spawn_blocking(move || { + CursorParser::for_env(&cursor_env).model_for_session(&session_id) + }) + .await + { + Ok(model) => model, + Err(error) => { + tracing::warn!( + conversation_id, + error = %error, + "cursor model chip: failed to read the session store" + ); + None + } + }; + let selector_model = selector_model + .map(str::trim) + .filter(|model| !model.is_empty()) + .map(str::to_string); + let Some(model) = store_model.or(selector_model) else { + return; + }; + // Match the external id in the same UPDATE: a row rebound to another + // session must not inherit this session's model. + match conversation_service::seed_model_if_empty_for_session( + conn, + conversation_id, + external_id, + &model, + ) + .await + { + Ok(true) => emit_conversation_upsert(emitter, conn, conversation_id).await, + Ok(false) => {} + Err(error) => tracing::warn!( + conversation_id, + error = %error, + "cursor model chip: failed to store the model" + ), + } +} + async fn list_all_conversations_core_with_codex_titles( conn: &sea_orm::DatabaseConnection, emitter: &EventEmitter, @@ -7610,4 +7827,132 @@ mod tests { crate::commands::turn_window::prefix_fingerprint(&[]) ); } + + #[tokio::test] + async fn backfill_retries_a_failed_lookup_when_only_the_wal_mtime_changes() { + let db = fresh_in_memory_db().await; + let tmp = tempfile::tempdir().unwrap(); + let session = "0198c9aa-aaaa-bbbb-cccc-walretry0001"; + let config_dir = tmp.path().to_str().unwrap(); + save_agent_env_json( + &db, + AgentType::Cursor, + &serde_json::json!({ "CURSOR_CONFIG_DIR": config_dir }).to_string(), + ) + .await; + let loaded = agent_setting_service::load_agent_env(&db.conn, AgentType::Cursor).await; + assert_eq!( + loaded.var("CURSOR_CONFIG_DIR").as_deref(), + Some(config_dir), + "the lookup must stay inside the temp dir" + ); + + let folder_id = seed_folder(&db, "/tmp/cursor-wal-retry").await; + let conv_id = create_conversation_core(&db.conn, folder_id, AgentType::Cursor, None) + .await + .unwrap(); + conversation_service::bind_external_id(&db.conn, conv_id, session, &[]) + .await + .unwrap(); + let missing_id = create_conversation_core(&db.conn, folder_id, AgentType::Cursor, None) + .await + .unwrap(); + conversation_service::bind_external_id( + &db.conn, + missing_id, + "missing-session-not-on-disk", + &[], + ) + .await + .unwrap(); + cursor_model_backfill_state() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .failed_at_mtime + .remove(session); + + let db_path = + crate::parsers::cursor::write_model_session_for_test(tmp.path(), session, None); + let old = std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000); + std::fs::File::options() + .write(true) + .open(&db_path) + .unwrap() + .set_modified(old) + .unwrap(); + let old = std::fs::metadata(&db_path).unwrap().modified().unwrap(); + + backfill_cursor_models(&db.conn, &EventEmitter::Noop) + .await + .unwrap(); + let missed = conversation_service::get_by_id(&db.conn, conv_id) + .await + .unwrap(); + assert!(missed.model.as_deref().unwrap_or("").trim().is_empty()); + assert!( + !cursor_model_backfill_state() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .failed_at_mtime + .contains_key("missing-session-not-on-disk"), + "a lookup with no readable mtime must not stay cached" + ); + assert_eq!( + cursor_model_backfill_state() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .failed_at_mtime + .get(session) + .cloned(), + Some(Some(old)) + ); + + let db_path = + crate::parsers::cursor::write_model_session_for_test(tmp.path(), session, Some("grok-4.7-high")); + std::fs::File::options() + .write(true) + .open(&db_path) + .unwrap() + .set_modified(old) + .unwrap(); + assert_eq!( + std::fs::metadata(&db_path).unwrap().modified().unwrap(), + old, + "store.db was rewritten, then its mtime was put back" + ); + // The model is already in the new store.db. A cache that ignores + // mtime would read it now. The restored mtime must keep this a miss. + backfill_cursor_models(&db.conn, &EventEmitter::Noop) + .await + .unwrap(); + let still_cached = conversation_service::get_by_id(&db.conn, conv_id) + .await + .unwrap(); + assert!( + still_cached.model.as_deref().unwrap_or("").trim().is_empty(), + "same mtime must not look the store up again" + ); + + let wal = db_path.with_file_name("store.db-wal"); + std::fs::write(&wal, b"wal").unwrap(); + std::fs::File::options() + .write(true) + .open(&wal) + .unwrap() + .set_modified(old + std::time::Duration::from_secs(60)) + .unwrap(); + + backfill_cursor_models(&db.conn, &EventEmitter::Noop) + .await + .unwrap(); + let seeded = conversation_service::get_by_id(&db.conn, conv_id) + .await + .unwrap(); + assert_eq!(seeded.model.as_deref(), Some("grok-4.7-high")); + cursor_model_backfill_state() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .failed_at_mtime + .remove(session); + } } diff --git a/src-tauri/src/db/service/conversation_service.rs b/src-tauri/src/db/service/conversation_service.rs index 8551a705a1..b71c4451e2 100644 --- a/src-tauri/src/db/service/conversation_service.rs +++ b/src-tauri/src/db/service/conversation_service.rs @@ -307,6 +307,97 @@ pub async fn seed_model_if_empty( Ok(res.rows_affected > 0) } +/// Cursor rows whose sidebar chip is blank, with the session id needed to +/// read the model out of that session's store. Deleted rows and rows that +/// already have a model are not returned. Rows in soft-deleted folders and +/// loop conversations are excluded too: the workspace list hides them (same +/// filter as `refresh_codex_auto_titles`), so a successful backfill must not +/// broadcast an upsert that reinserts a hidden row into another client's +/// workspace state. +pub async fn cursor_sessions_missing_model( + conn: &DatabaseConnection, +) -> Result, DbError> { + let rows = conversation::Entity::find() + .filter(conversation::Column::DeletedAt.is_null()) + .filter(conversation::Column::AgentType.eq(AgentType::Cursor.as_wire().into_owned())) + .filter(conversation::Column::ExternalId.is_not_null()) + .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(), + ), + ) + .filter( + sea_orm::Condition::any() + .add(conversation::Column::Model.is_null()) + .add(conversation::Column::Model.eq("")), + ) + .all(conn) + .await?; + Ok(rows + .into_iter() + .filter_map(|row| { + let external_id = row.external_id.filter(|id| !id.trim().is_empty())?; + Some((row.id, external_id)) + }) + .collect()) +} + +/// [`seed_model_if_empty`] plus an atomic identity check: the row must still +/// be bound to the session the model was read from. A model lookup walks the +/// session store off the async runtime; if `bind_external_id` rebinds the row +/// to a different session during that walk, an ID-only write would persist +/// the previous session's model on the replacement (and later backfills would +/// then skip it). Matching the external id in the same UPDATE turns that race +/// into a no-op. +/// +/// The same UPDATE also requires a live folder and a non-loop row, the filter +/// [`cursor_sessions_missing_model`] selects with. Both callers broadcast an +/// upsert when this returns `true`, and the client inserts any upserted row it +/// does not have. A folder soft-deleted while the store was read, or a loop +/// conversation finishing a Cursor turn, must not come back into another +/// client's workspace list that way. +pub async fn seed_model_if_empty_for_session( + conn: &DatabaseConnection, + conversation_id: i32, + external_id: &str, + model: &str, +) -> Result { + use sea_orm::sea_query::Expr; + let model = model.trim(); + if model.is_empty() || external_id.trim().is_empty() { + return Ok(false); + } + let res = conversation::Entity::update_many() + .col_expr(conversation::Column::Model, Expr::value(model)) + .filter(conversation::Column::Id.eq(conversation_id)) + .filter(conversation::Column::DeletedAt.is_null()) + .filter(conversation::Column::ExternalId.eq(external_id)) + .filter(conversation::Column::AgentType.eq(AgentType::Cursor.as_wire().into_owned())) + .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(), + ), + ) + .filter( + sea_orm::Condition::any() + .add(conversation::Column::Model.is_null()) + .add(conversation::Column::Model.eq("")), + ) + .exec(conn) + .await?; + Ok(res.rows_affected > 0) +} + /// Lock a row's title WITHOUT rewriting it. For a conversation whose name was /// typed by the user somewhere else — a work task's title, an automation's name /// — the seed passed to [`create`] already IS the name; all that's missing is @@ -1837,6 +1928,78 @@ mod tests { ); } + #[tokio::test] + async fn seed_model_for_session_skips_rows_the_workspace_list_hides() { + let db = fresh_in_memory_db().await; + let live_folder = seed_folder(&db, "/tmp/codeg-seed-model-live").await; + let dead_folder = seed_folder(&db, "/tmp/codeg-seed-model-dead").await; + let cursor_row = |folder_id: i32, session: &'static str| { + let conn = db.conn.clone(); + async move { + let row = create(&conn, folder_id, AgentType::Cursor, None, None) + .await + .expect("create"); + bind_external_id(&conn, row.id, session, &[]) + .await + .expect("bind"); + row.id + } + }; + + let visible = cursor_row(live_folder, "cursor-visible").await; + let in_dead_folder = cursor_row(dead_folder, "cursor-dead-folder").await; + let loop_row = cursor_row(live_folder, "cursor-loop").await; + let mut active: conversation::ActiveModel = conversation::Entity::find_by_id(loop_row) + .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"); + // Soft-deleted after the row was picked, as if during the store read. + crate::db::service::folder_service::soft_delete_folder(&db.conn, dead_folder) + .await + .expect("soft delete folder"); + + assert!( + seed_model_if_empty_for_session(&db.conn, visible, "cursor-visible", "grok-4.7-high") + .await + .expect("seed visible"), + "a live row is seeded" + ); + assert!( + !seed_model_if_empty_for_session( + &db.conn, + in_dead_folder, + "cursor-dead-folder", + "grok-4.7-high" + ) + .await + .expect("seed dead folder"), + "a row in a soft-deleted folder is not seeded, so no upsert brings it back" + ); + assert!( + !seed_model_if_empty_for_session(&db.conn, loop_row, "cursor-loop", "grok-4.7-high") + .await + .expect("seed loop"), + "a loop row is not seeded, so no upsert puts it in the workspace list" + ); + for (id, expected) in [ + (visible, Some("grok-4.7-high")), + (in_dead_folder, None), + (loop_row, None), + ] { + let model = conversation::Entity::find_by_id(id) + .one(&db.conn) + .await + .expect("query") + .expect("row") + .model; + assert_eq!(model.as_deref(), expected, "row {id}"); + } + } + #[tokio::test] async fn update_pin_sets_and_clears_without_bumping_updated_at() { let db = fresh_in_memory_db().await; diff --git a/src-tauri/src/parsers/cursor.rs b/src-tauri/src/parsers/cursor.rs index 46aa75a7b1..930bab792f 100644 --- a/src-tauri/src/parsers/cursor.rs +++ b/src-tauri/src/parsers/cursor.rs @@ -166,7 +166,10 @@ impl CursorParser { .as_deref() .map(title_from_user_text) }); - Some(summary_from(chat_id, &meta, &state, &sidecar, title)) + let model = resolved_session_model(&conn, &meta); + Some(summary_from( + chat_id, &meta, &state, &sidecar, title, model, + )) } fn build_detail(&self, chat_dir: &Path, chat_id: &str) -> Option { @@ -223,7 +226,8 @@ impl CursorParser { }); let sidecar = read_sidecar_meta(chat_dir); let title = meta.name.clone().or_else(|| sidecar.title.clone()).or(title); - let summary = summary_from(chat_id, &meta, &state, &sidecar, title); + let model = resolved_session_model(&conn, &meta); + let summary = summary_from(chat_id, &meta, &state, &sidecar, title, model); Some(ConversationDetail { summary, @@ -265,6 +269,35 @@ impl CursorParser { out.push(summary); } } + + /// Model of one Cursor session, for the sidebar chip. + /// + /// `lastUsedModel` when the store still has it. Current ACP stores do + /// not: the model is `content[].providerOptions.cursor.modelName` on an + /// assistant message. The latest such message wins. A blob that the + /// latest root does not reference is ignored. + pub(crate) fn model_for_session(&self, session_id: &str) -> Option { + let chat_dir = self.find_chat_dir(session_id)?; + let conn = open_store(&chat_dir.join("store.db"))?; + let meta = read_chat_meta(&conn)?; + resolved_session_model(&conn, &meta) + } + + /// Modification time of the session's store, for negative caching: a + /// failed model lookup is retried only after the store actually changed. + /// + /// Cursor writes in WAL mode, so a turn can land in `store.db-wal` while + /// `store.db` keeps its old mtime. The newer of the two is the cache key. + /// A missing WAL is not an error; a file we cannot stat is left out, and + /// `None` means neither file could be read (the caller must not cache + /// that as a permanent miss). + pub(crate) fn store_mtime_for_session( + &self, + session_id: &str, + ) -> Option { + let chat_dir = self.find_chat_dir(session_id)?; + newer_mtime(&chat_dir.join("store.db"), &chat_dir.join("store.db-wal")) + } } impl Default for CursorParser { @@ -304,6 +337,7 @@ fn summary_from( state: &DecodedState, sidecar: &SidecarMeta, title: Option, + model: Option, ) -> ConversationSummary { // The sidecar cwd is authoritative for ACP sessions: their stores are // keyed by session uuid (no md5(cwd) bucket), and the workspace refs @@ -331,7 +365,7 @@ fn summary_from( ended_at, // user + assistant per agent turn (approximation before full decode). message_count: (state.turn_blob_ids.len() as u32).saturating_mul(2), - model: meta.last_used_model.clone(), + model, git_branch: state.git_branch.clone(), parent_id: None, parent_tool_use_id: None, @@ -474,6 +508,97 @@ fn json_ms(v: Option<&Value>) -> Option { .and_then(|dt| u64::try_from(dt.timestamp_millis()).ok()) } +/// `lastUsedModel` when the store has it. Otherwise the newest +/// `content[].providerOptions.cursor.modelName` on an assistant message +/// reachable from the latest root: field 1 in order, then an inline field 4. +/// A later assistant message that has no model name does not erase an earlier +/// one. This does not walk the blob graph. +fn resolved_session_model(conn: &Connection, meta: &ChatMeta) -> Option { + if let Some(model) = meta + .last_used_model + .as_deref() + .map(str::trim) + .filter(|model| !model.is_empty()) + { + return Some(model.to_string()); + } + let root_id = meta.latest_root_blob_id.as_deref()?; + let root = read_blob(conn, root_id)?; + assistant_model_from_root(conn, &root) +} + +/// How many field-1 message refs, counted from the end of a root, a lookup may +/// read. The latest assistant message is last, so the cap keeps the tail; it +/// only bounds a session whose recent messages name no model at all. Real +/// sessions are far smaller than this. +const ASSISTANT_MESSAGE_REF_CAP: usize = 4_000; + +fn assistant_model_from_root(conn: &Connection, root: &[u8]) -> Option { + let mut refs: Vec<&[u8]> = Vec::new(); + let mut inline: Option<&[u8]> = None; + for (field_no, val) in wire::Fields::new(root) { + match field_no { + 1 => { + if let Some(id) = val.bytes().filter(|id| id.len() == 32) { + refs.push(id); + } + } + 4 => { + if let Some(bytes) = val.bytes().filter(|bytes| !bytes.is_empty()) { + inline = Some(bytes); + } + } + _ => {} + } + } + // Newest first: the inline field-4 message, then the field-1 refs from + // the end. The first message that names a model is the answer, so a + // normal session reads one or two blobs rather than its whole history. + if let Some(model) = inline.and_then(model_name_of_assistant_message) { + return Some(model); + } + refs.iter() + .rev() + .take(ASSISTANT_MESSAGE_REF_CAP) + .find_map(|id| read_blob(conn, id).and_then(|blob| model_name_of_assistant_message(&blob))) +} + +/// Model id on one assistant message. Only `reasoning` / `redacted-reasoning` +/// entries carry `modelName` in real stores; `text` and `tool-call` omit it, +/// so entries without a non-empty name are skipped. The last non-empty name +/// in the content array wins. Anything that is not an assistant JSON message +/// — including a user turn, a broken blob, or a leftover fingerprint object — +/// contributes nothing. +fn model_name_of_assistant_message(bytes: &[u8]) -> Option { + let value: Value = serde_json::from_slice(bytes).ok()?; + if value.get("role").and_then(Value::as_str) != Some("assistant") { + return None; + } + let content = value.get("content").and_then(Value::as_array)?; + let mut found = None; + for item in content { + if let Some(name) = item + .pointer("/providerOptions/cursor/modelName") + .and_then(Value::as_str) + .map(str::trim) + .filter(|name| !name.is_empty()) + { + found = Some(name.to_string()); + } + } + found +} + +fn newer_mtime(db: &Path, wal: &Path) -> Option { + let db_mtime = std::fs::metadata(db).and_then(|meta| meta.modified()).ok(); + let wal_mtime = std::fs::metadata(wal).and_then(|meta| meta.modified()).ok(); + match (db_mtime, wal_mtime) { + (Some(db_mtime), Some(wal_mtime)) => Some(db_mtime.max(wal_mtime)), + (Some(mtime), None) | (None, Some(mtime)) => Some(mtime), + (None, None) => None, + } +} + fn read_blob(conn: &Connection, id: &[u8]) -> Option> { conn.query_row( "SELECT data FROM blobs WHERE id = ?1", @@ -1961,6 +2086,86 @@ fn read_subdirs(dir: &Path) -> Vec { .collect() } +/// Writes one ACP session store under `config_dir` for tests. `model` is the +/// assistant `modelName`; `None` writes a reply that has no model. The file +/// is replaced if it already exists. Tests must pass a temp directory. +#[cfg(test)] +pub(crate) fn write_model_session_for_test( + config_dir: &Path, + session_id: &str, + model: Option<&str>, +) -> PathBuf { + fn push_varint(out: &mut Vec, mut value: u64) { + loop { + let byte = (value & 0x7f) as u8; + value >>= 7; + if value == 0 { + out.push(byte); + break; + } + out.push(byte | 0x80); + } + } + fn push_bytes(out: &mut Vec, field: u64, data: &[u8]) { + push_varint(out, (field << 3) | 2); + push_varint(out, data.len() as u64); + out.extend_from_slice(data); + } + + let dir = config_dir.join("acp-sessions").join(session_id); + std::fs::create_dir_all(&dir).expect("session dir"); + let path = dir.join("store.db"); + let _ = std::fs::remove_file(&path); + let conn = Connection::open(&path).expect("store"); + conn.execute_batch( + "CREATE TABLE blobs (id TEXT PRIMARY KEY, data BLOB); + CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT);", + ) + .expect("schema"); + let message = match model { + Some(model) => serde_json::json!({ + "role": "assistant", + "content": [ + {"type": "reasoning", "providerOptions": {"cursor": {"modelName": model}}}, + {"type": "text", "text": "ok"}, + {"type": "tool-call"} + ] + }), + None => serde_json::json!({ + "role": "assistant", + "content": [ + {"type": "text", "text": "ok"}, + {"type": "tool-call"} + ] + }), + }; + let message_bytes = serde_json::to_vec(&message).expect("message"); + let message_id = [0x11u8; 32]; + let root_id = [0x22u8; 32]; + let mut root = Vec::new(); + push_bytes(&mut root, 1, &message_id); + conn.execute( + "INSERT INTO blobs (id, data) VALUES (?1, ?2)", + rusqlite::params![hex_encode(&message_id), message_bytes], + ) + .expect("message blob"); + conn.execute( + "INSERT INTO blobs (id, data) VALUES (?1, ?2)", + rusqlite::params![hex_encode(&root_id), root], + ) + .expect("root blob"); + let meta = serde_json::json!({ + "latestRootBlobId": hex_encode(&root_id), + }); + let meta_hex = hex_encode(meta.to_string().as_bytes()); + conn.execute( + "INSERT INTO meta (key, value) VALUES ('0', ?1)", + [meta_hex], + ) + .expect("meta"); + path +} + #[cfg(test)] mod tests { use super::*; @@ -2012,6 +2217,16 @@ mod tests { Self { conn, next_id: 1 } } + fn put_blob_exact(&self, id: &[u8], data: &[u8]) -> Vec { + self.conn + .execute( + "INSERT OR REPLACE INTO blobs (id, data) VALUES (?1, ?2)", + rusqlite::params![hex_encode(id), data], + ) + .unwrap(); + id.to_vec() + } + fn put_blob(&mut self, data: &[u8]) -> Vec { let id = vec![0xab, 0xcd, self.next_id]; self.next_id += 1; @@ -2193,6 +2408,293 @@ mod tests { assert!(s.ended_at.is_some()); } + fn blob_id(n: u8) -> [u8; 32] { + let mut id = [0u8; 32]; + id[31] = n; + id + } + + fn message_json(role: &str, content: Vec) -> Vec { + serde_json::to_vec(&serde_json::json!({ "role": role, "content": content })).unwrap() + } + + fn content_part(kind: &str, model_name: Option<&str>) -> Value { + let mut part = serde_json::json!({ "type": kind }); + if let Some(model_name) = model_name { + part["providerOptions"] = serde_json::json!({ + "cursor": { "modelName": model_name } + }); + } + part + } + + /// An ACP store under `tmp/acp-sessions//`. `messages` are field-1 + /// refs in order; `inline` is field 4. Nothing here reads `~/.cursor`. + fn write_acp_store( + tmp: &Path, + session: &str, + messages: &[Vec], + inline: Option<&[u8]>, + last_used_model: Option<&str>, + ) -> CursorParser { + let chat_dir = tmp.join("acp-sessions").join(session); + std::fs::create_dir_all(&chat_dir).unwrap(); + let store = StoreBuilder::create(&chat_dir.join("store.db")); + let mut root = Vec::new(); + for (index, message) in messages.iter().enumerate() { + let id = blob_id(index as u8 + 1); + store.put_blob_exact(&id, message); + field_bytes(1, &id, &mut root); + } + if let Some(inline) = inline { + field_bytes(4, inline, &mut root); + } + let root_id = blob_id(0xff); + store.put_blob_exact(&root_id, &root); + let mut meta = serde_json::json!({ + "agentId": session, + "latestRootBlobId": hex_encode(&root_id), + "createdAt": 1_u64, + }); + if let Some(model) = last_used_model { + meta["lastUsedModel"] = Value::String(model.to_string()); + } + store.put_meta(&meta); + CursorParser::with_base_dir(tmp.join("chats")) + } + + #[test] + fn model_for_session_reads_model_name_from_field1_messages() { + let tmp = tempfile::tempdir().unwrap(); + let session = "0198c9aa-aaaa-bbbb-cccc-ddddeeeeffff"; + let assistant = message_json( + "assistant", + vec![content_part( + "reasoning", + Some("accounts/fireworks/models/kimi-k3"), + )], + ); + let parser = write_acp_store( + tmp.path(), + session, + &[ + message_json("system", vec![content_part("text", None)]), + message_json("user", vec![content_part("text", Some("not-a-model"))]), + assistant, + ], + None, + None, + ); + assert_eq!( + parser.model_for_session(session).as_deref(), + Some("accounts/fireworks/models/kimi-k3") + ); + } + + #[test] + fn model_for_session_uses_the_latest_assistant_after_a_switch() { + let tmp = tempfile::tempdir().unwrap(); + let session = "0198c9aa-bbbb-cccc-dddd-eeeeffff0000"; + let parser = write_acp_store( + tmp.path(), + session, + &[ + message_json( + "assistant", + vec![content_part("reasoning", Some("grok-4.7-high"))], + ), + message_json("user", vec![content_part("text", None)]), + message_json( + "assistant", + vec![content_part("reasoning", Some("grok-4.7-xhigh-fast"))], + ), + ], + None, + None, + ); + assert_eq!( + parser.model_for_session(session).as_deref(), + Some("grok-4.7-xhigh-fast") + ); + } + + #[test] + fn model_for_session_reads_an_inline_field4_message() { + let tmp = tempfile::tempdir().unwrap(); + let session = "0198c9aa-cccc-dddd-eeee-ffff00001111"; + let inline = message_json( + "assistant", + vec![ + content_part("reasoning", Some("grok-4.7-high")), + content_part("text", None), + content_part("tool-call", None), + ], + ); + let parser = write_acp_store( + tmp.path(), + session, + &[ + message_json("system", vec![content_part("text", None)]), + message_json("user", vec![content_part("text", None)]), + ], + Some(&inline), + None, + ); + assert_eq!( + parser.model_for_session(session).as_deref(), + Some("grok-4.7-high") + ); + } + + #[test] + fn model_for_session_keeps_an_earlier_model_when_later_messages_have_none() { + let tmp = tempfile::tempdir().unwrap(); + let session = "0198c9aa-dddd-eeee-ffff-000011112222"; + let inline = message_json( + "assistant", + vec![content_part("text", None), content_part("tool-call", None)], + ); + let parser = write_acp_store( + tmp.path(), + session, + &[message_json( + "assistant", + vec![content_part("redacted-reasoning", Some("composer-2.5-fast"))], + )], + Some(&inline), + None, + ); + assert_eq!( + parser.model_for_session(session).as_deref(), + Some("composer-2.5-fast") + ); + } + + #[test] + fn model_for_session_skips_entries_without_model_name_and_junk_blobs() { + let tmp = tempfile::tempdir().unwrap(); + let session = "0198c9aa-eeee-ffff-0000-111122223333"; + let parser = write_acp_store( + tmp.path(), + session, + &[ + b"{not json".to_vec(), + message_json( + "assistant", + vec![ + content_part("text", None), + content_part("tool-call", None), + content_part("reasoning", Some("cursor-grok-4.6-high")), + content_part("text", Some("")), + ], + ), + ], + None, + None, + ); + // A blob the root does not reference, even one that names a model, + // must not win. The fingerprint marker is not a model source. + let stale = tmp + .path() + .join("acp-sessions") + .join(session) + .join("store.db"); + let conn = Connection::open(&stale).unwrap(); + conn.execute( + "INSERT INTO blobs (id, data) VALUES (?1, ?2)", + rusqlite::params![ + hex_encode(&[0xab; 32]), + br#"{"systemPromptFingerprint":{"model":"grok-4.7"},"role":"assistant","content":[{"type":"reasoning","providerOptions":{"cursor":{"modelName":"grok-4.7"}}}]}"# + ], + ) + .unwrap(); + assert_eq!( + parser.model_for_session(session).as_deref(), + Some("cursor-grok-4.6-high") + ); + } + + #[test] + fn model_for_session_prefers_last_used_model_when_the_store_still_has_it() { + let tmp = tempfile::tempdir().unwrap(); + let session = "0198c9aa-ffff-0000-1111-222233334444"; + let parser = write_acp_store( + tmp.path(), + session, + &[message_json( + "assistant", + vec![content_part("reasoning", Some("grok-4.7-high"))], + )], + None, + Some("composer-2.5"), + ); + assert_eq!( + parser.model_for_session(session).as_deref(), + Some("composer-2.5") + ); + } + + #[test] + fn model_for_session_ignores_a_missing_blob_and_a_non_assistant_role() { + let tmp = tempfile::tempdir().unwrap(); + let session = "0198c9aa-0000-1111-2222-333344445555"; + let chat_dir = tmp.path().join("acp-sessions").join(session); + std::fs::create_dir_all(&chat_dir).unwrap(); + let store = StoreBuilder::create(&chat_dir.join("store.db")); + let assistant = message_json( + "assistant", + vec![content_part("reasoning", Some("grok-4.7-high-fast"))], + ); + let present = blob_id(2); + store.put_blob_exact(&present, &assistant); + let mut root = Vec::new(); + field_bytes(1, &blob_id(1), &mut root); + field_bytes(1, &present, &mut root); + let root_id = blob_id(0xff); + store.put_blob_exact(&root_id, &root); + store.put_meta(&serde_json::json!({ + "latestRootBlobId": hex_encode(&root_id), + })); + let parser = CursorParser::with_base_dir(tmp.path().join("chats")); + assert_eq!( + parser.model_for_session(session).as_deref(), + Some("grok-4.7-high-fast") + ); + } + + #[test] + fn store_mtime_uses_the_newer_of_db_and_wal() { + let tmp = tempfile::tempdir().unwrap(); + let session = "0198c9aa-1111-2222-3333-444455556666"; + let dir = tmp.path().join("acp-sessions").join(session); + std::fs::create_dir_all(&dir).unwrap(); + let db = dir.join("store.db"); + let wal = dir.join("store.db-wal"); + std::fs::write(&db, b"db").unwrap(); + let older = std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000); + let newer = std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_100); + std::fs::File::options() + .write(true) + .open(&db) + .unwrap() + .set_modified(older) + .unwrap(); + let parser = CursorParser::with_base_dir(tmp.path().join("chats")); + assert_eq!(parser.store_mtime_for_session(session), Some(older)); + + std::fs::write(&wal, b"wal").unwrap(); + std::fs::File::options() + .write(true) + .open(&wal) + .unwrap() + .set_modified(newer) + .unwrap(); + assert_eq!(parser.store_mtime_for_session(session), Some(newer)); + + std::fs::remove_file(&wal).unwrap(); + assert_eq!(parser.store_mtime_for_session(session), Some(older)); + } + #[test] fn decodes_turns_blocks_and_tools() { let tmp = tempfile::tempdir().unwrap();