openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-27 18:48:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve imported session fidelity

Browse Source
This commit is contained in:
charlesgong-openai
2026-07-08 13:38:21 -04:00
parent 23aac925e7
commit dd975f36cd
24 changed files with 656 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/external_agent_session_import.rs
+58 -16
View File
@@ -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<Semaphore>,
reserved_source_identities: Arc<Mutex<HashSet<String>>>,
thread_manager: Arc<ThreadManager>,
thread_store: Arc<dyn ThreadStore>,
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::<Utc>::from_timestamp(timestamp, 0))
.unwrap_or(now);
let updated_at = updated_at
.and_then(|timestamp| DateTime::<Utc>::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)
}
}
codex-rs/app-server/tests/suite/conversation_summary.rs
+1
View File
@@ -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?;
codex-rs/app-server/tests/suite/v2/external_agent_config.rs
+32 -3
View File
@@ -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!(
codex-rs/app-server/tests/suite/v2/remote_thread_store.rs
+1
View File
@@ -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?;
codex-rs/app-server/tests/suite/v2/thread_read.rs
+1
View File
@@ -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?;
codex-rs/app-server/tests/suite/v2/thread_unarchive.rs
+1
View File
@@ -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?;
codex-rs/core/src/session/session.rs
+2
View File
@@ -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?
codex-rs/core/src/session/tests.rs
+4
View File
@@ -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,
},
},
)
codex-rs/external-agent-sessions/src/detect.rs
+19 -25
View File
@@ -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()
);
}
codex-rs/external-agent-sessions/src/export.rs
+118 -5
View File
@@ -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<ConversationMessage>) -> Vec<Rollou
let mut response_item_bytes = 0i64;
let mut last_model_visible_tokens = 0i64;
let mut user_turn_count = 0usize;
let completed_at = messages.last().and_then(|message| message.timestamp);
let mut current_turn_started_at = None;
let mut current_turn_completed_at = None;
for message in messages {
match message.role {
MessageRole::User => {
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<ConversationMessage>) -> Vec<Rollou
if current_turn.is_none() {
continue;
}
if let Some(timestamp) = message.timestamp {
current_turn_completed_at = Some(
current_turn_completed_at
.map_or(timestamp, |completed_at| completed_at.max(timestamp)),
);
}
response_item_bytes =
response_item_bytes.saturating_add(message_byte_count(&message));
last_model_visible_tokens = approx_tokens_from_byte_count_i64(response_item_bytes);
@@ -114,7 +138,11 @@ fn rollout_items_from_messages(messages: Vec<ConversationMessage>) -> Vec<Rollou
if let Some(turn_id) = current_turn {
items.push(external_session_imported_marker_item());
items.push(token_count_item(last_model_visible_tokens));
items.push(turn_complete_item(turn_id, completed_at));
items.push(turn_complete_item(
turn_id,
current_turn_started_at,
current_turn_completed_at,
));
}
items
@@ -164,12 +192,21 @@ fn token_count_item(last_model_visible_tokens: i64) -> RolloutItem {
}))
}
fn turn_complete_item(turn_id: String, completed_at: Option<i64>) -> RolloutItem {
fn turn_complete_item(
turn_id: String,
started_at: Option<i64>,
completed_at: Option<i64>,
) -> 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<_>>(),
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, &timestamp)
}
fn record_at(role: &str, text: &str, cwd: &Path, timestamp: &str) -> JsonValue {
serde_json::json!({
"type": role,
"cwd": cwd,
codex-rs/external-agent-sessions/src/ledger.rs
+49 -22
View File
@@ -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<String>,
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<String>,
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<bool> {
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<bool> {
pub(super) fn contains_source_identity(
&self,
source_path: &Path,
source_session_id: Option<&str>,
) -> io::Result<bool> {
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<bool> {
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<ImportedExternalAgentSessionLedger> {
codex-rs/external-agent-sessions/src/ledger_tests.rs
+78 -1
View File
@@ -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")
);
}
codex-rs/external-agent-sessions/src/lib.rs
+39 -5
View File
@@ -32,6 +32,9 @@ pub struct ImportedExternalAgentSession {
pub cwd: PathBuf,
pub title: Option<String>,
pub first_user_message: Option<String>,
pub source_session_id: Option<String>,
pub created_at: Option<i64>,
pub updated_at: Option<i64>,
pub rollout_items: Vec<RolloutItem>,
}
@@ -46,15 +49,19 @@ pub fn prepare_validated_session_import(
codex_home: &Path,
session: ExternalAgentSessionMigration,
) -> io::Result<Option<PendingSessionImport>> {
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 {
codex-rs/external-agent-sessions/src/records.rs
+25
View File
@@ -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<String>,
pub migration: ExternalAgentSessionMigration,
}
pub(super) struct ParsedSessionImport {
pub cwd: Option<PathBuf>,
pub source_title: Option<String>,
pub source_session_id: Option<String>,
pub messages: Vec<ConversationMessage>,
pub content_sha256: String,
}
@@ -37,6 +39,7 @@ pub fn summarize_session(path: &Path) -> io::Result<Option<SessionSummary>> {
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<Option<SessionSummary>> {
let Ok(mut record) = serde_json::from_str::<JsonValue>(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<Option<SessionSummary>> {
};
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<ParsedSessionImport
let mut cwd = None;
let mut custom_title = None;
let mut ai_title = None;
let mut source_session_id = None;
let mut messages = Vec::new();
let mut line = String::new();
let mut hasher = Sha256::new();
@@ -115,6 +123,9 @@ pub(super) fn read_session_import(path: &Path) -> io::Result<ParsedSessionImport
let Ok(mut record) = serde_json::from_str::<JsonValue>(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<ParsedSessionImport
Ok(ParsedSessionImport {
cwd,
source_title: custom_title.or(ai_title),
source_session_id,
messages,
content_sha256: format!("{:x}", hasher.finalize()),
})
@@ -147,6 +159,14 @@ fn ai_title_from_record(record: &JsonValue) -> 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!(
codex-rs/rollout/src/recorder.rs
+86 -11
View File
@@ -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<SelectedCapabilityRoot>,
multi_agent_version: Option<MultiAgentVersion>,
history_mode: ThreadHistoryMode,
initial_window_id: Option<String>,
initial_window_id: Box<Option<String>>,
historical_timestamps: Option<Box<HistoricalRolloutTimestamps>>,
},
Resume {
path: PathBuf,
},
}
#[doc(hidden)]
#[derive(Clone, Default)]
pub struct HistoricalRolloutTimestamps {
created_at: Option<DateTime<Utc>>,
updated_at: Option<DateTime<Utc>>,
}
enum RolloutCmd {
AddItems(Vec<RolloutItem>),
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<Utc>) -> 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<Utc>) -> 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<Self> {
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<DateTime<Utc>>,
) -> std::io::Result<LogFileInfo> {
// 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<SessionMeta>,
cwd: PathBuf,
rollout_path: PathBuf,
updated_at: Option<DateTime<Utc>>,
last_logged_error: Option<String>,
}
@@ -1564,6 +1622,7 @@ impl RolloutWriterState {
meta: Option<SessionMeta>,
cwd: PathBuf,
rollout_path: PathBuf,
updated_at: Option<DateTime<Utc>>,
) -> 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<SessionMeta>,
cwd: PathBuf,
rollout_path: PathBuf,
updated_at: Option<DateTime<Utc>>,
) -> 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 {
codex-rs/rollout/src/recorder_tests.rs
+1
View File
@@ -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 {
codex-rs/state/src/runtime/threads.rs
+20
View File
@@ -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<Utc>,
) -> anyhow::Result<bool> {
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,
codex-rs/thread-store/src/in_memory.rs
+8 -1
View File
@@ -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
codex-rs/thread-store/src/lib.rs
+1
View File
@@ -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;
codex-rs/thread-store/src/local/create_thread.rs
+24 -21
View File
@@ -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}"),
})
}
codex-rs/thread-store/src/local/mod.rs
+4
View File
@@ -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,
}
}
codex-rs/thread-store/src/local/update_thread_metadata.rs
+30 -2
View File
@@ -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,
}
}
codex-rs/thread-store/src/thread_metadata_sync.rs
+2
View File
@@ -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,
},
}
}
codex-rs/thread-store/src/types.rs
+52 -2
View File
@@ -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<HistoricalThreadTimestamps>,
}
/// Original timestamps for a thread imported from another agent.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct HistoricalThreadTimestamps {
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
/// Extra configuration fields for a thread.
@@ -548,6 +558,8 @@ pub struct ThreadMetadataPatch {
pub created_at: Option<DateTime<Utc>>,
/// Last update timestamp for this metadata observation.
pub updated_at: Option<DateTime<Utc>>,
/// Replacement product-recency timestamp when historical chronology must be restored.
pub recency_at: Option<DateTime<Utc>>,
/// Advance product recency to at least this timestamp.
pub advance_recency_at: Option<DateTime<Utc>>,
/// 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::<Utc>::from_timestamp(1_700_000_000, 0).expect("historical recency");
let live_recency = DateTime::<Utc>::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 {