From dd975f36cd2c6add9a283e205a8e88d894c38d03 Mon Sep 17 00:00:00 2001 From: charlesgong-openai Date: Tue, 7 Jul 2026 18:04:56 -0400 Subject: [PATCH] Preserve imported session fidelity --- .../external_agent_session_import.rs | 74 ++++++++--- .../tests/suite/conversation_summary.rs | 1 + .../tests/suite/v2/external_agent_config.rs | 35 ++++- .../tests/suite/v2/remote_thread_store.rs | 1 + .../app-server/tests/suite/v2/thread_read.rs | 1 + .../tests/suite/v2/thread_unarchive.rs | 1 + codex-rs/core/src/session/session.rs | 2 + codex-rs/core/src/session/tests.rs | 4 + .../external-agent-sessions/src/detect.rs | 44 +++---- .../external-agent-sessions/src/export.rs | 123 +++++++++++++++++- .../external-agent-sessions/src/ledger.rs | 71 ++++++---- .../src/ledger_tests.rs | 79 ++++++++++- codex-rs/external-agent-sessions/src/lib.rs | 44 ++++++- .../external-agent-sessions/src/records.rs | 25 ++++ codex-rs/rollout/src/recorder.rs | 97 ++++++++++++-- codex-rs/rollout/src/recorder_tests.rs | 1 + codex-rs/state/src/runtime/threads.rs | 20 +++ codex-rs/thread-store/src/in_memory.rs | 9 +- codex-rs/thread-store/src/lib.rs | 1 + .../thread-store/src/local/create_thread.rs | 45 ++++--- codex-rs/thread-store/src/local/mod.rs | 4 + .../src/local/update_thread_metadata.rs | 32 ++++- .../thread-store/src/thread_metadata_sync.rs | 2 + codex-rs/thread-store/src/types.rs | 54 +++++++- 24 files changed, 656 insertions(+), 114 deletions(-) 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..c1d94ff690 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,8 @@ +use std::collections::HashSet; use std::path::PathBuf; use std::sync::Arc; +use chrono::DateTime; use chrono::Utc; use codex_arg0::Arg0DispatchPaths; use codex_core::ThreadManager; @@ -20,11 +22,13 @@ 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::HistoricalThreadTimestamps; use codex_thread_store::ThreadMetadataPatch; use codex_thread_store::ThreadPersistenceMetadata; use codex_thread_store::ThreadStore; use codex_thread_store::UpdateThreadMetadataParams; use futures::StreamExt; +use tokio::sync::Mutex; use tokio::sync::Semaphore; use crate::config::external_agent_config::ExternalAgentConfigImportItemResult; @@ -37,6 +41,7 @@ const SESSION_IMPORT_CONCURRENCY: usize = 5; pub(super) struct ExternalAgentSessionImporter { codex_home: PathBuf, permits: Arc, + reserved_source_identities: Arc>>, thread_manager: Arc, thread_store: Arc, config_manager: ConfigManager, @@ -54,6 +59,7 @@ impl ExternalAgentSessionImporter { Self { codex_home, permits: Arc::new(Semaphore::new(1)), + reserved_source_identities: Arc::new(Mutex::new(HashSet::new())), thread_manager, thread_store, config_manager, @@ -134,16 +140,38 @@ impl ExternalAgentSessionImporter { else { return Ok(None); }; - let imported_thread_id = - self.persist_session(pending_import.session) - .await - .map_err(|message| SessionImportFailure { - source_path: pending_import.source_path.clone(), + let source_identity = pending_import + .session + .source_session_id + .as_ref() + .map(|session_id| format!("session:{session_id}")) + .unwrap_or_else(|| format!("path:{}", pending_import.source_path.display())); + if !self + .reserved_source_identities + .lock() + .await + .insert(source_identity.clone()) + { + return Ok(None); + } + let source_session_id = pending_import.session.source_session_id.clone(); + let imported_thread_id = match self.persist_session(pending_import.session).await { + Ok(imported_thread_id) => imported_thread_id, + Err(message) => { + self.reserved_source_identities + .lock() + .await + .remove(&source_identity); + return Err(SessionImportFailure { + source_path: pending_import.source_path, message, stage: "session_persist", - })?; + }); + } + }; Ok(Some(CompletedExternalAgentSessionImport { source_path: pending_import.source_path, + source_session_id, source_content_sha256: pending_import.source_content_sha256, imported_thread_id, })) @@ -168,6 +196,9 @@ impl ExternalAgentSessionImporter { cwd, title, first_user_message, + source_session_id: _, + created_at, + updated_at, mut rollout_items, } = session; let config = self @@ -204,6 +235,12 @@ impl ExternalAgentSessionImporter { ThreadMemoryMode::Disabled }; let now = Utc::now(); + let created_at = created_at + .and_then(|timestamp| DateTime::::from_timestamp(timestamp, 0)) + .unwrap_or(now); + let updated_at = updated_at + .and_then(|timestamp| DateTime::::from_timestamp(timestamp, 0)) + .unwrap_or(created_at); let create_params = CreateThreadParams { session_id: thread_id.into(), thread_id, @@ -228,6 +265,10 @@ impl ExternalAgentSessionImporter { cwd: Some(cwd.clone()), model_provider: model_provider.clone(), memory_mode, + historical_timestamps: Some(HistoricalThreadTimestamps { + created_at, + updated_at, + }), }, }; rollout_items.retain(is_persisted_rollout_item); @@ -238,8 +279,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), + recency_at: Some(updated_at), source: Some(source.clone()), thread_source: Some(None), agent_nickname: Some(source.get_nickname()), @@ -269,14 +311,6 @@ impl ExternalAgentSessionImporter { return Err(format!("failed to import session: {err}")); } - self.thread_store - .update_thread_metadata(UpdateThreadMetadataParams { - thread_id, - patch: metadata, - include_archived: false, - }) - .await - .map_err(|err| format!("failed to update imported session: {err}"))?; self.thread_store .persist_thread(thread_id) .await @@ -285,6 +319,14 @@ impl ExternalAgentSessionImporter { .shutdown_thread(thread_id) .await .map_err(|err| format!("failed to shutdown imported session: {err}"))?; + self.thread_store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + patch: metadata, + include_archived: false, + }) + .await + .map_err(|err| format!("failed to update imported session: {err}"))?; Ok(thread_id) } } diff --git a/codex-rs/app-server/tests/suite/conversation_summary.rs b/codex-rs/app-server/tests/suite/conversation_summary.rs index 9cb1409d79..697c7a3ff2 100644 --- a/codex-rs/app-server/tests/suite/conversation_summary.rs +++ b/codex-rs/app-server/tests/suite/conversation_summary.rs @@ -143,6 +143,7 @@ async fn get_conversation_summary_by_thread_id_reads_pathless_store_thread() -> cwd: None, model_provider: "test-provider".to_string(), memory_mode: ThreadMemoryMode::Disabled, + historical_timestamps: None, }, }) .await?; 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 6bef968941..0b366e79f2 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 @@ -612,7 +612,8 @@ 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 = "2026-06-03T12:00:00Z"; + let source_updated_at = "2026-06-03T12:00:05Z"; let session_dir = external_agent_home(codex_home.path()).join("projects/repo"); let session_path = session_dir.join("session.jsonl"); std::fs::create_dir_all(&project_root)?; @@ -623,14 +624,14 @@ 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, "message": { "content": "first request" }, }) .to_string(), serde_json::json!({ "type": "assistant", "cwd": &project_root, - "timestamp": &recent_timestamp, + "timestamp": source_updated_at, "message": { "content": "first answer" }, }) .to_string(), @@ -745,6 +746,22 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { assert_eq!(imported_thread_id, thread.id.to_string()); assert_eq!(thread.preview, "first request"); assert_eq!(thread.name.as_deref(), Some("source session title")); + let source_created_at = chrono::DateTime::parse_from_rfc3339(source_created_at)?.timestamp(); + let source_updated_at = chrono::DateTime::parse_from_rfc3339(source_updated_at)?.timestamp(); + assert_eq!( + ( + thread.created_at, + thread.updated_at, + thread.recency_at, + thread.cwd.as_path(), + ), + ( + source_created_at, + source_updated_at, + Some(source_updated_at), + project_root.as_path(), + ) + ); let request_id = mcp .send_thread_read_request(ThreadReadParams { @@ -759,6 +776,18 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { .await??; let response: ThreadReadResponse = to_response(response)?; assert_eq!(response.thread.turns.len(), 1); + assert_eq!( + ( + response.thread.turns[0].started_at, + response.thread.turns[0].completed_at, + response.thread.turns[0].duration_ms, + ), + ( + Some(source_created_at), + Some(source_updated_at), + Some(5_000), + ) + ); let items = &response.thread.turns[0].items; assert_eq!(items.len(), 3); assert_eq!( diff --git a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs index 2002b08a88..86391bde9b 100644 --- a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs +++ b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs @@ -166,6 +166,7 @@ async fn thread_delete_with_non_local_thread_store_does_not_create_local_persist cwd: Some(codex_home.path().to_path_buf()), model_provider: "mock_provider".to_string(), memory_mode: ThreadMemoryMode::Enabled, + historical_timestamps: None, }, }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index 4162e7ebd6..f99e9123eb 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -1549,6 +1549,7 @@ async fn seed_pathless_store_thread( cwd: None, model_provider: "test-provider".to_string(), memory_mode: ThreadMemoryMode::Disabled, + historical_timestamps: None, }, }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs index 66d51d2780..e8bd7606e2 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs @@ -229,6 +229,7 @@ async fn thread_unarchive_preserves_pathless_store_metadata() -> Result<()> { cwd: None, model_provider: "test-provider".to_string(), memory_mode: ThreadMemoryMode::Disabled, + historical_timestamps: None, }, }) .await?; diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 08abb46377..e12e1d85b5 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -614,6 +614,7 @@ impl Session { } else { ThreadMemoryMode::Disabled }, + historical_timestamps: None, }, }; LiveThread::create(Arc::clone(&thread_store), params).await? @@ -632,6 +633,7 @@ impl Session { } else { ThreadMemoryMode::Disabled }, + historical_timestamps: None, }, }; LiveThread::resume(Arc::clone(&thread_store), params).await? diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index ac8f9d049e..92bc1a837e 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -4082,6 +4082,7 @@ async fn attach_thread_persistence(session: &mut Session) -> PathBuf { } else { ThreadMemoryMode::Disabled }, + historical_timestamps: None, }, }, ) @@ -6974,6 +6975,7 @@ async fn shutdown_complete_does_not_append_to_thread_store_after_shutdown() { } else { ThreadMemoryMode::Disabled }, + historical_timestamps: None, }, }, ) @@ -7051,6 +7053,7 @@ async fn submission_loop_channel_close_runs_full_thread_teardown() { } else { ThreadMemoryMode::Disabled }, + historical_timestamps: None, }, }, ) @@ -9360,6 +9363,7 @@ async fn attach_in_memory_thread_store( } else { ThreadMemoryMode::Disabled }, + historical_timestamps: None, }, }, ) diff --git a/codex-rs/external-agent-sessions/src/detect.rs b/codex-rs/external-agent-sessions/src/detect.rs index f42c90766b..78e895e713 100644 --- a/codex-rs/external-agent-sessions/src/detect.rs +++ b/codex-rs/external-agent-sessions/src/detect.rs @@ -84,7 +84,14 @@ pub fn detect_recent_sessions( let mut migrations = Vec::new(); let mut ledger_changed = false; for (modified_at, path) in file_candidates { - match ledger.refresh_current_source(&path, modified_at.0) { + let Ok(Some(summary)) = summarize_session(&path) else { + continue; + }; + match ledger.refresh_current_source( + &path, + summary.source_session_id.as_deref(), + modified_at.0, + ) { Ok(false) => {} Ok(true) => { ledger_changed = true; @@ -92,9 +99,6 @@ pub fn detect_recent_sessions( } Err(_) => continue, } - let Ok(Some(summary)) = summarize_session(&path) else { - continue; - }; let migration = summary.migration; if !migration.cwd.is_dir() { continue; @@ -323,7 +327,7 @@ mod tests { let sessions = detect_recent_sessions(&external_agent_home, root.path()).expect("detect"); - assert_eq!(sessions, vec![oldest_session.clone()]); + assert_eq!(sessions, vec![oldest_session]); for session in sessions { record_imported_session(root.path(), &session.path, ThreadId::new()) .expect("record import"); @@ -347,17 +351,11 @@ mod tests { ); } - let sessions = detect_recent_sessions(&external_agent_home, root.path()).expect("detect"); - - assert_eq!(sessions, expected); - for session in sessions { - record_imported_session(root.path(), &session.path, ThreadId::new()) - .expect("record import"); - } - - let sessions = detect_recent_sessions(&external_agent_home, root.path()).expect("detect"); - - assert_eq!(sessions, vec![oldest_session]); + assert!( + detect_recent_sessions(&external_agent_home, root.path()) + .expect("detect") + .is_empty() + ); } #[test] @@ -383,7 +381,7 @@ mod tests { } #[test] - fn redetects_sessions_when_source_contents_change_after_import() { + fn skips_imported_sessions_when_source_contents_change() { let root = TempDir::new().expect("tempdir"); let external_agent_home = root.path().join(".external"); let project_root = root.path().join("repo"); @@ -405,14 +403,10 @@ mod tests { ) .expect("update session"); - let sessions = detect_recent_sessions(&external_agent_home, root.path()).expect("detect"); - assert_eq!( - sessions, - vec![ExternalAgentSessionMigration { - path: session_path, - cwd: project_root, - title: Some("hello there".to_string()), - }] + assert!( + detect_recent_sessions(&external_agent_home, root.path()) + .expect("detect") + .is_empty() ); } diff --git a/codex-rs/external-agent-sessions/src/export.rs b/codex-rs/external-agent-sessions/src/export.rs index 6391d27bd3..82cdefe831 100644 --- a/codex-rs/external-agent-sessions/src/export.rs +++ b/codex-rs/external-agent-sessions/src/export.rs @@ -41,6 +41,14 @@ pub(crate) fn load_session_for_import_with_content_sha256( .find(|message| message.role == MessageRole::User) .map(|message| summarize_for_label(&message.text)); let title = parsed.source_title.or_else(|| first_user_message.clone()); + let created_at = messages + .iter() + .filter_map(|message| message.timestamp) + .min(); + let updated_at = messages + .iter() + .filter_map(|message| message.timestamp) + .max(); let rollout_items = rollout_items_from_messages(messages); if rollout_items.is_empty() { return Ok(None); @@ -50,6 +58,9 @@ pub(crate) fn load_session_for_import_with_content_sha256( cwd, title, first_user_message, + source_session_id: parsed.source_session_id, + created_at, + updated_at, rollout_items, }, parsed.content_sha256, @@ -62,16 +73,23 @@ fn rollout_items_from_messages(messages: Vec) -> Vec { if let Some(turn_id) = current_turn.take() { - items.push(turn_complete_item(turn_id, /*completed_at*/ None)); + items.push(turn_complete_item( + turn_id, + current_turn_started_at, + current_turn_completed_at, + )); } user_turn_count += 1; let turn_id = format!("external-import-turn-{user_turn_count}"); + current_turn_started_at = message.timestamp; + current_turn_completed_at = message.timestamp; items.push(RolloutItem::EventMsg(EventMsg::TurnStarted( TurnStartedEvent { turn_id: turn_id.clone(), @@ -96,6 +114,12 @@ fn rollout_items_from_messages(messages: Vec) -> Vec) -> Vec RolloutItem { })) } -fn turn_complete_item(turn_id: String, completed_at: Option) -> RolloutItem { +fn turn_complete_item( + turn_id: String, + started_at: Option, + completed_at: Option, +) -> RolloutItem { + let duration_ms = started_at + .zip(completed_at) + .and_then(|(started_at, completed_at)| completed_at.checked_sub(started_at)) + .filter(|duration| *duration >= 0) + .and_then(|duration| duration.checked_mul(1_000)); RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent { turn_id, last_agent_message: None, completed_at, - duration_ms: None, + duration_ms, time_to_first_token_ms: None, })) } @@ -218,6 +255,78 @@ mod tests { ); } + #[test] + fn preserves_source_session_and_turn_timing() { + 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"); + let first_started_at = "2026-06-03T12:00:00Z"; + let first_completed_at = "2026-06-03T12:00:05Z"; + let second_started_at = "2026-06-03T12:01:00Z"; + let second_completed_at = "2026-06-03T12:01:09Z"; + std::fs::write( + &path, + jsonl(&[ + record_at("user", "first request", &project_root, first_started_at), + record_at( + "assistant", + "first answer", + &project_root, + first_completed_at, + ), + record_at("user", "second request", &project_root, second_started_at), + record_at( + "assistant", + "second answer", + &project_root, + second_completed_at, + ), + ]), + ) + .expect("session"); + + let imported = load_session_for_import(&path) + .expect("load") + .expect("session"); + let turns = build_turns_from_rollout_items(&imported.rollout_items); + let first_started_at = chrono::DateTime::parse_from_rfc3339(first_started_at) + .expect("first started at") + .timestamp(); + let first_completed_at = chrono::DateTime::parse_from_rfc3339(first_completed_at) + .expect("first completed at") + .timestamp(); + let second_started_at = chrono::DateTime::parse_from_rfc3339(second_started_at) + .expect("second started at") + .timestamp(); + let second_completed_at = chrono::DateTime::parse_from_rfc3339(second_completed_at) + .expect("second completed at") + .timestamp(); + + assert_eq!( + (imported.created_at, imported.updated_at), + (Some(first_started_at), Some(second_completed_at)) + ); + assert_eq!( + turns + .iter() + .map(|turn| (turn.started_at, turn.completed_at, turn.duration_ms)) + .collect::>(), + vec![ + ( + Some(first_started_at), + Some(first_completed_at), + Some(5_000) + ), + ( + Some(second_started_at), + Some(second_completed_at), + Some(9_000), + ), + ] + ); + } + #[test] fn adds_import_marker_without_copying_last_agent_message() { let root = TempDir::new().expect("tempdir"); @@ -407,6 +516,10 @@ mod tests { 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, ×tamp) + } + + fn record_at(role: &str, text: &str, cwd: &Path, timestamp: &str) -> JsonValue { serde_json::json!({ "type": role, "cwd": cwd, diff --git a/codex-rs/external-agent-sessions/src/ledger.rs b/codex-rs/external-agent-sessions/src/ledger.rs index 9a3b2042b8..b5f8fd3da9 100644 --- a/codex-rs/external-agent-sessions/src/ledger.rs +++ b/codex-rs/external-agent-sessions/src/ledger.rs @@ -23,6 +23,8 @@ pub(super) struct ImportedExternalAgentSessionLedger { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] struct ImportedExternalAgentSessionRecord { source_path: PathBuf, + #[serde(default)] + source_session_id: Option, content_sha256: String, imported_thread_id: ThreadId, imported_at: i64, @@ -33,6 +35,7 @@ struct ImportedExternalAgentSessionRecord { #[derive(Debug, PartialEq, Eq)] pub struct CompletedExternalAgentSessionImport { pub source_path: PathBuf, + pub source_session_id: Option, pub source_content_sha256: String, pub imported_thread_id: ThreadId, } @@ -46,8 +49,9 @@ pub(super) struct ImportedSourceState { pub fn has_current_session_been_imported( codex_home: &Path, source_path: &Path, + source_session_id: Option<&str>, ) -> io::Result { - load_import_ledger(codex_home)?.contains_current_source(source_path) + load_import_ledger(codex_home)?.contains_source_identity(source_path, source_session_id) } #[cfg(test)] @@ -62,6 +66,7 @@ pub(crate) fn record_imported_session( vec![CompletedExternalAgentSessionImport { source_content_sha256: session_content_sha256(&source_path)?, source_path, + source_session_id: None, imported_thread_id, }], ) @@ -79,10 +84,18 @@ pub fn record_completed_session_imports( for import in imports { let source_modified_at = session_modified_at(&import.source_path).ok().flatten(); if let Some(index) = ledger.records.iter().rposition(|record| { - record.source_path == import.source_path - && record.content_sha256 == import.source_content_sha256 + record_matches_source_identity( + record, + &import.source_path, + import.source_session_id.as_deref(), + ) }) { let mut record = ledger.records.remove(index); + record.source_path = import.source_path; + if import.source_session_id.is_some() { + record.source_session_id = import.source_session_id; + } + record.content_sha256 = import.source_content_sha256; record.imported_thread_id = import.imported_thread_id; record.imported_at = imported_at; record.source_modified_at = source_modified_at.or(record.source_modified_at); @@ -91,6 +104,7 @@ pub fn record_completed_session_imports( } ledger.records.push(ImportedExternalAgentSessionRecord { source_path: import.source_path, + source_session_id: import.source_session_id, content_sha256: import.source_content_sha256, imported_thread_id: import.imported_thread_id, imported_at, @@ -115,44 +129,40 @@ impl ImportedExternalAgentSessionLedger { states } - pub(super) fn contains_current_source(&self, source_path: &Path) -> io::Result { + pub(super) fn contains_source_identity( + &self, + source_path: &Path, + source_session_id: Option<&str>, + ) -> io::Result { if self.records.is_empty() { return Ok(false); } let source_path = canonical_source_path(source_path)?; - if !self + Ok(self .records .iter() - .any(|record| record.source_path == source_path) - { - return Ok(false); - } - let content_sha256 = session_content_sha256(&source_path)?; - Ok(self.records.iter().any(|record| { - record.source_path == source_path && record.content_sha256 == content_sha256 - })) + .any(|record| record_matches_source_identity(record, &source_path, source_session_id))) } pub(super) fn refresh_current_source( &mut self, source_path: &Path, + source_session_id: Option<&str>, source_modified_at: i64, ) -> io::Result { let source_path = canonical_source_path(source_path)?; - if !self - .records - .iter() - .any(|record| record.source_path == source_path) - { - return Ok(false); - } - let content_sha256 = session_content_sha256(&source_path)?; let Some(index) = self.records.iter().rposition(|record| { - record.source_path == source_path && record.content_sha256 == content_sha256 + record_matches_source_identity(record, &source_path, source_session_id) }) else { return Ok(false); }; + let content_sha256 = session_content_sha256(&source_path)?; let mut record = self.records.remove(index); + record.source_path = source_path; + if let Some(source_session_id) = source_session_id { + record.source_session_id = Some(source_session_id.to_string()); + } + record.content_sha256 = content_sha256; record.imported_at = now_unix_seconds(); record.source_modified_at = Some(source_modified_at); self.records.push(record); @@ -160,6 +170,23 @@ impl ImportedExternalAgentSessionLedger { } } +fn record_matches_source_identity( + record: &ImportedExternalAgentSessionRecord, + source_path: &Path, + source_session_id: Option<&str>, +) -> bool { + record.source_path == source_path + || source_session_id.is_some_and(|source_session_id| { + record.source_session_id.as_deref() == Some(source_session_id) + || record.source_session_id.is_none() + && record + .source_path + .file_stem() + .and_then(|stem| stem.to_str()) + == Some(source_session_id) + }) +} + pub(super) fn load_import_ledger( codex_home: &Path, ) -> io::Result { diff --git a/codex-rs/external-agent-sessions/src/ledger_tests.rs b/codex-rs/external-agent-sessions/src/ledger_tests.rs index f4b9dd6f7c..6bf34391cc 100644 --- a/codex-rs/external-agent-sessions/src/ledger_tests.rs +++ b/codex-rs/external-agent-sessions/src/ledger_tests.rs @@ -13,7 +13,7 @@ fn empty_ledger_does_not_read_source() { assert!( !ImportedExternalAgentSessionLedger::default() - .contains_current_source(&missing_source) + .contains_source_identity(&missing_source, None) .expect("empty ledger cannot contain sources") ); } @@ -33,6 +33,7 @@ fn completed_imports_do_not_read_source_files() { &codex_home, vec![CompletedExternalAgentSessionImport { source_path: source_path.clone(), + source_session_id: None, source_content_sha256: format!("{:x}", Sha256::digest(contents)), imported_thread_id, }], @@ -62,6 +63,7 @@ fn completed_import_refreshes_existing_record_metadata() { &codex_home, vec![CompletedExternalAgentSessionImport { source_path: source_path.clone(), + source_session_id: None, source_content_sha256: content_sha256.clone(), imported_thread_id: first_thread_id, }], @@ -71,6 +73,7 @@ fn completed_import_refreshes_existing_record_metadata() { &codex_home, vec![CompletedExternalAgentSessionImport { source_path: source_path.clone(), + source_session_id: None, source_content_sha256: content_sha256, imported_thread_id: second_thread_id, }], @@ -83,3 +86,77 @@ fn completed_import_refreshes_existing_record_metadata() { assert_eq!(ledger.records[0].imported_thread_id, second_thread_id); assert!(ledger.records[0].source_modified_at.is_some()); } + +#[test] +fn stable_session_id_deduplicates_moved_and_changed_sources() { + let root = TempDir::new().expect("tempdir"); + let codex_home = root.path().join("codex-home"); + let first_path = root.path().join("first-session.jsonl"); + let second_path = root.path().join("moved-session.jsonl"); + std::fs::write(&first_path, "first contents").expect("first source"); + std::fs::write(&second_path, "updated contents").expect("moved source"); + let first_path = std::fs::canonicalize(first_path).expect("canonical first source"); + let second_path = std::fs::canonicalize(second_path).expect("canonical moved source"); + let source_session_id = "source-session-id"; + let first_thread_id = ThreadId::new(); + let second_thread_id = ThreadId::new(); + + record_completed_session_imports( + &codex_home, + vec![CompletedExternalAgentSessionImport { + source_path: first_path, + source_session_id: Some(source_session_id.to_string()), + source_content_sha256: format!("{:x}", Sha256::digest(b"first contents")), + imported_thread_id: first_thread_id, + }], + ) + .expect("record first import"); + + let ledger = super::load_import_ledger(&codex_home).expect("ledger"); + assert!( + ledger + .contains_source_identity(&second_path, Some(source_session_id)) + .expect("match moved source") + ); + + record_completed_session_imports( + &codex_home, + vec![CompletedExternalAgentSessionImport { + source_path: second_path.clone(), + source_session_id: Some(source_session_id.to_string()), + source_content_sha256: format!("{:x}", Sha256::digest(b"updated contents")), + imported_thread_id: second_thread_id, + }], + ) + .expect("record moved import"); + + let ledger = super::load_import_ledger(&codex_home).expect("updated ledger"); + assert_eq!(ledger.records.len(), 1); + assert_eq!(ledger.records[0].source_path, second_path); + assert_eq!( + ledger.records[0].source_session_id.as_deref(), + Some(source_session_id) + ); + assert_eq!(ledger.records[0].imported_thread_id, second_thread_id); +} + +#[test] +fn legacy_ledger_uses_source_filename_as_session_id() { + let root = TempDir::new().expect("tempdir"); + let codex_home = root.path().join("codex-home"); + let source_session_id = "source-session-id"; + let first_path = root.path().join(format!("{source_session_id}.jsonl")); + let moved_path = root.path().join("moved-session.jsonl"); + std::fs::write(&first_path, "first contents").expect("first source"); + std::fs::write(&moved_path, "first contents").expect("moved source"); + super::record_imported_session(&codex_home, &first_path, ThreadId::new()) + .expect("record legacy import"); + let moved_path = std::fs::canonicalize(moved_path).expect("canonical moved source"); + + let ledger = super::load_import_ledger(&codex_home).expect("ledger"); + assert!( + ledger + .contains_source_identity(&moved_path, Some(source_session_id)) + .expect("match legacy source") + ); +} diff --git a/codex-rs/external-agent-sessions/src/lib.rs b/codex-rs/external-agent-sessions/src/lib.rs index 0b7a4eb2ba..781edfe207 100644 --- a/codex-rs/external-agent-sessions/src/lib.rs +++ b/codex-rs/external-agent-sessions/src/lib.rs @@ -32,6 +32,9 @@ pub struct ImportedExternalAgentSession { pub cwd: PathBuf, pub title: Option, pub first_user_message: Option, + pub source_session_id: Option, + pub created_at: Option, + pub updated_at: Option, pub rollout_items: Vec, } @@ -46,15 +49,19 @@ pub fn prepare_validated_session_import( codex_home: &Path, session: ExternalAgentSessionMigration, ) -> io::Result> { - let has_been_imported = has_current_session_been_imported(codex_home, &session.path)?; - if has_been_imported { - return Ok(None); - } let Some((source_path, imported_session, source_content_sha256)) = load_importable_session(&session.path)? else { return Ok(None); }; + let has_been_imported = has_current_session_been_imported( + codex_home, + &source_path, + imported_session.source_session_id.as_deref(), + )?; + if has_been_imported { + return Ok(None); + } Ok(Some(PendingSessionImport { source_path, source_content_sha256, @@ -127,9 +134,31 @@ mod tests { let root = TempDir::new().expect("tempdir"); let codex_home = root.path().join("codex-home"); let source_path = root.path().join("session.jsonl"); - std::fs::write(&source_path, "{}\n").expect("session"); + let source_record = serde_json::json!({ + "type": "user", + "cwd": root.path(), + "sessionId": "source-session-id", + "timestamp": "2026-06-03T12:00:00Z", + "message": { "content": "first request" }, + }); + std::fs::write(&source_path, source_record.to_string()).expect("session"); ledger::record_imported_session(&codex_home, &source_path, ThreadId::new()) .expect("record import"); + std::fs::write( + &source_path, + format!( + "{}\n{}", + source_record, + serde_json::json!({ + "type": "assistant", + "cwd": root.path(), + "sessionId": "source-session-id", + "timestamp": "2026-06-03T12:00:05Z", + "message": { "content": "first answer" }, + }) + ), + ) + .expect("updated session"); let pending = prepare_validated_session_import(&codex_home, session_migration(&source_path)) @@ -156,6 +185,7 @@ mod tests { let contents = serde_json::json!({ "type": "user", "cwd": root.path(), + "sessionId": "source-session-id", "timestamp": "2026-06-03T12:00:00Z", "message": { "content": "first request" }, }) @@ -171,6 +201,10 @@ mod tests { pending.source_content_sha256, format!("{:x}", Sha256::digest(contents)) ); + assert_eq!( + pending.session.source_session_id.as_deref(), + Some("source-session-id") + ); } fn session_migration(path: &Path) -> ExternalAgentSessionMigration { diff --git a/codex-rs/external-agent-sessions/src/records.rs b/codex-rs/external-agent-sessions/src/records.rs index 00307fa1d5..9b9c9d21c7 100644 --- a/codex-rs/external-agent-sessions/src/records.rs +++ b/codex-rs/external-agent-sessions/src/records.rs @@ -20,12 +20,14 @@ const EXTERNAL_AGENT_TOOL_RESULT_TAG: &str = "external_agent_tool_result"; pub struct SessionSummary { pub latest_timestamp: i64, + pub source_session_id: Option, pub migration: ExternalAgentSessionMigration, } pub(super) struct ParsedSessionImport { pub cwd: Option, pub source_title: Option, + pub source_session_id: Option, pub messages: Vec, pub content_sha256: String, } @@ -37,6 +39,7 @@ pub fn summarize_session(path: &Path) -> io::Result> { let mut custom_title = None; let mut ai_title = None; let mut title = None; + let mut source_session_id = None; let mut latest_timestamp = None; let mut saw_message = false; @@ -49,6 +52,9 @@ pub fn summarize_session(path: &Path) -> io::Result> { let Ok(mut record) = serde_json::from_str::(trimmed) else { continue; }; + if source_session_id.is_none() { + source_session_id = source_session_id_from_record(&record).map(str::to_string); + } if cwd.is_none() { cwd = record .get("cwd") @@ -85,6 +91,7 @@ pub fn summarize_session(path: &Path) -> io::Result> { }; Ok(Some(SessionSummary { latest_timestamp, + source_session_id, migration: ExternalAgentSessionMigration { path: path.to_path_buf(), cwd, @@ -99,6 +106,7 @@ pub(super) fn read_session_import(path: &Path) -> io::Result io::Result(trimmed) else { continue; }; + if source_session_id.is_none() { + source_session_id = source_session_id_from_record(&record).map(str::to_string); + } if cwd.is_none() { cwd = record .get("cwd") @@ -134,6 +145,7 @@ pub(super) fn read_session_import(path: &Path) -> io::Result Option<&str> { title_from_record(record, "ai-title", "aiTitle") } +fn source_session_id_from_record(record: &JsonValue) -> Option<&str> { + record + .get("sessionId") + .and_then(JsonValue::as_str) + .map(str::trim) + .filter(|session_id| !session_id.is_empty()) +} + fn title_from_record<'a>(record: &'a JsonValue, record_type: &str, field: &str) -> Option<&'a str> { (record.get("type").and_then(JsonValue::as_str) == Some(record_type)) .then(|| record.get(field).and_then(JsonValue::as_str)) @@ -351,6 +371,7 @@ mod tests { serde_json::json!({ "type": "user", "cwd": root.path(), + "sessionId": "source-session-id", "timestamp": "2026-06-03T12:00:00Z", "message": { "content": "first request" }, }) @@ -374,6 +395,10 @@ mod tests { assert_eq!(parsed.cwd.as_deref(), Some(root.path())); assert_eq!(parsed.source_title.as_deref(), Some("custom title")); + assert_eq!( + parsed.source_session_id.as_deref(), + Some("source-session-id") + ); assert_eq!(parsed.messages.len(), 1); assert_eq!(parsed.messages[0].text, "first request"); assert_eq!( diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 87e97997a2..529ccfbf76 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -9,7 +9,9 @@ use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; +use chrono::DateTime; use chrono::SecondsFormat; +use chrono::Utc; use codex_protocol::SessionId; use codex_protocol::ThreadId; use codex_protocol::capabilities::SelectedCapabilityRoot; @@ -96,13 +98,21 @@ pub enum RolloutRecorderParams { selected_capability_roots: Vec, multi_agent_version: Option, history_mode: ThreadHistoryMode, - initial_window_id: Option, + initial_window_id: Box>, + historical_timestamps: Option>, }, Resume { path: PathBuf, }, } +#[doc(hidden)] +#[derive(Clone, Default)] +pub struct HistoricalRolloutTimestamps { + created_at: Option>, + updated_at: Option>, +} + enum RolloutCmd { AddItems(Vec), Persist { @@ -189,7 +199,8 @@ impl RolloutRecorderParams { selected_capability_roots: Vec::new(), multi_agent_version: None, history_mode: Default::default(), - initial_window_id: None, + initial_window_id: Box::default(), + historical_timestamps: None, } } @@ -244,7 +255,33 @@ impl RolloutRecorderParams { .. } = &mut self { - *window_id = Some(initial_window_id); + **window_id = Some(initial_window_id); + } + self + } + + pub fn with_created_at(mut self, created_at: DateTime) -> Self { + if let Self::Create { + historical_timestamps, + .. + } = &mut self + { + historical_timestamps + .get_or_insert_with(Default::default) + .created_at = Some(created_at); + } + self + } + + pub fn with_updated_at(mut self, updated_at: DateTime) -> Self { + if let Self::Create { + historical_timestamps, + .. + } = &mut self + { + historical_timestamps + .get_or_insert_with(Default::default) + .updated_at = Some(updated_at); } self } @@ -751,7 +788,7 @@ impl RolloutRecorder { config: &impl RolloutConfigView, params: RolloutRecorderParams, ) -> std::io::Result { - let (file, deferred_log_file_info, rollout_path, meta) = match params { + let (file, deferred_log_file_info, rollout_path, meta, updated_at) = match params { RolloutRecorderParams::Create { session_id, conversation_id, @@ -766,8 +803,16 @@ impl RolloutRecorder { multi_agent_version, history_mode, initial_window_id, + historical_timestamps, } => { - let log_file_info = precompute_log_file_info(config, conversation_id)?; + let historical_timestamps = historical_timestamps + .map(|timestamps| *timestamps) + .unwrap_or_default(); + let log_file_info = precompute_log_file_info( + config, + conversation_id, + historical_timestamps.created_at, + )?; let path = log_file_info.path.clone(); let thread_id = log_file_info.conversation_id; let started_at = log_file_info.timestamp; @@ -805,10 +850,16 @@ impl RolloutRecorder { memory_mode: (!config.generate_memories()).then_some("disabled".to_string()), history_mode, multi_agent_version, - context_window: initial_window_id.map(SessionContextWindow::new), + context_window: (*initial_window_id).map(SessionContextWindow::new), }; - (None, Some(log_file_info), path, Some(session_meta)) + ( + None, + Some(log_file_info), + path, + Some(session_meta), + historical_timestamps.updated_at, + ) } RolloutRecorderParams::Resume { path } => { let path = compression::materialize_rollout_for_append(path.as_path()).await?; @@ -822,6 +873,7 @@ impl RolloutRecorder { None, path, None, + None, ) } }; @@ -847,6 +899,7 @@ impl RolloutRecorder { meta, cwd, rollout_path_for_spawn.clone(), + updated_at, ) .await; if let Err(err) = result { @@ -1498,10 +1551,14 @@ struct LogFileInfo { fn precompute_log_file_info( config: &impl RolloutConfigView, conversation_id: ThreadId, + created_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 created_at { + Some(created_at) => OffsetDateTime::from(std::time::SystemTime::from(created_at)), + 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()); @@ -1554,6 +1611,7 @@ struct RolloutWriterState { meta: Option, cwd: PathBuf, rollout_path: PathBuf, + updated_at: Option>, last_logged_error: Option, } @@ -1564,6 +1622,7 @@ impl RolloutWriterState { meta: Option, cwd: PathBuf, rollout_path: PathBuf, + updated_at: Option>, ) -> Self { Self { writer: file.map(|file| JsonlWriter { file }), @@ -1572,6 +1631,7 @@ impl RolloutWriterState { meta, cwd, rollout_path, + updated_at, last_logged_error: None, } } @@ -1604,7 +1664,14 @@ impl RolloutWriterState { if self.is_deferred() && self.pending_items.is_empty() { return Ok(()); } - self.write_pending_with_recovery("shutdown").await + self.write_pending_with_recovery("shutdown").await?; + if let Some(updated_at) = self.updated_at { + std::fs::OpenOptions::new() + .write(true) + .open(self.rollout_path.as_path())? + .set_times(std::fs::FileTimes::new().set_modified(updated_at.into()))?; + } + Ok(()) } async fn write_pending_with_recovery(&mut self, operation: &str) -> std::io::Result<()> { @@ -1722,8 +1789,16 @@ async fn rollout_writer( meta: Option, cwd: PathBuf, rollout_path: PathBuf, + updated_at: Option>, ) -> std::io::Result<()> { - let mut state = RolloutWriterState::new(file, deferred_log_file_info, meta, cwd, rollout_path); + let mut state = RolloutWriterState::new( + file, + deferred_log_file_info, + meta, + cwd, + rollout_path, + updated_at, + ); // Process rollout commands while let Some(cmd) = rx.recv().await { diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index 7fcaff08ab..4586af894d 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -574,6 +574,7 @@ async fn writer_state_retries_write_error_before_reporting_flush_success() -> st /*meta*/ None, home.path().to_path_buf(), rollout_path.clone(), + /*updated_at*/ None, ); state.add_items(vec![RolloutItem::EventMsg(EventMsg::AgentMessage( AgentMessageEvent { diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index c80b2b6716..f9899a3339 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -687,6 +687,26 @@ WHERE id = ? Ok(result.rows_affected() > 0) } + /// Replace product recency exactly for historical import and repair paths. + pub async fn set_thread_recency_at( + &self, + thread_id: ThreadId, + recency_at: DateTime, + ) -> anyhow::Result { + let recency_at_seconds = datetime_to_epoch_seconds(recency_at); + let recency_at_millis = datetime_to_epoch_millis(recency_at); + self.thread_recency_at_millis + .fetch_max(recency_at_millis, Ordering::Relaxed); + let result = + sqlx::query("UPDATE threads SET recency_at = ?, recency_at_ms = ? WHERE id = ?") + .bind(recency_at_seconds) + .bind(recency_at_millis) + .bind(thread_id.to_string()) + .execute(self.pool.as_ref()) + .await?; + Ok(result.rows_affected() > 0) + } + /// Allocate a persisted `updated_at` value for thread-list cursor ordering. /// /// We keep a process-local high-water mark so hot rollout writes can get unique, diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index cb3f198ff3..e08664a02c 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -137,6 +137,7 @@ mod tests { cwd: None, model_provider: "test-provider".to_string(), memory_mode: ThreadMemoryMode::Enabled, + historical_timestamps: None, }, }) .await @@ -337,6 +338,7 @@ mod tests { cwd: None, model_provider: "test-provider".to_string(), memory_mode: ThreadMemoryMode::Enabled, + historical_timestamps: None, } } @@ -777,7 +779,12 @@ fn stored_thread_from_state( .and_then(|metadata| metadata.updated_at) .unwrap_or_else(Utc::now), recency_at: metadata - .and_then(|metadata| metadata.advance_recency_at.or(metadata.updated_at)) + .and_then(|metadata| { + metadata + .recency_at + .or(metadata.advance_recency_at) + .or(metadata.updated_at) + }) .unwrap_or_else(Utc::now), archived_at: None, cwd: metadata diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 34a0ab2ba1..38fffd261c 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::HistoricalThreadTimestamps; 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..d058ba629f 100644 --- a/codex-rs/thread-store/src/local/create_thread.rs +++ b/codex-rs/thread-store/src/local/create_thread.rs @@ -27,26 +27,29 @@ 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 mut 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); + if let Some(timestamps) = params.metadata.historical_timestamps { + recorder_params = recorder_params + .with_created_at(timestamps.created_at) + .with_updated_at(timestamps.updated_at); + } + 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/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 5939ea606e..b8e0b1a592 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -703,6 +703,7 @@ mod tests { cwd: Some(home.path().to_path_buf()), model_provider: "different-provider".to_string(), memory_mode: ThreadMemoryMode::Enabled, + historical_timestamps: None, }, }, ) @@ -757,6 +758,7 @@ mod tests { cwd: Some(home.path().to_path_buf()), model_provider: "different-provider".to_string(), memory_mode: ThreadMemoryMode::Enabled, + historical_timestamps: None, }, }, ) @@ -968,6 +970,7 @@ mod tests { cwd: None, model_provider: "test-provider".to_string(), memory_mode: ThreadMemoryMode::Enabled, + historical_timestamps: None, }, }) .await @@ -1300,6 +1303,7 @@ mod tests { cwd: Some(std::env::current_dir().expect("cwd")), model_provider: "test-provider".to_string(), memory_mode: ThreadMemoryMode::Enabled, + historical_timestamps: None, } } diff --git a/codex-rs/thread-store/src/local/update_thread_metadata.rs b/codex-rs/thread-store/src/local/update_thread_metadata.rs index 580df1dc61..82fabb0def 100644 --- a/codex-rs/thread-store/src/local/update_thread_metadata.rs +++ b/codex-rs/thread-store/src/local/update_thread_metadata.rs @@ -221,6 +221,7 @@ async fn apply_metadata_update( message: format!("failed to read thread metadata for {thread_id}: {err}"), })?; let advance_recency_at = patch.advance_recency_at; + let recency_at = patch.recency_at; if existing.is_none() && rollout_path.is_none() { let resolved = resolve_rollout_path(store, thread_id, include_archived).await?; rollout_path_archived = resolved.archived; @@ -274,7 +275,11 @@ async fn apply_metadata_update( if let Some(updated_at) = patch.updated_at { metadata.updated_at = updated_at; } - if existing.is_none() + if let Some(recency_at) = recency_at { + metadata.recency_at = recency_at; + } + if recency_at.is_none() + && existing.is_none() && let Some(recency_at) = advance_recency_at { metadata.recency_at = recency_at; @@ -329,7 +334,25 @@ async fn apply_metadata_update( .map_err(|err| ThreadStoreError::Internal { message: format!("failed to update thread metadata for {thread_id}: {err}"), })?; - if existing.is_some() + if let Some(recency_at) = recency_at { + let replaced = state_db + .set_thread_recency_at(thread_id, recency_at) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!( + "failed to replace thread recency_at for {thread_id}: {err}" + ), + })?; + if !replaced { + return Err(ThreadStoreError::Internal { + message: format!( + "thread metadata disappeared before replacing recency_at for {thread_id}" + ), + }); + } + } + if recency_at.is_none() + && existing.is_some() && let Some(recency_at) = advance_recency_at { state_db @@ -409,6 +432,9 @@ async fn metadata_for_missing_sqlite_row( builder.cwd = patch.cwd.clone().map(normalize_cwd).unwrap_or_default(); builder.cli_version = patch.cli_version.clone(); let mut metadata = builder.build(store.config.default_model_provider_id.as_str()); + if let Some(recency_at) = patch.recency_at { + metadata.recency_at = recency_at; + } if rollout_path_archived { metadata.archived_at = Some(metadata.updated_at); } @@ -486,6 +512,7 @@ fn has_observed_metadata_facts(patch: &ThreadMetadataPatch) -> bool { || patch.model.is_some() || patch.reasoning_effort.is_some() || patch.created_at.is_some() + || patch.recency_at.is_some() || patch.source.is_some() || patch.thread_source.is_some() || patch.agent_nickname.is_some() @@ -1822,6 +1849,7 @@ mod tests { cwd: Some(std::env::current_dir().expect("cwd")), model_provider: "test-provider".to_string(), memory_mode: ThreadMemoryMode::Enabled, + historical_timestamps: None, } } diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index 1057715a4a..c8f3cb65cd 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -361,6 +361,7 @@ fn update_has_metadata_facts(update: &ThreadMetadataPatch) -> bool { || update.model.is_some() || update.reasoning_effort.is_some() || update.created_at.is_some() + || update.recency_at.is_some() || update.advance_recency_at.is_some() || update.source.is_some() || update.thread_source.is_some() @@ -569,6 +570,7 @@ mod tests { cwd: None, model_provider: "test-provider".to_string(), memory_mode: ThreadMemoryMode::Enabled, + historical_timestamps: None, }, } } diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 4e688eca6f..af5df48fc4 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -58,6 +58,16 @@ pub struct ThreadPersistenceMetadata { pub model_provider: String, /// Memory mode associated with the live thread. pub memory_mode: MemoryMode, + /// Original timestamps to retain when materializing imported history. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub historical_timestamps: Option, +} + +/// Original timestamps for a thread imported from another agent. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct HistoricalThreadTimestamps { + pub created_at: DateTime, + pub updated_at: DateTime, } /// Extra configuration fields for a thread. @@ -548,6 +558,8 @@ pub struct ThreadMetadataPatch { pub created_at: Option>, /// Last update timestamp for this metadata observation. pub updated_at: Option>, + /// Replacement product-recency timestamp when historical chronology must be restored. + pub recency_at: Option>, /// Advance product recency to at least this timestamp. pub advance_recency_at: Option>, /// Session source. @@ -632,8 +644,16 @@ impl ThreadMetadataPatch { if next.updated_at.is_some() { self.updated_at = next.updated_at; } - if next.advance_recency_at.is_some() { - self.advance_recency_at = next.advance_recency_at; + match (next.recency_at, next.advance_recency_at) { + (Some(recency_at), _) => { + self.recency_at = Some(recency_at); + self.advance_recency_at = None; + } + (None, Some(advance_recency_at)) => { + self.recency_at = None; + self.advance_recency_at = Some(advance_recency_at); + } + (None, None) => {} } if next.source.is_some() { self.source = next.source; @@ -688,6 +708,7 @@ impl ThreadMetadataPatch { && self.reasoning_effort.is_none() && self.created_at.is_none() && self.updated_at.is_none() + && self.recency_at.is_none() && self.advance_recency_at.is_none() && self.source.is_none() && self.thread_source.is_none() @@ -853,6 +874,35 @@ mod tests { ); } + #[test] + fn thread_metadata_patch_merge_uses_latest_recency_operation() { + let historical_recency = + DateTime::::from_timestamp(1_700_000_000, 0).expect("historical recency"); + let live_recency = DateTime::::from_timestamp(1_800_000_000, 0).expect("live recency"); + let mut current = ThreadMetadataPatch { + advance_recency_at: Some(live_recency), + ..Default::default() + }; + + current.merge(ThreadMetadataPatch { + recency_at: Some(historical_recency), + ..Default::default() + }); + assert_eq!( + (current.recency_at, current.advance_recency_at), + (Some(historical_recency), None) + ); + + current.merge(ThreadMetadataPatch { + advance_recency_at: Some(live_recency), + ..Default::default() + }); + assert_eq!( + (current.recency_at, current.advance_recency_at), + (None, Some(live_recency)) + ); + } + fn session_meta(history_mode: ThreadHistoryMode) -> RolloutItem { RolloutItem::SessionMeta(codex_protocol::protocol::SessionMetaLine { meta: codex_protocol::protocol::SessionMeta {