use anyhow::Result;use app_test_support::TestAppServer;use app_test_support::create_fake_parented_rollout_with_source;use app_test_support::create_fake_rollout;use app_test_support::create_fake_rollout_with_source;use app_test_support::create_final_assistant_message_sse_response;use app_test_support::create_mock_responses_server_sequence;use app_test_support::rollout_path;use app_test_support::test_absolute_path;use app_test_support::to_response;use chrono::DateTime;use chrono::Utc;use codex_app_server_protocol::GitInfo as ApiGitInfo;use codex_app_server_protocol::JSONRPCError;use codex_app_server_protocol::JSONRPCResponse;use codex_app_server_protocol::RequestId;use codex_app_server_protocol::SessionSource;use codex_app_server_protocol::SortDirection;use codex_app_server_protocol::ThreadListCwdFilter;use codex_app_server_protocol::ThreadListResponse;use codex_app_server_protocol::ThreadSearchResponse;use codex_app_server_protocol::ThreadSortKey;use codex_app_server_protocol::ThreadSourceKind;use codex_app_server_protocol::ThreadStartParams;use codex_app_server_protocol::ThreadStartResponse;use codex_app_server_protocol::ThreadStatus;use codex_app_server_protocol::TurnStartParams;use codex_app_server_protocol::TurnStartResponse;use codex_app_server_protocol::UserInput;use codex_core::ARCHIVED_SESSIONS_SUBDIR;use codex_git_utils::GitSha;use codex_protocol::ThreadId;use codex_protocol::protocol::GitInfo as CoreGitInfo;use codex_protocol::protocol::RolloutItem;use codex_protocol::protocol::RolloutLine;use codex_protocol::protocol::SessionSource as CoreSessionSource;use codex_protocol::protocol::SubAgentSource;use codex_state::DirectionalThreadSpawnEdgeStatus;use core_test_support::responses;use pretty_assertions::assert_eq;use std::cmp::Reverse;use std::fs;use std::fs::FileTimes;use std::fs::OpenOptions;use std::path::Path;use tempfile::TempDir;use tokio::time::timeout;use uuid::Uuid;const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);async fn init_mcp(codex_home: &Path) -> Result<TestAppServer> { let mut mcp = TestAppServer::new(codex_home).await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; Ok(mcp)}async fn list_threads( mcp: &mut TestAppServer, cursor: Option<String>, limit: Option<u32>, providers: Option<Vec<String>>, source_kinds: Option<Vec<ThreadSourceKind>>, archived: Option<bool>,) -> Result<ThreadListResponse> { list_threads_with_sort( mcp, cursor, limit, providers, source_kinds, /*sort_key*/ None, archived, ) .await}async fn list_threads_with_sort( mcp: &mut TestAppServer, cursor: Option<String>, limit: Option<u32>, providers: Option<Vec<String>>, source_kinds: Option<Vec<ThreadSourceKind>>, sort_key: Option<ThreadSortKey>, archived: Option<bool>,) -> Result<ThreadListResponse> { let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { cursor, limit, sort_key, sort_direction: None, model_providers: providers, source_kinds, archived, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: None, }) .await?; let resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; to_response::<ThreadListResponse>(resp)}async fn list_threads_for_parent( mcp: &mut TestAppServer, parent_thread_id: ThreadId, cursor: Option<String>, limit: u32, model_providers: Option<Vec<String>>, source_kinds: Option<Vec<ThreadSourceKind>>,) -> Result<ThreadListResponse> { let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { cursor, limit: Some(limit), sort_key: None, sort_direction: None, model_providers, source_kinds, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: Some(parent_thread_id.to_string()), }) .await?; let response = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; to_response::<ThreadListResponse>(response)}fn create_fake_rollouts<F, G>( codex_home: &Path, count: usize, provider_for_index: F, timestamp_for_index: G, preview: &str,) -> Result<Vec<String>>where F: Fn(usize) -> &'static str, G: Fn(usize) -> (String, String),{ let mut ids = Vec::with_capacity(count); for i in 0..count { let (ts_file, ts_rfc) = timestamp_for_index(i); ids.push(create_fake_rollout( codex_home, &ts_file, &ts_rfc, preview, Some(provider_for_index(i)), /*git_info*/ None, )?); } Ok(ids)}fn timestamp_at( year: i32, month: u32, day: u32, hour: u32, minute: u32, second: u32,) -> (String, String) { ( format!("{year:04}-{month:02}-{day:02}T{hour:02}-{minute:02}-{second:02}"), format!("{year:04}-{month:02}-{day:02}T{hour:02}:{minute:02}:{second:02}Z"), )}#[allow(dead_code)]fn set_rollout_mtime(path: &Path, updated_at_rfc3339: &str) -> Result<()> { let parsed = DateTime::parse_from_rfc3339(updated_at_rfc3339)?.with_timezone(&Utc); let times = FileTimes::new().set_modified(parsed.into()); OpenOptions::new() .append(true) .open(path)? .set_times(times)?; Ok(())}fn set_rollout_cwd(path: &Path, cwd: &Path) -> Result<()> { let content = fs::read_to_string(path)?; let mut lines: Vec<String> = content.lines().map(str::to_string).collect(); let first_line = lines .first_mut() .ok_or_else(|| anyhow::anyhow!("rollout at {} is empty", path.display()))?; let mut rollout_line: RolloutLine = serde_json::from_str(first_line)?; let RolloutItem::SessionMeta(mut session_meta_line) = rollout_line.item else { return Err(anyhow::anyhow!( "rollout at {} does not start with session metadata", path.display() )); }; session_meta_line.meta.cwd = cwd.to_path_buf(); rollout_line.item = RolloutItem::SessionMeta(session_meta_line); *first_line = serde_json::to_string(&rollout_line)?; fs::write(path, lines.join("\n") + "\n")?; Ok(())}#[tokio::test]async fn thread_list_basic_empty() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data, next_cursor, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, /*archived*/ None, ) .await?; assert!(data.is_empty()); assert_eq!(next_cursor, None); Ok(())}#[tokio::test]async fn thread_list_reports_system_error_idle_flag_after_failed_turn() -> Result<()> { let responses = vec![ create_final_assistant_message_sse_response("seeded")?, responses::sse_failed("resp-2", "server_error", "simulated failure"), ]; let server = create_mock_responses_server_sequence(responses).await; let codex_home = TempDir::new()?; create_runtime_config(codex_home.path(), &server.uri())?; let mut mcp = init_mcp(codex_home.path()).await?; let start_id = mcp .send_thread_start_request(ThreadStartParams { model: Some("mock-model".to_string()), ..Default::default() }) .await?; let start_resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(start_id)), ) .await??; let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?; let seed_turn_id = mcp .send_turn_start_request(TurnStartParams { thread_id: thread.id.clone(), client_user_message_id: None, input: vec![UserInput::Text { text: "seed history".to_string(), text_elements: Vec::new(), }], ..Default::default() }) .await?; let seed_turn_resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(seed_turn_id)), ) .await??; let _: TurnStartResponse = to_response::<TurnStartResponse>(seed_turn_resp)?; timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_notification_message("turn/completed"), ) .await??; let failed_turn_id = mcp .send_turn_start_request(TurnStartParams { thread_id: thread.id.clone(), client_user_message_id: None, input: vec![UserInput::Text { text: "fail turn".to_string(), text_elements: Vec::new(), }], ..Default::default() }) .await?; let failed_turn_resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(failed_turn_id)), ) .await??; let _: TurnStartResponse = to_response::<TurnStartResponse>(failed_turn_resp)?; timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_notification_message("error"), ) .await??; let ThreadListResponse { data, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), Some(vec![ ThreadSourceKind::AppServer, ThreadSourceKind::Cli, ThreadSourceKind::VsCode, ]), /*archived*/ None, ) .await?; let listed = data .iter() .find(|candidate| candidate.id == thread.id) .expect("expected started thread to be listed"); assert_eq!(listed.status, ThreadStatus::SystemError,); Ok(())}// Minimal config.toml for listing.fn create_minimal_config(codex_home: &std::path::Path) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); std::fs::write( config_toml, r#"model = "mock-model"approval_policy = "never""#, )}fn create_runtime_config(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> { let config_toml = codex_home.join("config.toml"); std::fs::write( config_toml, format!( r#"model = "mock-model"approval_policy = "never"sandbox_mode = "read-only"model_provider = "mock_provider"[model_providers.mock_provider]name = "Mock provider for test"base_url = "{server_uri}/v1"wire_api = "responses"request_max_retries = 0stream_max_retries = 0"# ), )}#[tokio::test]async fn thread_list_pagination_next_cursor_none_on_last_page() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; // Create three rollouts so we can paginate with limit=2. let _a = create_fake_rollout( codex_home.path(), "2025-01-02T12-00-00", "2025-01-02T12:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let _b = create_fake_rollout( codex_home.path(), "2025-01-01T13-00-00", "2025-01-01T13:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let _c = create_fake_rollout( codex_home.path(), "2025-01-01T12-00-00", "2025-01-01T12:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let mut mcp = init_mcp(codex_home.path()).await?; // Page 1: limit 2 → expect next_cursor Some. let ThreadListResponse { data: data1, next_cursor: cursor1, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(2), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, /*archived*/ None, ) .await?; assert_eq!(data1.len(), 2); for thread in &data1 { assert_eq!(thread.preview, "Hello"); assert_eq!(thread.model_provider, "mock_provider"); assert!(thread.created_at > 0); assert_eq!(thread.updated_at, thread.created_at); assert_eq!(thread.cwd, test_absolute_path("/")); assert_eq!(thread.cli_version, "0.0.0"); assert_eq!(thread.source, SessionSource::Cli); assert_eq!(thread.git_info, None); assert_eq!(thread.status, ThreadStatus::NotLoaded); } let cursor1 = cursor1.expect("expected nextCursor on first page"); // Page 2: with cursor → expect next_cursor None when no more results. let ThreadListResponse { data: data2, next_cursor: cursor2, .. } = list_threads( &mut mcp, Some(cursor1), Some(2), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, /*archived*/ None, ) .await?; assert!(data2.len() <= 2); for thread in &data2 { assert_eq!(thread.preview, "Hello"); assert_eq!(thread.model_provider, "mock_provider"); assert!(thread.created_at > 0); assert_eq!(thread.updated_at, thread.created_at); assert_eq!(thread.cwd, test_absolute_path("/")); assert_eq!(thread.cli_version, "0.0.0"); assert_eq!(thread.source, SessionSource::Cli); assert_eq!(thread.git_info, None); assert_eq!(thread.status, ThreadStatus::NotLoaded); } assert_eq!(cursor2, None, "expected nextCursor to be null on last page"); Ok(())}#[tokio::test]async fn thread_list_respects_provider_filter() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; // Create rollouts under two providers. let _a = create_fake_rollout( codex_home.path(), "2025-01-02T10-00-00", "2025-01-02T10:00:00Z", "X", Some("mock_provider"), /*git_info*/ None, )?; // mock_provider let _b = create_fake_rollout( codex_home.path(), "2025-01-02T11-00-00", "2025-01-02T11:00:00Z", "X", Some("other_provider"), /*git_info*/ None, )?; let mut mcp = init_mcp(codex_home.path()).await?; // Filter to only other_provider; expect 1 item, nextCursor None. let ThreadListResponse { data, next_cursor, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["other_provider".to_string()]), /*source_kinds*/ None, /*archived*/ None, ) .await?; assert_eq!(data.len(), 1); assert_eq!(next_cursor, None); let thread = &data[0]; assert_eq!(thread.preview, "X"); assert_eq!(thread.model_provider, "other_provider"); let expected_ts = chrono::DateTime::parse_from_rfc3339("2025-01-02T11:00:00Z")?.timestamp(); assert_eq!(thread.created_at, expected_ts); assert_eq!(thread.updated_at, expected_ts); assert_eq!(thread.cwd, test_absolute_path("/")); assert_eq!(thread.cli_version, "0.0.0"); assert_eq!(thread.source, SessionSource::Cli); assert_eq!(thread.git_info, None); Ok(())}#[tokio::test]async fn thread_list_respects_cwd_filters() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let first_filtered_id = create_fake_rollout( codex_home.path(), "2025-01-02T10-00-00", "2025-01-02T10:00:00Z", "first filtered", Some("mock_provider"), /*git_info*/ None, )?; let second_filtered_id = create_fake_rollout( codex_home.path(), "2025-01-02T12-00-00", "2025-01-02T12:00:00Z", "second filtered", Some("mock_provider"), /*git_info*/ None, )?; let unfiltered_id = create_fake_rollout( codex_home.path(), "2025-01-02T11-00-00", "2025-01-02T11:00:00Z", "unfiltered", Some("mock_provider"), /*git_info*/ None, )?; let first_target_cwd = codex_home.path().join("first-target-cwd"); let second_target_cwd = codex_home.path().join("second-target-cwd"); fs::create_dir_all(&first_target_cwd)?; fs::create_dir_all(&second_target_cwd)?; set_rollout_cwd( rollout_path(codex_home.path(), "2025-01-02T10-00-00", &first_filtered_id).as_path(), &first_target_cwd, )?; set_rollout_cwd( rollout_path( codex_home.path(), "2025-01-02T12-00-00", &second_filtered_id, ) .as_path(), &second_target_cwd, )?; let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { cursor: None, limit: Some(10), sort_key: None, sort_direction: None, model_providers: Some(vec!["mock_provider".to_string()]), source_kinds: None, archived: None, cwd: Some(ThreadListCwdFilter::Many(vec![ first_target_cwd.to_string_lossy().into_owned(), second_target_cwd.to_string_lossy().into_owned(), ])), use_state_db_only: false, search_term: None, parent_thread_id: None, }) .await?; let resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let ThreadListResponse { data, next_cursor, .. } = to_response::<ThreadListResponse>(resp)?; assert_eq!(next_cursor, None); let filtered_ids: Vec<_> = data.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!( filtered_ids, vec![second_filtered_id.as_str(), first_filtered_id.as_str()] ); assert!(!filtered_ids.contains(&unfiltered_id.as_str())); assert_eq!(data[0].cwd.as_path(), second_target_cwd.as_path()); assert_eq!(data[1].cwd.as_path(), first_target_cwd.as_path()); Ok(())}#[tokio::test]async fn thread_list_respects_search_term_filter() -> Result<()> { let codex_home = TempDir::new()?; std::fs::write( codex_home.path().join("config.toml"), r#"model = "mock-model"approval_policy = "never"suppress_unstable_features_warning = true[features]sqlite = true"#, )?; let older_match = create_fake_rollout( codex_home.path(), "2025-01-02T10-00-00", "2025-01-02T10:00:00Z", "match: needle", Some("mock_provider"), /*git_info*/ None, )?; let _non_match = create_fake_rollout( codex_home.path(), "2025-01-02T11-00-00", "2025-01-02T11:00:00Z", "no hit here", Some("mock_provider"), /*git_info*/ None, )?; let newer_match = create_fake_rollout( codex_home.path(), "2025-01-02T12-00-00", "2025-01-02T12:00:00Z", "needle suffix", Some("mock_provider"), /*git_info*/ None, )?; // `thread/list` applies `search_term` on the sqlite fast path. This test creates // rollouts manually, so mark the DB backfill complete and then run an unsearched // list large enough to repair every rollout the searched list should find. let state_db = codex_state::StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".into()) .await?; state_db .mark_backfill_complete(/*last_watermark*/ None) .await?; let rollout_config = codex_rollout::RolloutConfig { codex_home: codex_home.path().to_path_buf(), sqlite_home: codex_home.path().to_path_buf(), cwd: codex_home.path().to_path_buf(), model_provider_id: "mock_provider".to_string(), generate_memories: false, }; let repaired_page = codex_core::RolloutRecorder::list_threads( Some(state_db.clone()), &rollout_config, /*page_size*/ 10, /*cursor*/ None, codex_core::ThreadSortKey::CreatedAt, codex_core::SortDirection::Desc, &[], /*model_providers*/ None, /*cwd_filters*/ None, "mock_provider", /*search_term*/ None, ) .await?; assert_eq!(repaired_page.items.len(), 3); let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { cursor: None, limit: Some(10), sort_key: None, sort_direction: None, model_providers: Some(vec!["mock_provider".to_string()]), source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: Some("needle".to_string()), parent_thread_id: None, }) .await?; let resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let ThreadListResponse { data, next_cursor, .. } = to_response::<ThreadListResponse>(resp)?; assert_eq!(next_cursor, None); let ids: Vec<_> = data.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!(ids, vec![newer_match, older_match]); Ok(())}#[tokio::test]async fn thread_search_returns_content_matches() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let older_match = create_fake_rollout( codex_home.path(), "2025-01-02T10-00-00", "2025-01-02T10:00:00Z", "match: needle", Some("mock_provider"), /*git_info*/ None, )?; let _non_match = create_fake_rollout( codex_home.path(), "2025-01-02T11-00-00", "2025-01-02T11:00:00Z", "no hit here", Some("mock_provider"), /*git_info*/ None, )?; let newer_match = create_fake_rollout( codex_home.path(), "2025-01-02T12-00-00", "2025-01-02T12:00:00Z", "mixed NEEDLE suffix", Some("mock_provider"), /*git_info*/ None, )?; let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_search_request(codex_app_server_protocol::ThreadSearchParams { cursor: None, limit: Some(10), sort_key: None, sort_direction: None, source_kinds: None, archived: None, search_term: "needle".to_string(), }) .await?; let resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let ThreadSearchResponse { data, next_cursor, .. } = to_response::<ThreadSearchResponse>(resp)?; assert_eq!(next_cursor, None); let ids: Vec<_> = data .iter() .map(|result| result.thread.id.as_str()) .collect(); assert_eq!(ids, vec![newer_match, older_match]); assert_eq!(data[0].snippet, "mixed NEEDLE suffix"); Ok(())}#[tokio::test]async fn thread_search_matches_json_escaped_content() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let search_term = r#"quoted "needle" \ path"#; let thread_id = create_fake_rollout( codex_home.path(), "2025-01-02T10-00-00", "2025-01-02T10:00:00Z", search_term, Some("mock_provider"), /*git_info*/ None, )?; let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_search_request(codex_app_server_protocol::ThreadSearchParams { cursor: None, limit: Some(10), sort_key: None, sort_direction: None, source_kinds: None, archived: None, search_term: search_term.to_string(), }) .await?; let resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let ThreadSearchResponse { data, .. } = to_response::<ThreadSearchResponse>(resp)?; assert_eq!(data.len(), 1); assert_eq!(data[0].thread.id, thread_id); assert_eq!(data[0].snippet, search_term); Ok(())}#[tokio::test]async fn thread_search_filters_by_source_kind() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let cli_id = create_fake_rollout( codex_home.path(), "2025-02-01T10-00-00", "2025-02-01T10:00:00Z", "shared needle", Some("mock_provider"), /*git_info*/ None, )?; let exec_id = create_fake_rollout_with_source( codex_home.path(), "2025-02-01T11-00-00", "2025-02-01T11:00:00Z", "shared needle", Some("mock_provider"), /*git_info*/ None, CoreSessionSource::Exec, )?; let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_search_request(codex_app_server_protocol::ThreadSearchParams { cursor: None, limit: Some(10), sort_key: None, sort_direction: None, source_kinds: Some(vec![ThreadSourceKind::Exec]), archived: None, search_term: "needle".to_string(), }) .await?; let resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let ThreadSearchResponse { data, .. } = to_response::<ThreadSearchResponse>(resp)?; let ids: Vec<_> = data .iter() .map(|result| result.thread.id.as_str()) .collect(); assert_eq!(ids, vec![exec_id.as_str()]); assert_ne!(cli_id, exec_id); Ok(())}#[tokio::test]async fn thread_list_state_db_only_returns_sqlite_without_jsonl_repair() -> Result<()> { let codex_home = TempDir::new()?; std::fs::write( codex_home.path().join("config.toml"), r#"model = "mock-model"approval_policy = "never"suppress_unstable_features_warning = true[features]sqlite = true"#, )?; let thread_id = create_fake_rollout( codex_home.path(), "2025-01-02T10-00-00", "2025-01-02T10:00:00Z", "state db only should not see this before repair", Some("mock_provider"), /*git_info*/ None, )?; let state_db = codex_state::StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".into()) .await?; state_db .mark_backfill_complete(/*last_watermark*/ None) .await?; let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { cursor: None, limit: Some(10), sort_key: None, sort_direction: None, model_providers: Some(vec!["mock_provider".to_string()]), source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: None, }) .await?; let resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let repaired_response = to_response::<ThreadListResponse>(resp)?; let ids: Vec<_> = repaired_response .data .iter() .map(|thread| thread.id.as_str()) .collect(); assert_eq!(ids, vec![thread_id.as_str()]); let thread_uuid = ThreadId::from_string(&thread_id)?; let stale_cwd = codex_home.path().join("stale-cwd"); let mut metadata = state_db .get_thread(thread_uuid) .await? .expect("thread should be repaired into sqlite"); metadata.cwd = stale_cwd.clone(); state_db.upsert_thread(&metadata).await?; let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { cursor: None, limit: Some(10), sort_key: None, sort_direction: None, model_providers: Some(vec!["mock_provider".to_string()]), source_kinds: None, archived: None, cwd: Some(ThreadListCwdFilter::One( stale_cwd.to_string_lossy().into_owned(), )), use_state_db_only: true, search_term: None, parent_thread_id: None, }) .await?; let resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let state_db_only_response = to_response::<ThreadListResponse>(resp)?; let ids: Vec<_> = state_db_only_response .data .iter() .map(|thread| thread.id.as_str()) .collect(); assert_eq!(ids, vec![thread_id.as_str()]); let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { cursor: None, limit: Some(10), sort_key: None, sort_direction: None, model_providers: Some(vec!["mock_provider".to_string()]), source_kinds: None, archived: None, cwd: Some(ThreadListCwdFilter::One( stale_cwd.to_string_lossy().into_owned(), )), use_state_db_only: false, search_term: None, parent_thread_id: None, }) .await?; let resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let scanned_response = to_response::<ThreadListResponse>(resp)?; assert_eq!(scanned_response.data.len(), 0); Ok(())}#[tokio::test]async fn thread_list_parent_filter_reads_direct_children_from_state_db() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let parent_id = ThreadId::new(); let older_child_id = ThreadId::new(); let newer_child_id = ThreadId::new(); let grandchild_id = ThreadId::new(); let state_db = codex_state::StateRuntime::init( codex_home.path().to_path_buf(), "mock_provider".to_string(), ) .await?; for (thread_id, created_at, source, model_provider) in [ ( older_child_id, "2025-02-01T10:00:00Z", CoreSessionSource::SubAgent(SubAgentSource::Other("agent_job:job-1".to_string())), "other_provider", ), ( newer_child_id, "2025-02-01T11:00:00Z", CoreSessionSource::Cli, "mock_provider", ), ( grandchild_id, "2025-02-01T12:00:00Z", CoreSessionSource::SubAgent(SubAgentSource::Other("agent_job:job-2".to_string())), "mock_provider", ), ] { let created_at = DateTime::parse_from_rfc3339(created_at)?.with_timezone(&Utc); let mut builder = codex_state::ThreadMetadataBuilder::new( thread_id, codex_home.path().join(format!("{thread_id}.jsonl")), created_at, source, ); builder.model_provider = Some(model_provider.to_string()); builder.cwd = codex_home.path().to_path_buf(); builder.cli_version = Some("0.0.0".to_string()); let mut metadata = builder.build(model_provider); metadata.preview = Some("child thread".to_string()); metadata.first_user_message = metadata.preview.clone(); state_db.upsert_thread(&metadata).await?; } for (parent_thread_id, child_thread_id) in [ (parent_id, older_child_id), (parent_id, newer_child_id), (newer_child_id, grandchild_id), ] { state_db .upsert_thread_spawn_edge( parent_thread_id, child_thread_id, DirectionalThreadSpawnEdgeStatus::Open, ) .await?; } state_db .mark_backfill_complete(/*last_watermark*/ None) .await?; let mut mcp = init_mcp(codex_home.path()).await?; let first_page = list_threads_for_parent( &mut mcp, parent_id, /*cursor*/ None, /*limit*/ 1, /*model_providers*/ None, /*source_kinds*/ None, ) .await?; let second_page = list_threads_for_parent( &mut mcp, parent_id, first_page.next_cursor.clone(), /*limit*/ 1, /*model_providers*/ None, /*source_kinds*/ None, ) .await?; assert_eq!( first_page .data .iter() .map(|thread| thread.id.clone()) .collect::<Vec<_>>(), vec![newer_child_id.to_string()] ); assert_eq!( second_page .data .iter() .map(|thread| thread.id.clone()) .collect::<Vec<_>>(), vec![older_child_id.to_string()] ); assert_eq!(second_page.next_cursor, None); let expected_parent_id = parent_id.to_string(); assert!( first_page .data .iter() .chain(&second_page.data) .all(|thread| thread.parent_thread_id.as_deref() == Some(expected_parent_id.as_str())) ); let interactive_only = list_threads_for_parent( &mut mcp, parent_id, /*cursor*/ None, /*limit*/ 10, /*model_providers*/ None, /*source_kinds*/ Some(Vec::new()), ) .await?; assert_eq!( interactive_only .data .iter() .map(|thread| thread.id.clone()) .collect::<Vec<_>>(), vec![newer_child_id.to_string()] ); Ok(())}#[tokio::test]async fn thread_list_parent_filter_rejects_malformed_thread_id() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { cursor: None, limit: Some(10), sort_key: None, sort_direction: None, model_providers: None, source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: Some("not-a-thread-id".to_string()), }) .await?; let error = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_error_message(RequestId::Integer(request_id)), ) .await??; assert_eq!(error.error.code, -32600); Ok(())}#[tokio::test]async fn thread_list_empty_source_kinds_defaults_to_interactive_only() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let cli_id = create_fake_rollout( codex_home.path(), "2025-02-01T10-00-00", "2025-02-01T10:00:00Z", "CLI", Some("mock_provider"), /*git_info*/ None, )?; let exec_id = create_fake_rollout_with_source( codex_home.path(), "2025-02-01T11-00-00", "2025-02-01T11:00:00Z", "Exec", Some("mock_provider"), /*git_info*/ None, CoreSessionSource::Exec, )?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data, next_cursor, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), Some(Vec::new()), /*archived*/ None, ) .await?; assert_eq!(next_cursor, None); let ids: Vec<_> = data.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!(ids, vec![cli_id.as_str()]); assert_ne!(cli_id, exec_id); assert_eq!(data[0].source, SessionSource::Cli); Ok(())}#[tokio::test]async fn thread_list_filters_by_source_kind_subagent_thread_spawn() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let cli_id = create_fake_rollout( codex_home.path(), "2025-02-01T10-00-00", "2025-02-01T10:00:00Z", "CLI", Some("mock_provider"), /*git_info*/ None, )?; let parent_thread_id = ThreadId::from_string(&Uuid::new_v4().to_string())?; let subagent_id = create_fake_rollout_with_source( codex_home.path(), "2025-02-01T11-00-00", "2025-02-01T11:00:00Z", "SubAgent", Some("mock_provider"), /*git_info*/ None, CoreSessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), )?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data, next_cursor, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), Some(vec![ThreadSourceKind::SubAgentThreadSpawn]), /*archived*/ None, ) .await?; assert_eq!(next_cursor, None); let ids: Vec<_> = data.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!(ids, vec![subagent_id.as_str()]); assert_ne!(cli_id, subagent_id); assert!(matches!(data[0].source, SessionSource::SubAgent(_))); assert_eq!(data[0].session_id, subagent_id); Ok(())}#[tokio::test]async fn thread_list_filters_by_subagent_variant() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let parent_thread_id = ThreadId::from_string(&Uuid::new_v4().to_string())?; let review_id = create_fake_parented_rollout_with_source( codex_home.path(), "2025-02-02T09-00-00", "2025-02-02T09:00:00Z", "Review", Some("mock_provider"), /*git_info*/ None, CoreSessionSource::SubAgent(SubAgentSource::Review), parent_thread_id, )?; let compact_id = create_fake_rollout_with_source( codex_home.path(), "2025-02-02T10-00-00", "2025-02-02T10:00:00Z", "Compact", Some("mock_provider"), /*git_info*/ None, CoreSessionSource::SubAgent(SubAgentSource::Compact), )?; let spawn_id = create_fake_rollout_with_source( codex_home.path(), "2025-02-02T11-00-00", "2025-02-02T11:00:00Z", "Spawn", Some("mock_provider"), /*git_info*/ None, CoreSessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth: 1, agent_path: None, agent_nickname: None, agent_role: None, }), )?; let other_id = create_fake_rollout_with_source( codex_home.path(), "2025-02-02T12-00-00", "2025-02-02T12:00:00Z", "Other", Some("mock_provider"), /*git_info*/ None, CoreSessionSource::SubAgent(SubAgentSource::Other("custom".to_string())), )?; let mut mcp = init_mcp(codex_home.path()).await?; let review = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), Some(vec![ThreadSourceKind::SubAgentReview]), /*archived*/ None, ) .await?; let review_ids: Vec<_> = review .data .iter() .map(|thread| thread.id.as_str()) .collect(); assert_eq!(review_ids, vec![review_id.as_str()]); assert_eq!( review.data[0].parent_thread_id, Some(parent_thread_id.to_string()) ); let compact = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), Some(vec![ThreadSourceKind::SubAgentCompact]), /*archived*/ None, ) .await?; let compact_ids: Vec<_> = compact .data .iter() .map(|thread| thread.id.as_str()) .collect(); assert_eq!(compact_ids, vec![compact_id.as_str()]); let spawn = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), Some(vec![ThreadSourceKind::SubAgentThreadSpawn]), /*archived*/ None, ) .await?; let spawn_ids: Vec<_> = spawn.data.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!(spawn_ids, vec![spawn_id.as_str()]); let other = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), Some(vec![ThreadSourceKind::SubAgentOther]), /*archived*/ None, ) .await?; let other_ids: Vec<_> = other.data.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!(other_ids, vec![other_id.as_str()]); Ok(())}#[tokio::test]async fn thread_list_fetches_until_limit_or_exhausted() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; // Newest 16 conversations belong to a different provider; the older 8 are the // only ones that match the filter. We request 8 so the server must keep // paging past the first two pages to reach the desired count. create_fake_rollouts( codex_home.path(), /*count*/ 24, |i| { if i < 16 { "skip_provider" } else { "target_provider" } }, |i| { timestamp_at( /*year*/ 2025, /*month*/ 3, 30 - i as u32, /*hour*/ 12, /*minute*/ 0, /*second*/ 0, ) }, "Hello", )?; let mut mcp = init_mcp(codex_home.path()).await?; // Request 8 threads for the target provider; the matches only start on the // third page so we rely on pagination to reach the limit. let ThreadListResponse { data, next_cursor, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(8), Some(vec!["target_provider".to_string()]), /*source_kinds*/ None, /*archived*/ None, ) .await?; assert_eq!( data.len(), 8, "should keep paging until the requested count is filled" ); assert!( data.iter() .all(|thread| thread.model_provider == "target_provider"), "all returned threads must match the requested provider" ); assert_eq!( next_cursor, None, "once the requested count is satisfied on the final page, nextCursor should be None" ); Ok(())}#[tokio::test]async fn thread_list_enforces_max_limit() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; create_fake_rollouts( codex_home.path(), /*count*/ 105, |_| "mock_provider", |i| { let month = 5 + (i / 28); let day = (i % 28) + 1; timestamp_at( /*year*/ 2025, month as u32, day as u32, /*hour*/ 0, /*minute*/ 0, /*second*/ 0, ) }, "Hello", )?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data, next_cursor, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(200), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, /*archived*/ None, ) .await?; assert_eq!( data.len(), 100, "limit should be clamped to the maximum page size" ); assert!( next_cursor.is_some(), "when more than the maximum exist, nextCursor should continue pagination" ); Ok(())}#[tokio::test]async fn thread_list_stops_when_not_enough_filtered_results_exist() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; // Only the last 7 conversations match the provider filter; we ask for 10 to // ensure the server exhausts pagination without looping forever. create_fake_rollouts( codex_home.path(), /*count*/ 22, |i| { if i < 15 { "skip_provider" } else { "target_provider" } }, |i| { timestamp_at( /*year*/ 2025, /*month*/ 4, 28 - i as u32, /*hour*/ 8, /*minute*/ 0, /*second*/ 0, ) }, "Hello", )?; let mut mcp = init_mcp(codex_home.path()).await?; // Request more threads than exist after filtering; expect all matches to be // returned with nextCursor None. let ThreadListResponse { data, next_cursor, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["target_provider".to_string()]), /*source_kinds*/ None, /*archived*/ None, ) .await?; assert_eq!( data.len(), 7, "all available filtered threads should be returned" ); assert!( data.iter() .all(|thread| thread.model_provider == "target_provider"), "results should still respect the provider filter" ); assert_eq!( next_cursor, None, "when results are exhausted before reaching the limit, nextCursor should be None" ); Ok(())}#[tokio::test]async fn thread_list_includes_git_info() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let git_info = CoreGitInfo { commit_hash: Some(GitSha::new("abc123")), branch: Some("main".to_string()), repository_url: Some("https://example.com/repo.git".to_string()), }; let conversation_id = create_fake_rollout( codex_home.path(), "2025-02-01T09-00-00", "2025-02-01T09:00:00Z", "Git info preview", Some("mock_provider"), Some(git_info), )?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, /*archived*/ None, ) .await?; let thread = data .iter() .find(|t| t.id == conversation_id) .expect("expected thread for created rollout"); let expected_git = ApiGitInfo { sha: Some("abc123".to_string()), branch: Some("main".to_string()), origin_url: Some("https://example.com/repo.git".to_string()), }; assert_eq!(thread.git_info, Some(expected_git)); assert_eq!(thread.source, SessionSource::Cli); assert_eq!(thread.cwd, test_absolute_path("/")); assert_eq!(thread.cli_version, "0.0.0"); Ok(())}#[tokio::test]async fn thread_list_default_sorts_by_created_at() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let id_a = create_fake_rollout( codex_home.path(), "2025-01-02T12-00-00", "2025-01-02T12:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let id_b = create_fake_rollout( codex_home.path(), "2025-01-01T13-00-00", "2025-01-01T13:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let id_c = create_fake_rollout( codex_home.path(), "2025-01-01T12-00-00", "2025-01-01T12:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data, .. } = list_threads_with_sort( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, /*sort_key*/ None, /*archived*/ None, ) .await?; let ids: Vec<_> = data.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!(ids, vec![id_a.as_str(), id_b.as_str(), id_c.as_str()]); Ok(())}#[tokio::test]async fn thread_list_sort_updated_at_orders_by_mtime() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let id_old = create_fake_rollout( codex_home.path(), "2025-01-01T10-00-00", "2025-01-01T10:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let id_mid = create_fake_rollout( codex_home.path(), "2025-01-01T11-00-00", "2025-01-01T11:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let id_new = create_fake_rollout( codex_home.path(), "2025-01-01T12-00-00", "2025-01-01T12:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; set_rollout_mtime( rollout_path(codex_home.path(), "2025-01-01T10-00-00", &id_old).as_path(), "2025-01-03T00:00:00Z", )?; set_rollout_mtime( rollout_path(codex_home.path(), "2025-01-01T11-00-00", &id_mid).as_path(), "2025-01-02T00:00:00Z", )?; set_rollout_mtime( rollout_path(codex_home.path(), "2025-01-01T12-00-00", &id_new).as_path(), "2025-01-01T00:00:00Z", )?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data, .. } = list_threads_with_sort( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, Some(ThreadSortKey::UpdatedAt), /*archived*/ None, ) .await?; let ids: Vec<_> = data.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!(ids, vec![id_old.as_str(), id_mid.as_str(), id_new.as_str()]); Ok(())}#[tokio::test]async fn thread_list_updated_at_paginates_with_cursor() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let id_a = create_fake_rollout( codex_home.path(), "2025-02-01T10-00-00", "2025-02-01T10:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let id_b = create_fake_rollout( codex_home.path(), "2025-02-01T11-00-00", "2025-02-01T11:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let id_c = create_fake_rollout( codex_home.path(), "2025-02-01T12-00-00", "2025-02-01T12:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; set_rollout_mtime( rollout_path(codex_home.path(), "2025-02-01T10-00-00", &id_a).as_path(), "2025-02-03T00:00:00Z", )?; set_rollout_mtime( rollout_path(codex_home.path(), "2025-02-01T11-00-00", &id_b).as_path(), "2025-02-02T00:00:00Z", )?; set_rollout_mtime( rollout_path(codex_home.path(), "2025-02-01T12-00-00", &id_c).as_path(), "2025-02-01T00:00:00Z", )?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data: page1, next_cursor: cursor1, .. } = list_threads_with_sort( &mut mcp, /*cursor*/ None, Some(2), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, Some(ThreadSortKey::UpdatedAt), /*archived*/ None, ) .await?; let ids_page1: Vec<_> = page1.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!(ids_page1, vec![id_a.as_str(), id_b.as_str()]); let cursor1 = cursor1.expect("expected nextCursor on first page"); let ThreadListResponse { data: page2, next_cursor: cursor2, .. } = list_threads_with_sort( &mut mcp, Some(cursor1), Some(2), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, Some(ThreadSortKey::UpdatedAt), /*archived*/ None, ) .await?; let ids_page2: Vec<_> = page2.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!(ids_page2, vec![id_c.as_str()]); assert_eq!(cursor2, None); Ok(())}#[tokio::test]async fn thread_list_backwards_cursor_can_seed_forward_delta_sync() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let id_old = create_fake_rollout( codex_home.path(), "2025-02-01T10-00-00", "2025-02-01T10:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let id_watermark = create_fake_rollout( codex_home.path(), "2025-02-01T11-00-00", "2025-02-01T11:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; set_rollout_mtime( rollout_path(codex_home.path(), "2025-02-01T10-00-00", &id_old).as_path(), "2025-02-02T00:00:00Z", )?; set_rollout_mtime( rollout_path(codex_home.path(), "2025-02-01T11-00-00", &id_watermark).as_path(), "2025-02-03T00:00:00Z", )?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data: page1, backwards_cursor, .. } = { let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { cursor: None, limit: Some(1), sort_key: Some(ThreadSortKey::UpdatedAt), sort_direction: Some(SortDirection::Desc), model_providers: Some(vec!["mock_provider".to_string()]), source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: None, }) .await?; let resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; to_response::<ThreadListResponse>(resp)? }; let ids_page1: Vec<_> = page1.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!(ids_page1, vec![id_watermark.as_str()]); let backwards_cursor = backwards_cursor.expect("expected backwardsCursor on first page"); assert_eq!(backwards_cursor, "2025-02-02T23:59:59.999Z"); let id_new = create_fake_rollout( codex_home.path(), "2025-02-01T12-00-00", "2025-02-01T12:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; set_rollout_mtime( rollout_path(codex_home.path(), "2025-02-01T12-00-00", &id_new).as_path(), "2025-02-04T00:00:00Z", )?; let ThreadListResponse { data: delta_page, .. } = { let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { cursor: Some(backwards_cursor), limit: Some(10), sort_key: Some(ThreadSortKey::UpdatedAt), sort_direction: Some(SortDirection::Asc), model_providers: Some(vec!["mock_provider".to_string()]), source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: None, }) .await?; let resp: JSONRPCResponse = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; to_response::<ThreadListResponse>(resp)? }; let ids_delta: Vec<_> = delta_page.iter().map(|thread| thread.id.as_str()).collect(); assert_eq!(ids_delta, vec![id_watermark.as_str(), id_new.as_str()]); Ok(())}#[tokio::test]async fn thread_list_created_at_tie_breaks_by_uuid() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let id_a = create_fake_rollout( codex_home.path(), "2025-02-01T10-00-00", "2025-02-01T10:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let id_b = create_fake_rollout( codex_home.path(), "2025-02-01T10-00-00", "2025-02-01T10:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, /*archived*/ None, ) .await?; let ids: Vec<_> = data.iter().map(|thread| thread.id.as_str()).collect(); let mut expected = [id_a, id_b]; expected.sort_by_key(|id| Reverse(Uuid::parse_str(id).expect("uuid should parse"))); let expected: Vec<_> = expected.iter().map(String::as_str).collect(); assert_eq!(ids, expected); Ok(())}#[tokio::test]async fn thread_list_updated_at_tie_breaks_by_uuid() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let id_a = create_fake_rollout( codex_home.path(), "2025-02-01T10-00-00", "2025-02-01T10:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let id_b = create_fake_rollout( codex_home.path(), "2025-02-01T11-00-00", "2025-02-01T11:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; let updated_at = "2025-02-03T00:00:00Z"; set_rollout_mtime( rollout_path(codex_home.path(), "2025-02-01T10-00-00", &id_a).as_path(), updated_at, )?; set_rollout_mtime( rollout_path(codex_home.path(), "2025-02-01T11-00-00", &id_b).as_path(), updated_at, )?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data, .. } = list_threads_with_sort( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, Some(ThreadSortKey::UpdatedAt), /*archived*/ None, ) .await?; let ids: Vec<_> = data.iter().map(|thread| thread.id.as_str()).collect(); let mut expected = [id_a, id_b]; expected.sort_by_key(|id| Reverse(Uuid::parse_str(id).expect("uuid should parse"))); let expected: Vec<_> = expected.iter().map(String::as_str).collect(); assert_eq!(ids, expected); Ok(())}#[tokio::test]async fn thread_list_updated_at_uses_mtime() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let thread_id = create_fake_rollout( codex_home.path(), "2025-02-01T10-00-00", "2025-02-01T10:00:00Z", "Hello", Some("mock_provider"), /*git_info*/ None, )?; set_rollout_mtime( rollout_path(codex_home.path(), "2025-02-01T10-00-00", &thread_id).as_path(), "2025-02-05T00:00:00Z", )?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data, .. } = list_threads_with_sort( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, Some(ThreadSortKey::UpdatedAt), /*archived*/ None, ) .await?; let thread = data .iter() .find(|item| item.id == thread_id) .expect("expected thread for created rollout"); let expected_created = chrono::DateTime::parse_from_rfc3339("2025-02-01T10:00:00Z")?.timestamp(); let expected_updated = chrono::DateTime::parse_from_rfc3339("2025-02-05T00:00:00Z")?.timestamp(); assert_eq!(thread.created_at, expected_created); assert_eq!(thread.updated_at, expected_updated); Ok(())}#[tokio::test]async fn thread_list_archived_filter() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let active_id = create_fake_rollout( codex_home.path(), "2025-03-01T10-00-00", "2025-03-01T10:00:00Z", "Active", Some("mock_provider"), /*git_info*/ None, )?; let archived_id = create_fake_rollout( codex_home.path(), "2025-03-01T09-00-00", "2025-03-01T09:00:00Z", "Archived", Some("mock_provider"), /*git_info*/ None, )?; let archived_dir = codex_home.path().join(ARCHIVED_SESSIONS_SUBDIR); fs::create_dir_all(&archived_dir)?; let archived_source = rollout_path(codex_home.path(), "2025-03-01T09-00-00", &archived_id); let archived_dest = archived_dir.join( archived_source .file_name() .expect("archived rollout should have a file name"), ); fs::rename(&archived_source, &archived_dest)?; let mut mcp = init_mcp(codex_home.path()).await?; let ThreadListResponse { data, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, /*archived*/ None, ) .await?; assert_eq!(data.len(), 1); assert_eq!(data[0].id, active_id); let ThreadListResponse { data, .. } = list_threads( &mut mcp, /*cursor*/ None, Some(10), Some(vec!["mock_provider".to_string()]), /*source_kinds*/ None, Some(true), ) .await?; assert_eq!(data.len(), 1); assert_eq!(data[0].id, archived_id); Ok(())}#[tokio::test]async fn thread_list_invalid_cursor_returns_error() -> Result<()> { let codex_home = TempDir::new()?; create_minimal_config(codex_home.path())?; let mut mcp = init_mcp(codex_home.path()).await?; let request_id = mcp .send_thread_list_request(codex_app_server_protocol::ThreadListParams { cursor: Some("not-a-cursor".to_string()), limit: Some(2), sort_key: None, sort_direction: None, model_providers: Some(vec!["mock_provider".to_string()]), source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: None, }) .await?; let error: JSONRPCError = timeout( DEFAULT_READ_TIMEOUT, mcp.read_stream_until_error_message(RequestId::Integer(request_id)), ) .await??; assert_eq!(error.error.code, -32600); assert_eq!(error.error.message, "invalid cursor: not-a-cursor"); Ok(())}