diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index 67401985cd..490122d053 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -2222,7 +2222,7 @@ mod tests { Ok(()) } - async fn start_test_embedded_app_server( + pub(crate) async fn start_test_embedded_app_server( config: Config, ) -> color_eyre::Result { let state_db = diff --git a/codex-rs/tui/src/session_archive_commands.rs b/codex-rs/tui/src/session_archive_commands.rs index 1dea0ffe0f..ddf82f7461 100644 --- a/codex-rs/tui/src/session_archive_commands.rs +++ b/codex-rs/tui/src/session_archive_commands.rs @@ -5,6 +5,8 @@ use std::io::IsTerminal; use std::io::Write; +use std::path::Path; +use std::path::PathBuf; use std::sync::Arc; use crate::Cli; @@ -82,16 +84,25 @@ pub async fn run_session_archive_command( target: String, options: SessionArchiveCommandOptions, ) -> Result { - let mut app_server = start_app_server_for_archive_command(options).await?; - run_session_archive_action_with_app_server(&mut app_server, action, &target).await + let codex_home = find_codex_home().wrap_err("failed to find Codex home")?; + let mut app_server = + start_app_server_for_archive_command(options, codex_home.to_path_buf()).await?; + run_session_archive_action_with_app_server( + &mut app_server, + codex_home.as_path(), + action, + &target, + ) + .await } async fn run_session_archive_action_with_app_server( app_server: &mut AppServerSession, + codex_home: &Path, action: SessionArchiveAction, target: &str, ) -> Result { - let resolved = resolve_session_target(app_server, action, target).await?; + let resolved = resolve_session_target(app_server, codex_home, action, target).await?; let session_name = match action { SessionArchiveAction::Archive => { app_server.thread_archive(resolved.session_id).await?; @@ -120,6 +131,7 @@ async fn run_session_archive_action_with_app_server( async fn resolve_session_target( app_server: &mut AppServerSession, + codex_home: &Path, action: SessionArchiveAction, target: &str, ) -> Result { @@ -151,7 +163,9 @@ async fn resolve_session_target( SessionArchiveAction::Unarchive => ("archived", &[true]), }; for &archived in archived_values { - if let Some(thread) = lookup_session_by_exact_name(app_server, target, archived).await? { + if let Some(thread) = + lookup_session_by_exact_name(app_server, codex_home, target, archived).await? + { return session_target_from_app_server_thread(thread); } } @@ -162,50 +176,103 @@ async fn resolve_session_target( async fn lookup_session_by_exact_name( app_server: &mut AppServerSession, + codex_home: &Path, name: &str, archived: bool, ) -> Result> { - // Search is the fast path, but some stores attach renamed titles after applying the filter. - for search_term in [Some(name), None] { - let mut cursor = None; - loop { - let response = app_server - .thread_list(ThreadListParams { - cursor: cursor.clone(), - limit: Some(100), - sort_key: Some(ThreadSortKey::UpdatedAt), - sort_direction: None, - model_providers: None, - source_kinds: Some(super::resume_source_kinds( - /*include_non_interactive*/ false, - )), - archived: Some(archived), - section_id: None, - parent_thread_id: None, - ancestor_thread_id: None, - cwd: None, - use_state_db_only: false, - search_term: search_term.map(str::to_string), - }) - .await - .wrap_err("failed to list sessions while resolving session name")?; + let uses_remote_workspace = app_server.uses_remote_workspace(); + // Remote workspaces stay on their existing server-side path. Local workspaces trust SQLite + // names, then scan and repair only after a miss or an unusable rollout path. + let lookup_modes = if uses_remote_workspace { + &[SessionNameLookupMode::ScanAndRepair][..] + } else { + &[ + SessionNameLookupMode::StateDbOnly, + SessionNameLookupMode::ScanAndRepair, + ][..] + }; + for &lookup_mode in lookup_modes { + // Only the embedded server can safely use SQLite's recency cursor. Daemons may predate + // that sort key, and filesystem repair must paginate in the scanner's mtime order. + let sort_key = if lookup_mode == SessionNameLookupMode::StateDbOnly + && app_server.uses_embedded_app_server() + { + ThreadSortKey::RecencyAt + } else { + ThreadSortKey::UpdatedAt + }; + // Search is the fast path, but legacy stores attach renamed titles after filtering. + for search_term in [Some(name), None] { + let mut cursor = None; + loop { + let response = app_server + .thread_list(ThreadListParams { + cursor: cursor.clone(), + limit: Some(100), + sort_key: Some(sort_key), + sort_direction: None, + model_providers: None, + source_kinds: Some(super::resume_source_kinds( + /*include_non_interactive*/ false, + )), + archived: Some(archived), + section_id: None, + parent_thread_id: None, + ancestor_thread_id: None, + cwd: None, + use_state_db_only: lookup_mode == SessionNameLookupMode::StateDbOnly, + search_term: search_term.map(str::to_string), + }) + .await + .wrap_err("failed to list sessions while resolving session name")?; - if let Some(thread) = response - .data - .into_iter() - .find(|thread| thread.name.as_deref() == Some(name)) - { - return Ok(Some(thread)); + for thread in response + .data + .into_iter() + .filter(|thread| thread.name.as_deref() == Some(name)) + { + if !uses_remote_workspace { + // The action still requires a real rollout in the requested collection. + let thread_id = ThreadId::from_string(&thread.id).wrap_err_with(|| { + format!("app server returned invalid session id `{}`", thread.id) + })?; + let expected_root = codex_home.join(if archived { + codex_rollout::ARCHIVED_SESSIONS_SUBDIR + } else { + codex_rollout::SESSIONS_SUBDIR + }); + let valid_rollout = if let Some(path) = thread.path.as_deref() + && let Some(path) = codex_rollout::existing_rollout_path(path).await + && path.starts_with(expected_root) + && let Ok(session_meta) = + codex_rollout::read_session_meta_line(path.as_path()).await + { + session_meta.meta.id == thread_id + } else { + false + }; + if !valid_rollout { + continue; + } + } + return Ok(Some(thread)); + } + let Some(next_cursor) = response.next_cursor else { + break; + }; + cursor = Some(next_cursor); } - let Some(next_cursor) = response.next_cursor else { - break; - }; - cursor = Some(next_cursor); } } Ok(None) } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum SessionNameLookupMode { + StateDbOnly, + ScanAndRepair, +} + fn session_target_from_app_server_thread(thread: AppServerThread) -> Result { let session_id = ThreadId::from_string(&thread.id) .wrap_err_with(|| format!("app server returned invalid session id `{}`", thread.id))?; @@ -246,6 +313,7 @@ fn confirm_session_delete(target: &ResolvedSessionTarget) -> Result { async fn start_app_server_for_archive_command( options: SessionArchiveCommandOptions, + codex_home: PathBuf, ) -> Result { let SessionArchiveCommandOptions { cli, @@ -259,8 +327,6 @@ async fn start_app_server_for_archive_command( let cli_kv_overrides = overrides_cli .parse_overrides() .map_err(|err| eyre!("failed to parse -c overrides: {err}"))?; - let codex_home = find_codex_home().wrap_err("failed to find Codex home")?; - let mut launch_loader_overrides = loader_overrides.clone(); if let Some(profile_v2) = cli.config_profile_v2.as_ref() { launch_loader_overrides.user_config_path = Some(resolve_profile_v2_config_path( @@ -413,3 +479,7 @@ async fn start_app_server_for_archive_command( .with_remote_cwd_override(remote_cwd_override), ) } + +#[cfg(test)] +#[path = "session_archive_commands_tests.rs"] +mod tests; diff --git a/codex-rs/tui/src/session_archive_commands_tests.rs b/codex-rs/tui/src/session_archive_commands_tests.rs new file mode 100644 index 0000000000..351247a651 --- /dev/null +++ b/codex-rs/tui/src/session_archive_commands_tests.rs @@ -0,0 +1,373 @@ +use std::path::PathBuf; +use std::sync::Arc; + +use chrono::DateTime; +use chrono::Utc; +use codex_protocol::ThreadId; +use codex_protocol::protocol::SessionMeta; +use codex_protocol::protocol::SessionMetaLine; +use codex_protocol::protocol::SessionSource; +use codex_state::ThreadMetadataBuilder; +use codex_utils_absolute_path::test_support::PathExt; +use pretty_assertions::assert_eq; +use tempfile::TempDir; + +use super::DeleteConfirmation; +use super::SessionArchiveAction; +use super::run_session_archive_action_with_app_server; +use crate::app_server_session::AppServerSession; +use crate::app_server_session::ThreadParamsMode; +use crate::legacy_core::config::Config; +use crate::legacy_core::config::ConfigBuilder; +use crate::tests::start_test_embedded_app_server; + +async fn build_config(temp_dir: &TempDir) -> std::io::Result { + ConfigBuilder::default() + .codex_home(temp_dir.path().to_path_buf()) + .build() + .await +} + +async fn state_runtime(config: &Config) -> std::io::Result> { + let runtime = codex_state::StateRuntime::init( + codex_state::SqliteConfig::new_for_testing(config.codex_home.as_path().abs()), + config.model_provider_id.clone(), + ) + .await + .map_err(std::io::Error::other)?; + runtime + .mark_backfill_complete(/*last_watermark*/ None) + .await + .map_err(std::io::Error::other)?; + Ok(runtime) +} + +async fn start_app_server(config: Config) -> color_eyre::Result { + Ok(AppServerSession::new( + codex_app_server_client::AppServerClient::InProcess( + start_test_embedded_app_server(config).await?, + ), + ThreadParamsMode::Embedded, + )) +} + +fn write_rollout( + config: &Config, + thread_id: ThreadId, + archived: bool, + timestamp: &str, + preview: &str, +) -> color_eyre::Result { + let subdir = if archived { + "archived_sessions" + } else { + "sessions/2025/02/01" + }; + let rollout_path = config + .codex_home + .join(subdir) + .join(format!("rollout-2025-02-01T10-00-00-{thread_id}.jsonl")); + std::fs::create_dir_all(rollout_path.parent().expect("rollout parent"))?; + let session_meta = SessionMetaLine { + meta: SessionMeta { + session_id: thread_id.into(), + id: thread_id, + timestamp: timestamp.to_string(), + cwd: config.codex_home.join("project").to_path_buf(), + originator: "codex".to_string(), + cli_version: "0.0.0".to_string(), + source: SessionSource::Cli, + model_provider: Some(config.model_provider_id.clone()), + ..Default::default() + }, + git: None, + }; + let lines = [ + serde_json::json!({ + "timestamp": timestamp, + "type": "session_meta", + "payload": session_meta, + }), + serde_json::json!({ + "timestamp": timestamp, + "type": "response_item", + "payload": { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": preview}], + }, + }), + serde_json::json!({ + "timestamp": timestamp, + "type": "event_msg", + "payload": { + "type": "user_message", + "message": preview, + "kind": "plain", + }, + }), + ]; + std::fs::write( + &rollout_path, + lines + .iter() + .map(serde_json::Value::to_string) + .collect::>() + .join("\n") + + "\n", + )?; + Ok(rollout_path.to_path_buf()) +} + +fn thread_metadata( + config: &Config, + thread_id: ThreadId, + rollout_path: PathBuf, + name: &str, + archived: bool, +) -> codex_state::ThreadMetadata { + let created_at = DateTime::parse_from_rfc3339("2025-02-01T10:00:00Z") + .expect("timestamp should parse") + .with_timezone(&Utc); + let mut builder = ThreadMetadataBuilder::new( + thread_id, + rollout_path, + created_at, + serde_json::from_value(serde_json::json!("cli")) + .expect("cli session source should deserialize"), + ); + builder.cwd = config.codex_home.join("project").to_path_buf(); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.title = name.to_string(); + metadata.first_user_message = Some("preview text".to_string()); + metadata.preview = metadata.first_user_message.clone(); + metadata.archived_at = archived.then_some(created_at); + metadata +} + +#[tokio::test] +async fn archives_by_sqlite_name() -> color_eyre::Result<()> { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let runtime = state_runtime(&config).await?; + let thread_id = ThreadId::new(); + let rollout_path = write_rollout( + &config, + thread_id, + /*archived*/ false, + "2025-02-01T10:00:00Z", + "preview", + )?; + runtime + .upsert_thread(&thread_metadata( + &config, + thread_id, + rollout_path, + "saved-session", + /*archived*/ false, + )) + .await + .map_err(std::io::Error::other)?; + + let mut app_server = start_app_server(config.clone()).await?; + let message = run_session_archive_action_with_app_server( + &mut app_server, + config.codex_home.as_path(), + SessionArchiveAction::Archive, + "saved-session", + ) + .await?; + app_server.shutdown().await?; + + assert_eq!( + ( + message, + runtime + .get_thread(thread_id) + .await + .map_err(std::io::Error::other)? + .is_some_and(|metadata| metadata.archived_at.is_some()), + ), + ( + format!("Archived session saved-session ({thread_id})."), + true, + ), + ); + Ok(()) +} + +#[tokio::test] +async fn unarchives_by_sqlite_name() -> color_eyre::Result<()> { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let runtime = state_runtime(&config).await?; + let thread_id = ThreadId::new(); + let rollout_path = write_rollout( + &config, + thread_id, + /*archived*/ true, + "2025-02-01T10:00:00Z", + "preview", + )?; + runtime + .upsert_thread(&thread_metadata( + &config, + thread_id, + rollout_path, + "saved-session", + /*archived*/ true, + )) + .await + .map_err(std::io::Error::other)?; + + let mut app_server = start_app_server(config.clone()).await?; + let message = run_session_archive_action_with_app_server( + &mut app_server, + config.codex_home.as_path(), + SessionArchiveAction::Unarchive, + "saved-session", + ) + .await?; + app_server.shutdown().await?; + + assert_eq!( + ( + message, + runtime + .get_thread(thread_id) + .await + .map_err(std::io::Error::other)? + .is_some_and(|metadata| metadata.archived_at.is_some()), + ), + ( + format!("Unarchived session saved-session ({thread_id})."), + false, + ), + ); + Ok(()) +} + +#[tokio::test] +async fn deletes_valid_duplicate_after_stale_sqlite_hit() -> color_eyre::Result<()> { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let runtime = state_runtime(&config).await?; + let stale_id = ThreadId::new(); + let stale_rollout_path = write_rollout( + &config, + stale_id, + /*archived*/ true, + "2025-02-01T10:00:00Z", + "stale preview", + )?; + runtime + .upsert_thread(&thread_metadata( + &config, + stale_id, + stale_rollout_path.clone(), + "saved-session", + /*archived*/ false, + )) + .await + .map_err(std::io::Error::other)?; + + let thread_id = ThreadId::new(); + let rollout_path = write_rollout( + &config, + thread_id, + /*archived*/ false, + "2025-02-01T10:00:00Z", + "preview", + )?; + codex_rollout::append_thread_name(config.codex_home.as_path(), thread_id, "saved-session") + .await?; + + let mut app_server = start_app_server(config.clone()).await?; + let message = run_session_archive_action_with_app_server( + &mut app_server, + config.codex_home.as_path(), + SessionArchiveAction::Delete(DeleteConfirmation::Skip), + "saved-session", + ) + .await?; + app_server.shutdown().await?; + + assert_eq!( + ( + message, + rollout_path.exists(), + stale_rollout_path.exists(), + runtime + .get_thread(thread_id) + .await + .map_err(std::io::Error::other)?, + runtime + .get_thread(stale_id) + .await + .map_err(std::io::Error::other)? + .is_some(), + ), + ( + format!("Deleted session saved-session ({thread_id})."), + false, + true, + None, + true, + ), + ); + Ok(()) +} + +#[tokio::test] +async fn trusts_sqlite_name_over_legacy_index_for_delete() -> color_eyre::Result<()> { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let runtime = state_runtime(&config).await?; + let thread_id = ThreadId::new(); + let rollout_path = write_rollout( + &config, + thread_id, + /*archived*/ false, + "2025-02-01T10:00:00Z", + "preview", + )?; + runtime + .upsert_thread(&thread_metadata( + &config, + thread_id, + rollout_path.clone(), + "new-session", + /*archived*/ false, + )) + .await + .map_err(std::io::Error::other)?; + codex_rollout::append_thread_name(config.codex_home.as_path(), thread_id, "old-session") + .await?; + + let mut app_server = start_app_server(config.clone()).await?; + let message = run_session_archive_action_with_app_server( + &mut app_server, + config.codex_home.as_path(), + SessionArchiveAction::Delete(DeleteConfirmation::Skip), + "new-session", + ) + .await?; + app_server.shutdown().await?; + + assert_eq!( + ( + message, + rollout_path.exists(), + runtime + .get_thread(thread_id) + .await + .map_err(std::io::Error::other)?, + ), + ( + format!("Deleted session new-session ({thread_id})."), + false, + None, + ), + ); + Ok(()) +}