diff --git a/codex-rs/app-server/src/request_processors/external_agent_session_import.rs b/codex-rs/app-server/src/request_processors/external_agent_session_import.rs index e0a53a9c98..cc21d3567e 100644 --- a/codex-rs/app-server/src/request_processors/external_agent_session_import.rs +++ b/codex-rs/app-server/src/request_processors/external_agent_session_import.rs @@ -1,6 +1,7 @@ use std::path::PathBuf; use std::sync::Arc; +use chrono::DateTime; use chrono::Utc; use codex_arg0::Arg0DispatchPaths; use codex_core::ThreadManager; @@ -20,6 +21,7 @@ use codex_protocol::protocol::ThreadMemoryMode; use codex_rollout::is_persisted_rollout_item; use codex_thread_store::AppendThreadItemsParams; use codex_thread_store::CreateThreadParams; +use codex_thread_store::InitialThreadTimestamps; use codex_thread_store::ThreadMetadataPatch; use codex_thread_store::ThreadPersistenceMetadata; use codex_thread_store::ThreadStore; @@ -168,6 +170,8 @@ impl ExternalAgentSessionImporter { cwd, title, first_user_message, + source_created_at, + source_updated_at, mut rollout_items, } = session; let config = self @@ -203,7 +207,23 @@ impl ExternalAgentSessionImporter { } else { ThreadMemoryMode::Disabled }; - let now = Utc::now(); + let initial_timestamps = source_created_at + .zip(source_updated_at) + .and_then(|(created_at, updated_at)| { + Some(InitialThreadTimestamps { + created_at: DateTime::from_timestamp(created_at, /*nsecs*/ 0)?, + updated_at: DateTime::from_timestamp(updated_at, /*nsecs*/ 0)?, + }) + }) + .unwrap_or_else(|| { + let now = Utc::now(); + InitialThreadTimestamps { + created_at: now, + updated_at: now, + } + }); + let created_at = initial_timestamps.created_at; + let updated_at = initial_timestamps.updated_at; let create_params = CreateThreadParams { session_id: thread_id.into(), thread_id, @@ -238,8 +258,9 @@ impl ExternalAgentSessionImporter { title, preview: first_user_message.clone(), model_provider: Some(model_provider), - created_at: Some(now), - updated_at: Some(now), + created_at: Some(created_at), + updated_at: Some(updated_at), + advance_recency_at: Some(updated_at), source: Some(source.clone()), thread_source: Some(None), agent_nickname: Some(source.get_nickname()), @@ -253,7 +274,7 @@ impl ExternalAgentSessionImporter { }; self.thread_store - .create_thread(create_params) + .create_thread_with_initial_timestamps(create_params, initial_timestamps) .await .map_err(|err| format!("failed to import session: {err}"))?; if !rollout_items.is_empty() diff --git a/codex-rs/app-server/tests/suite/v2/external_agent_config.rs b/codex-rs/app-server/tests/suite/v2/external_agent_config.rs index e83b636618..5c659e3a43 100644 --- a/codex-rs/app-server/tests/suite/v2/external_agent_config.rs +++ b/codex-rs/app-server/tests/suite/v2/external_agent_config.rs @@ -18,6 +18,8 @@ use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::PluginListParams; use codex_app_server_protocol::PluginListResponse; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::SortDirection; +use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadListResponse; @@ -25,6 +27,9 @@ use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadSortKey; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::UserInput; use codex_config::types::AuthCredentialsStoreMode; @@ -51,6 +56,58 @@ fn assert_import_response(response: ExternalAgentConfigImportResponse) -> String response.import_id } +async fn list_threads( + mcp: &mut TestAppServer, + sort_key: ThreadSortKey, + use_state_db_only: bool, +) -> Result { + let request_id = mcp + .send_thread_list_request(ThreadListParams { + cursor: None, + limit: None, + sort_key: Some(sort_key), + sort_direction: Some(SortDirection::Desc), + model_providers: None, + source_kinds: None, + archived: None, + cwd: None, + use_state_db_only, + search_term: None, + parent_thread_id: None, + ancestor_thread_id: None, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + to_response(response) +} + +fn assert_thread_timestamps(thread: &Thread, created_at: i64, updated_at: i64) { + assert_eq!( + (thread.created_at, thread.updated_at, thread.recency_at), + (created_at, updated_at, Some(updated_at)) + ); +} + +fn remove_state_db_files(codex_home: &Path) -> Result<()> { + let state_db_path = codex_state::state_db_path(codex_home); + for path in [ + state_db_path.clone(), + PathBuf::from(format!("{}-wal", state_db_path.display())), + PathBuf::from(format!("{}-shm", state_db_path.display())), + ] { + match std::fs::remove_file(path) { + Ok(()) => {} + Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} + Err(err) => return Err(err.into()), + } + } + Ok(()) +} + #[tokio::test] async fn external_agent_config_import_sends_completion_notification_for_sync_only_import() -> Result<()> { @@ -612,7 +669,12 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; let project_root = codex_home.path().join("repo"); - let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + let source_created_at_text = "2024-01-02T03:04:05Z"; + let source_updated_at_text = "2024-02-03T04:05:06Z"; + let source_created_at = + chrono::DateTime::parse_from_rfc3339(source_created_at_text)?.timestamp(); + let source_updated_at = + chrono::DateTime::parse_from_rfc3339(source_updated_at_text)?.timestamp(); let session_dir = external_agent_home(codex_home.path()).join("projects/repo"); let session_path = session_dir.join("session.jsonl"); let control_request = "src/auth.rs:1-5"; @@ -625,21 +687,21 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { serde_json::json!({ "type": "user", "cwd": &project_root, - "timestamp": &recent_timestamp, + "timestamp": source_created_at_text, "message": { "content": control_request }, }) .to_string(), serde_json::json!({ "type": "user", "cwd": &project_root, - "timestamp": &recent_timestamp, + "timestamp": "2024-01-03T00:00:00Z", "message": { "content": first_request }, }) .to_string(), serde_json::json!({ "type": "assistant", "cwd": &project_root, - "timestamp": &recent_timestamp, + "timestamp": source_updated_at_text, "message": { "content": "first answer" }, }) .to_string(), @@ -656,6 +718,42 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + let request_id = mcp + .send_thread_start_request(ThreadStartParams { + cwd: Some(project_root.display().to_string()), + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let native_thread: ThreadStartResponse = to_response(response)?; + let native_thread_id = native_thread.thread.id; + let request_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: native_thread_id.clone(), + client_user_message_id: None, + input: vec![UserInput::Text { + text: "newer native request".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + let request_id = mcp .send_raw_request( "externalAgentConfig/detect", @@ -727,37 +825,58 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { .expect("session success should include imported thread id") .to_string(); - let request_id = mcp - .send_thread_list_request(ThreadListParams { - cursor: None, - limit: None, - 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: None, - ancestor_thread_id: None, - }) - .await?; - let response: JSONRPCResponse = timeout( - DEFAULT_TIMEOUT, - mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + let response = list_threads( + &mut mcp, + ThreadSortKey::CreatedAt, + /*use_state_db_only*/ false, ) - .await??; - let response: ThreadListResponse = to_response(response)?; + .await?; + assert_eq!( + response + .data + .iter() + .map(|thread| thread.id.as_str()) + .collect::>(), + vec![native_thread_id.as_str(), imported_thread_id.as_str()] + ); let thread = response .data - .first() + .iter() + .find(|thread| thread.id == imported_thread_id) .expect("expected imported thread") .clone(); - assert_eq!(imported_thread_id, thread.id.to_string()); + let native_thread = response + .data + .iter() + .find(|thread| thread.id == native_thread_id) + .expect("expected native thread"); + assert_thread_timestamps(&thread, source_created_at, source_updated_at); + assert!(native_thread.created_at > source_updated_at); + assert!(native_thread.updated_at > source_updated_at); + assert!( + native_thread + .recency_at + .is_some_and(|value| value > source_updated_at) + ); assert_eq!(thread.preview, control_request); assert_eq!(thread.name.as_deref(), Some("Fix auth flow")); + for sort_key in [ThreadSortKey::UpdatedAt, ThreadSortKey::RecencyAt] { + let response = list_threads(&mut mcp, sort_key, /*use_state_db_only*/ false).await?; + assert_eq!( + response.data.first().map(|thread| thread.id.as_str()), + Some(native_thread_id.as_str()) + ); + } + + let rollout_path = thread.path.clone().expect("imported rollout path"); + let session_meta = codex_rollout::read_session_meta_line(&rollout_path).await?; + let rollout_modified_at = std::fs::metadata(&rollout_path)? + .modified() + .map(chrono::DateTime::::from)?; + assert_eq!(session_meta.meta.timestamp, "2024-01-02T03:04:05.000Z"); + assert_eq!(rollout_modified_at.timestamp(), source_updated_at); + let request_id = mcp .send_thread_read_request(ThreadReadParams { thread_id: thread.id.clone(), @@ -809,9 +928,43 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { }) ); + drop(mcp); + remove_state_db_files(codex_home.path())?; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .with_env_overrides(&[("HOME", Some(home_dir.as_str()))]) + .build() + .await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + list_threads( + &mut mcp, + ThreadSortKey::CreatedAt, + /*use_state_db_only*/ false, + ) + .await?; + for sort_key in [ + ThreadSortKey::CreatedAt, + ThreadSortKey::UpdatedAt, + ThreadSortKey::RecencyAt, + ] { + let rebuilt = list_threads(&mut mcp, sort_key, /*use_state_db_only*/ true).await?; + assert_eq!( + rebuilt.data.first().map(|thread| thread.id.as_str()), + Some(native_thread_id.as_str()) + ); + let rebuilt_thread = rebuilt + .data + .iter() + .find(|thread| thread.id == imported_thread_id) + .expect("rebuilt imported thread"); + assert_thread_timestamps(rebuilt_thread, source_created_at, source_updated_at); + } + let request_id = mcp .send_thread_resume_request(ThreadResumeParams { - thread_id: thread.id.clone(), + thread_id: imported_thread_id.clone(), ..Default::default() }) .await?; @@ -824,7 +977,7 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { let request_id = mcp .send_turn_start_request(TurnStartParams { - thread_id: thread.id.clone(), + thread_id: imported_thread_id.clone(), client_user_message_id: None, input: vec![UserInput::Text { text: "follow up".to_string(), @@ -844,9 +997,36 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { ) .await??; + let response = list_threads( + &mut mcp, + ThreadSortKey::RecencyAt, + /*use_state_db_only*/ true, + ) + .await?; + let updated_thread = response + .data + .iter() + .find(|thread| thread.id == imported_thread_id) + .expect("updated imported thread"); + assert_eq!(updated_thread.created_at, source_created_at); + assert!(updated_thread.updated_at > source_updated_at); + assert!( + updated_thread + .recency_at + .is_some_and(|value| value > source_updated_at) + ); + assert_eq!( + response.data.first().map(|thread| thread.id.as_str()), + Some(imported_thread_id.as_str()) + ); + let resumed_rollout_modified_at = std::fs::metadata(&rollout_path)? + .modified() + .map(chrono::DateTime::::from)?; + assert!(resumed_rollout_modified_at.timestamp() > source_updated_at); + let request_id = mcp .send_thread_read_request(ThreadReadParams { - thread_id: thread.id, + thread_id: imported_thread_id.clone(), include_turns: true, }) .await?; @@ -862,6 +1042,40 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { other => panic!("expected agent message item, got {other:?}"), } + drop(mcp); + remove_state_db_files(codex_home.path())?; + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .with_env_overrides(&[("HOME", Some(home_dir.as_str()))]) + .build() + .await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + list_threads( + &mut mcp, + ThreadSortKey::RecencyAt, + /*use_state_db_only*/ false, + ) + .await?; + let repaired_after_resume = list_threads( + &mut mcp, + ThreadSortKey::RecencyAt, + /*use_state_db_only*/ true, + ) + .await?; + let repaired_thread = repaired_after_resume + .data + .iter() + .find(|thread| thread.id == imported_thread_id) + .expect("repaired resumed imported thread"); + assert_eq!(repaired_thread.created_at, source_created_at); + assert!(repaired_thread.updated_at > source_updated_at); + assert!( + repaired_thread + .recency_at + .is_some_and(|value| value > source_updated_at) + ); + Ok(()) } diff --git a/codex-rs/external-agent-sessions/src/export.rs b/codex-rs/external-agent-sessions/src/export.rs index 992770f855..750a3d2263 100644 --- a/codex-rs/external-agent-sessions/src/export.rs +++ b/codex-rs/external-agent-sessions/src/export.rs @@ -39,6 +39,9 @@ pub(crate) fn load_session_for_import_with_content_sha256( return Ok(None); }; let messages = parsed.messages; + let (source_created_at, source_updated_at) = source_chronology(&messages) + .map(|(created_at, updated_at)| (Some(created_at), Some(updated_at))) + .unwrap_or_default(); let first_user_message_text = messages .iter() .find(|message| message.role == MessageRole::User) @@ -64,12 +67,27 @@ pub(crate) fn load_session_for_import_with_content_sha256( cwd, title, first_user_message, + source_created_at, + source_updated_at, rollout_items, }, parsed.content_sha256, ))) } +fn source_chronology(messages: &[ConversationMessage]) -> Option<(i64, i64)> { + let mut timestamps = messages + .iter() + .skip_while(|message| message.role != MessageRole::User) + .filter_map(|message| message.timestamp); + let first = timestamps.next()?; + Some( + timestamps.fold((first, first), |(created_at, updated_at), timestamp| { + (created_at.min(timestamp), updated_at.max(timestamp)) + }), + ) +} + fn rollout_items_from_messages(messages: Vec) -> Vec { let mut items = Vec::new(); let mut current_turn = None; @@ -504,8 +522,107 @@ mod tests { assert_eq!(token_count.total_token_usage, token_count.last_token_usage); } + #[test] + fn derives_source_chronology_from_retained_valid_message_timestamps() { + let root = TempDir::new().expect("tempdir"); + let project_root = root.path().join("repo"); + std::fs::create_dir_all(&project_root).expect("project root"); + let path = root.path().join("session.jsonl"); + std::fs::write( + &path, + jsonl(&[ + record_at( + "assistant", + "discarded before first user", + &project_root, + Some("2020-01-01T00:00:00Z"), + ), + record_at( + "user", + "first retained message", + &project_root, + Some("2025-03-04T05:06:07Z"), + ), + record_at( + "assistant", + "invalid timestamp", + &project_root, + Some("not-a-timestamp"), + ), + record_at( + "assistant", + "earliest retained timestamp", + &project_root, + Some("2024-01-02T03:04:05Z"), + ), + record_at( + "assistant", + "latest retained timestamp", + &project_root, + Some("2026-05-06T07:08:09Z"), + ), + ]), + ) + .expect("session"); + + let imported = load_session_for_import(&path) + .expect("load") + .expect("session"); + + assert_eq!( + (imported.source_created_at, imported.source_updated_at), + ( + chrono::DateTime::parse_from_rfc3339("2024-01-02T03:04:05Z") + .ok() + .map(|timestamp| timestamp.timestamp()), + chrono::DateTime::parse_from_rfc3339("2026-05-06T07:08:09Z") + .ok() + .map(|timestamp| timestamp.timestamp()), + ) + ); + } + + #[test] + fn leaves_source_chronology_empty_without_valid_timestamps() { + let root = TempDir::new().expect("tempdir"); + let project_root = root.path().join("repo"); + std::fs::create_dir_all(&project_root).expect("project root"); + let path = root.path().join("session.jsonl"); + std::fs::write( + &path, + jsonl(&[ + record_at( + "user", + "missing timestamp", + &project_root, + /*timestamp*/ None, + ), + record_at( + "assistant", + "invalid timestamp", + &project_root, + Some("invalid"), + ), + ]), + ) + .expect("session"); + + let imported = load_session_for_import(&path) + .expect("load") + .expect("session"); + + assert_eq!( + (imported.source_created_at, imported.source_updated_at), + (None, None) + ); + } + fn record(role: &str, text: &str, cwd: &Path) -> JsonValue { let timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); + record_at(role, text, cwd, Some(timestamp.as_str())) + } + + fn record_at(role: &str, text: &str, cwd: &Path, timestamp: Option<&str>) -> JsonValue { serde_json::json!({ "type": role, "cwd": cwd, diff --git a/codex-rs/external-agent-sessions/src/lib.rs b/codex-rs/external-agent-sessions/src/lib.rs index b13322dffb..c65ef705eb 100644 --- a/codex-rs/external-agent-sessions/src/lib.rs +++ b/codex-rs/external-agent-sessions/src/lib.rs @@ -33,6 +33,8 @@ pub struct ImportedExternalAgentSession { pub cwd: PathBuf, pub title: Option, pub first_user_message: Option, + pub source_created_at: Option, + pub source_updated_at: Option, pub rollout_items: Vec, } diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 87e97997a2..ecc5f7646c 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -8,8 +8,11 @@ use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; +use std::time::SystemTime; +use chrono::DateTime; use chrono::SecondsFormat; +use chrono::Utc; use codex_protocol::SessionId; use codex_protocol::ThreadId; use codex_protocol::capabilities::SelectedCapabilityRoot; @@ -82,6 +85,7 @@ pub struct RolloutRecorder { } #[derive(Clone)] +#[allow(clippy::large_enum_variant)] pub enum RolloutRecorderParams { Create { session_id: SessionId, @@ -97,6 +101,8 @@ pub enum RolloutRecorderParams { multi_agent_version: Option, history_mode: ThreadHistoryMode, initial_window_id: Option, + initial_created_at: Option>, + initial_updated_at: Option>, }, Resume { path: PathBuf, @@ -190,6 +196,8 @@ impl RolloutRecorderParams { multi_agent_version: None, history_mode: Default::default(), initial_window_id: None, + initial_created_at: None, + initial_updated_at: None, } } @@ -249,6 +257,23 @@ impl RolloutRecorderParams { self } + pub fn with_initial_timestamps( + mut self, + created_at: DateTime, + updated_at: DateTime, + ) -> Self { + if let Self::Create { + initial_created_at, + initial_updated_at, + .. + } = &mut self + { + *initial_created_at = Some(created_at); + *initial_updated_at = Some(updated_at); + } + self + } + pub fn resume(path: PathBuf) -> Self { Self::Resume { path } } @@ -766,8 +791,15 @@ impl RolloutRecorder { multi_agent_version, history_mode, initial_window_id, + initial_created_at, + initial_updated_at, } => { - let log_file_info = precompute_log_file_info(config, conversation_id)?; + let log_file_info = precompute_log_file_info( + config, + conversation_id, + initial_created_at, + initial_updated_at, + )?; let path = log_file_info.path.clone(); let thread_id = log_file_info.conversation_id; let started_at = log_file_info.timestamp; @@ -1493,15 +1525,31 @@ struct LogFileInfo { /// Timestamp for the start of the session. timestamp: OffsetDateTime, + + /// Initial file modification time used when reconstructing imported chronology. + initial_modified_at: Option, } fn precompute_log_file_info( config: &impl RolloutConfigView, conversation_id: ThreadId, + initial_created_at: Option>, + initial_updated_at: Option>, ) -> std::io::Result { // Resolve ~/.codex/sessions/YYYY/MM/DD path. - let timestamp = OffsetDateTime::now_local() - .map_err(|e| IoError::other(format!("failed to get local time: {e}")))?; + let timestamp = match initial_created_at { + Some(created_at) => { + let timestamp = OffsetDateTime::from(SystemTime::from(created_at)); + let local_offset = time::UtcOffset::local_offset_at(timestamp).map_err(|e| { + IoError::other(format!( + "failed to get local offset for initial session time: {e}" + )) + })?; + timestamp.to_offset(local_offset) + } + None => OffsetDateTime::now_local() + .map_err(|e| IoError::other(format!("failed to get local time: {e}")))?, + }; let mut dir = config.codex_home().to_path_buf(); dir.push(SESSIONS_SUBDIR); dir.push(timestamp.year().to_string()); @@ -1524,6 +1572,7 @@ fn precompute_log_file_info( path, conversation_id, timestamp, + initial_modified_at: initial_updated_at.map(SystemTime::from), }) } @@ -1554,6 +1603,7 @@ struct RolloutWriterState { meta: Option, cwd: PathBuf, rollout_path: PathBuf, + initial_modified_at: Option, last_logged_error: Option, } @@ -1565,6 +1615,9 @@ impl RolloutWriterState { cwd: PathBuf, rollout_path: PathBuf, ) -> Self { + let initial_modified_at = deferred_log_file_info + .as_ref() + .and_then(|info| info.initial_modified_at); Self { writer: file.map(|file| JsonlWriter { file }), deferred_log_file_info, @@ -1572,6 +1625,7 @@ impl RolloutWriterState { meta, cwd, rollout_path, + initial_modified_at, last_logged_error: None, } } @@ -1689,6 +1743,13 @@ impl RolloutWriterState { if let Some(writer) = self.writer.as_mut() { writer.file.flush().await?; } + if let Some(modified_at) = self.initial_modified_at { + fs::OpenOptions::new() + .append(true) + .open(&self.rollout_path)? + .set_times(fs::FileTimes::new().set_modified(modified_at))?; + self.initial_modified_at = None; + } Ok(()) } diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index cb3f198ff3..9dadb852c2 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -22,6 +22,7 @@ use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::DeleteThreadParams; +use crate::InitialThreadTimestamps; use crate::ListThreadsParams; use crate::LoadThreadHistoryParams; use crate::ReadThreadByRolloutPathParams; @@ -612,6 +613,14 @@ impl ThreadStore for InMemoryThreadStore { Box::pin(InMemoryThreadStore::create_thread(self, params)) } + fn create_thread_with_initial_timestamps( + &self, + params: CreateThreadParams, + _timestamps: InitialThreadTimestamps, + ) -> ThreadStoreFuture<'_, ()> { + Box::pin(InMemoryThreadStore::create_thread(self, params)) + } + fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreFuture<'_, ()> { Box::pin(InMemoryThreadStore::resume_thread(self, params)) } diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 34a0ab2ba1..0819576883 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -29,6 +29,7 @@ pub use types::CreateThreadParams; pub use types::DeleteThreadParams; pub use types::ExtraConfig; pub use types::GitInfoPatch; +pub use types::InitialThreadTimestamps; pub use types::ItemPage; pub use types::ListItemsParams; pub use types::ListThreadsParams; diff --git a/codex-rs/thread-store/src/local/create_thread.rs b/codex-rs/thread-store/src/local/create_thread.rs index 9b370ef095..ea8f2bd27f 100644 --- a/codex-rs/thread-store/src/local/create_thread.rs +++ b/codex-rs/thread-store/src/local/create_thread.rs @@ -1,5 +1,6 @@ use super::LocalThreadStore; use crate::CreateThreadParams; +use crate::InitialThreadTimestamps; use crate::ThreadStoreError; use crate::ThreadStoreResult; use crate::error::reject_paginated_history_mode; @@ -11,6 +12,22 @@ use codex_rollout::RolloutRecorderParams; pub(super) async fn create_thread( store: &LocalThreadStore, params: CreateThreadParams, +) -> ThreadStoreResult { + create_thread_recorder(store, params, /*initial_timestamps*/ None).await +} + +pub(super) async fn create_thread_with_initial_timestamps( + store: &LocalThreadStore, + params: CreateThreadParams, + initial_timestamps: InitialThreadTimestamps, +) -> ThreadStoreResult { + create_thread_recorder(store, params, Some(initial_timestamps)).await +} + +async fn create_thread_recorder( + store: &LocalThreadStore, + params: CreateThreadParams, + initial_timestamps: Option, ) -> ThreadStoreResult { reject_paginated_history_mode(params.history_mode)?; let cwd = params @@ -27,26 +44,30 @@ pub(super) async fn create_thread( model_provider_id: params.metadata.model_provider.clone(), generate_memories: matches!(params.metadata.memory_mode, ThreadMemoryMode::Enabled), }; - RolloutRecorder::new( - &config, - RolloutRecorderParams::new( - params.thread_id, - params.forked_from_id, - params.parent_thread_id, - params.source, - params.thread_source, - params.originator, - params.base_instructions, - params.dynamic_tools, - ) - .with_session_id(params.session_id) - .with_selected_capability_roots(params.selected_capability_roots) - .with_multi_agent_version(params.multi_agent_version) - .with_history_mode(params.history_mode) - .with_initial_window_id(params.initial_window_id), + let recorder_params = RolloutRecorderParams::new( + params.thread_id, + params.forked_from_id, + params.parent_thread_id, + params.source, + params.thread_source, + params.originator, + params.base_instructions, + params.dynamic_tools, ) - .await - .map_err(|err| ThreadStoreError::Internal { - message: format!("failed to initialize local thread recorder: {err}"), - }) + .with_session_id(params.session_id) + .with_selected_capability_roots(params.selected_capability_roots) + .with_multi_agent_version(params.multi_agent_version) + .with_history_mode(params.history_mode) + .with_initial_window_id(params.initial_window_id); + let recorder_params = match initial_timestamps { + Some(timestamps) => { + recorder_params.with_initial_timestamps(timestamps.created_at, timestamps.updated_at) + } + None => recorder_params, + }; + RolloutRecorder::new(&config, recorder_params) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to initialize local thread recorder: {err}"), + }) } diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index 74e707bb58..4424cc3df3 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -12,6 +12,7 @@ use super::LocalThreadStore; use super::create_thread; use crate::AppendThreadItemsParams; use crate::CreateThreadParams; +use crate::InitialThreadTimestamps; use crate::ReadThreadParams; use crate::ResumeThreadParams; use crate::ThreadStoreError; @@ -34,6 +35,21 @@ pub(super) async fn create_thread( .await } +pub(super) async fn create_thread_with_initial_timestamps( + store: &LocalThreadStore, + params: CreateThreadParams, + timestamps: InitialThreadTimestamps, +) -> ThreadStoreResult<()> { + let thread_id = params.thread_id; + let history_mode = params.history_mode; + store.ensure_live_recorder_absent(thread_id).await?; + let recorder = + create_thread::create_thread_with_initial_timestamps(store, params, timestamps).await?; + store + .insert_live_recorder(thread_id, recorder, history_mode) + .await +} + pub(super) async fn resume_thread( store: &LocalThreadStore, params: ResumeThreadParams, diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 5939ea606e..962c6367f9 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -26,6 +26,7 @@ use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::DeleteThreadParams; +use crate::InitialThreadTimestamps; use crate::ListThreadsParams; use crate::LoadThreadHistoryParams; use crate::ReadThreadByRolloutPathParams; @@ -246,6 +247,16 @@ impl ThreadStore for LocalThreadStore { Box::pin(async move { live_writer::create_thread(self, params).await }) } + fn create_thread_with_initial_timestamps( + &self, + params: CreateThreadParams, + timestamps: InitialThreadTimestamps, + ) -> ThreadStoreFuture<'_, ()> { + Box::pin(async move { + live_writer::create_thread_with_initial_timestamps(self, params, timestamps).await + }) + } + fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreFuture<'_, ()> { Box::pin(async move { live_writer::resume_thread(self, params).await }) } @@ -325,6 +336,8 @@ impl ThreadStore for LocalThreadStore { mod tests { use std::sync::Arc; + use chrono::DateTime; + use chrono::Utc; use codex_protocol::ThreadId; use codex_protocol::models::BaseInstructions; use codex_protocol::models::FunctionCallOutputPayload; @@ -342,7 +355,9 @@ mod tests { use tempfile::TempDir; use super::*; + use crate::InitialThreadTimestamps; use crate::LiveThread; + use crate::ThreadMetadataPatch; use crate::ThreadPersistenceMetadata; use crate::local::test_support::test_config; use crate::local::test_support::write_archived_session_file; @@ -398,6 +413,123 @@ mod tests { ); } + #[tokio::test] + async fn initial_timestamps_survive_read_repair() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = LocalThreadStore::new(config.clone(), Some(runtime.clone())); + let thread_id = ThreadId::new(); + let created_at = DateTime::parse_from_rfc3339("2024-01-02T03:04:05Z") + .expect("created timestamp") + .with_timezone(&Utc); + let updated_at = DateTime::parse_from_rfc3339("2024-02-03T04:05:06Z") + .expect("updated timestamp") + .with_timezone(&Utc); + + store + .create_thread_with_initial_timestamps( + create_thread_params(thread_id), + InitialThreadTimestamps { + created_at, + updated_at, + }, + ) + .await + .expect("create historical thread"); + store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![user_message_item("source history")], + }) + .await + .expect("append source history"); + store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + patch: ThreadMetadataPatch { + created_at: Some(created_at), + updated_at: Some(updated_at), + advance_recency_at: Some(updated_at), + ..Default::default() + }, + include_archived: false, + }) + .await + .expect("seed source chronology"); + let rollout_path = store + .live_rollout_path(thread_id) + .await + .expect("read rollout path"); + store + .persist_thread(thread_id) + .await + .expect("persist historical thread"); + store + .shutdown_thread(thread_id) + .await + .expect("shutdown historical thread"); + + let session_meta = codex_rollout::read_session_meta_line(&rollout_path) + .await + .expect("read session metadata"); + let rollout_modified_at = std::fs::metadata(&rollout_path) + .expect("rollout metadata") + .modified() + .map(DateTime::::from) + .expect("rollout modified timestamp"); + assert_eq!(session_meta.meta.timestamp, "2024-01-02T03:04:05.000Z"); + assert_eq!(rollout_modified_at, updated_at); + let local_day = created_at + .with_timezone(&chrono::Local) + .format("%Y/%m/%d") + .to_string(); + assert!(rollout_path.starts_with(home.path().join("sessions").join(local_day))); + + assert_eq!( + runtime + .delete_thread(thread_id) + .await + .expect("delete sqlite metadata"), + 1 + ); + drop(store); + drop(runtime); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should reopen"); + let store = LocalThreadStore::new(config, Some(runtime.clone())); + store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect("read and repair historical thread"); + let repaired = runtime + .get_thread(thread_id) + .await + .expect("read repaired sqlite metadata") + .expect("repaired sqlite metadata"); + assert_eq!( + ( + repaired.created_at, + repaired.updated_at, + repaired.recency_at, + ), + (created_at, updated_at, updated_at) + ); + } + #[tokio::test] async fn raw_append_items_does_not_update_sqlite_metadata() { // This pins the ThreadStore contract: raw appends are history-only. Callers that need diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index 67ee603f49..1c7819404d 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -8,6 +8,7 @@ use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::DeleteThreadParams; +use crate::InitialThreadTimestamps; use crate::ItemPage; use crate::ListItemsParams; use crate::ListThreadsParams; @@ -45,6 +46,16 @@ pub trait ThreadStore: Any + Send + Sync { /// Creates a new live thread. fn create_thread(&self, params: CreateThreadParams) -> ThreadStoreFuture<'_, ()>; + /// Creates a thread from pre-existing history using its original chronology. + /// + /// This is a creation-only initialization hook. Normal appends after creation must advance + /// update and recency timestamps through the usual live-thread path. + fn create_thread_with_initial_timestamps( + &self, + params: CreateThreadParams, + timestamps: InitialThreadTimestamps, + ) -> ThreadStoreFuture<'_, ()>; + /// Reopens an existing thread for live appends. fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreFuture<'_, ()>; diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 4e688eca6f..fa8cb11ef8 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -60,6 +60,15 @@ pub struct ThreadPersistenceMetadata { pub memory_mode: MemoryMode, } +/// Source chronology used only while creating a thread from pre-existing history. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct InitialThreadTimestamps { + /// Original creation time of the imported thread. + pub created_at: DateTime, + /// Original last-activity time of the imported thread. + pub updated_at: DateTime, +} + /// Extra configuration fields for a thread. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ExtraConfig {}