openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-05 15:18:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: preserve imported session source chronology

Browse Source
This commit is contained in:
stefanstokic-oai
2026-07-08 20:18:22 -04:00
parent 20e5edfa74
commit 706f55071e
12 changed files with 672 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
codex-rs/app-server/src/request_processors/external_agent_session_import.rs
View File

@@ -1,6 +1,7 @@
use std::path::PathBuf;
use std::sync::Arc;
use chrono::DateTime;
use chrono::Utc;
use codex_arg0::Arg0DispatchPaths;
use codex_core::ThreadManager;
@@ -20,6 +21,7 @@ use codex_protocol::protocol::ThreadMemoryMode;
use codex_rollout::is_persisted_rollout_item;
use codex_thread_store::AppendThreadItemsParams;
use codex_thread_store::CreateThreadParams;
use codex_thread_store::InitialThreadTimestamps;
use codex_thread_store::ThreadMetadataPatch;
use codex_thread_store::ThreadPersistenceMetadata;
use codex_thread_store::ThreadStore;
@@ -168,6 +170,8 @@ impl ExternalAgentSessionImporter {
cwd,
title,
first_user_message,
source_created_at,
source_updated_at,
mut rollout_items,
} = session;
let config = self
@@ -203,7 +207,23 @@ impl ExternalAgentSessionImporter {
} else {
ThreadMemoryMode::Disabled
};
let now = Utc::now();
let initial_timestamps = source_created_at
.zip(source_updated_at)
.and_then(|(created_at, updated_at)| {
Some(InitialThreadTimestamps {
created_at: DateTime::from_timestamp(created_at, /*nsecs*/ 0)?,
updated_at: DateTime::from_timestamp(updated_at, /*nsecs*/ 0)?,
})
})
.unwrap_or_else(|| {
let now = Utc::now();
InitialThreadTimestamps {
created_at: now,
updated_at: now,
}
});
let created_at = initial_timestamps.created_at;
let updated_at = initial_timestamps.updated_at;
let create_params = CreateThreadParams {
session_id: thread_id.into(),
thread_id,
@@ -238,8 +258,9 @@ impl ExternalAgentSessionImporter {
title,
preview: first_user_message.clone(),
model_provider: Some(model_provider),
created_at: Some(now),
updated_at: Some(now),
created_at: Some(created_at),
updated_at: Some(updated_at),
advance_recency_at: Some(updated_at),
source: Some(source.clone()),
thread_source: Some(None),
agent_nickname: Some(source.get_nickname()),
@@ -253,7 +274,7 @@ impl ExternalAgentSessionImporter {
};
self.thread_store
.create_thread(create_params)
.create_thread_with_initial_timestamps(create_params, initial_timestamps)
.await
.map_err(|err| format!("failed to import session: {err}"))?;
if !rollout_items.is_empty()

274
codex-rs/app-server/tests/suite/v2/external_agent_config.rs
View File

@@ -18,6 +18,8 @@ use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::PluginListParams;
use codex_app_server_protocol::PluginListResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadListParams;
use codex_app_server_protocol::ThreadListResponse;
@@ -25,6 +27,9 @@ use codex_app_server_protocol::ThreadReadParams;
use codex_app_server_protocol::ThreadReadResponse;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadSortKey;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::UserInput;
use codex_config::types::AuthCredentialsStoreMode;
@@ -51,6 +56,58 @@ fn assert_import_response(response: ExternalAgentConfigImportResponse) -> String
response.import_id
}
async fn list_threads(
mcp: &mut TestAppServer,
sort_key: ThreadSortKey,
use_state_db_only: bool,
) -> Result<ThreadListResponse> {
let request_id = mcp
.send_thread_list_request(ThreadListParams {
cursor: None,
limit: None,
sort_key: Some(sort_key),
sort_direction: Some(SortDirection::Desc),
model_providers: None,
source_kinds: None,
archived: None,
cwd: None,
use_state_db_only,
search_term: None,
parent_thread_id: None,
ancestor_thread_id: None,
})
.await?;
let response: JSONRPCResponse = timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
to_response(response)
}
fn assert_thread_timestamps(thread: &Thread, created_at: i64, updated_at: i64) {
assert_eq!(
(thread.created_at, thread.updated_at, thread.recency_at),
(created_at, updated_at, Some(updated_at))
);
}
fn remove_state_db_files(codex_home: &Path) -> Result<()> {
let state_db_path = codex_state::state_db_path(codex_home);
for path in [
state_db_path.clone(),
PathBuf::from(format!("{}-wal", state_db_path.display())),
PathBuf::from(format!("{}-shm", state_db_path.display())),
] {
match std::fs::remove_file(path) {
Ok(()) => {}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
Err(err) => return Err(err.into()),
}
}
Ok(())
}
#[tokio::test]
async fn external_agent_config_import_sends_completion_notification_for_sync_only_import()
-> Result<()> {
@@ -612,7 +669,12 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> {
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let project_root = codex_home.path().join("repo");
let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
let source_created_at_text = "2024-01-02T03:04:05Z";
let source_updated_at_text = "2024-02-03T04:05:06Z";
let source_created_at =
chrono::DateTime::parse_from_rfc3339(source_created_at_text)?.timestamp();
let source_updated_at =
chrono::DateTime::parse_from_rfc3339(source_updated_at_text)?.timestamp();
let session_dir = external_agent_home(codex_home.path()).join("projects/repo");
let session_path = session_dir.join("session.jsonl");
let control_request = "<ide_selection>src/auth.rs:1-5</ide_selection>";
@@ -625,21 +687,21 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> {
serde_json::json!({
"type": "user",
"cwd": &project_root,
"timestamp": &recent_timestamp,
"timestamp": source_created_at_text,
"message": { "content": control_request },
})
.to_string(),
serde_json::json!({
"type": "user",
"cwd": &project_root,
"timestamp": &recent_timestamp,
"timestamp": "2024-01-03T00:00:00Z",
"message": { "content": first_request },
})
.to_string(),
serde_json::json!({
"type": "assistant",
"cwd": &project_root,
"timestamp": &recent_timestamp,
"timestamp": source_updated_at_text,
"message": { "content": "first answer" },
})
.to_string(),
@@ -656,6 +718,42 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> {
.await?;
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
let request_id = mcp
.send_thread_start_request(ThreadStartParams {
cwd: Some(project_root.display().to_string()),
model: Some("mock-model".to_string()),
..Default::default()
})
.await?;
let response: JSONRPCResponse = timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let native_thread: ThreadStartResponse = to_response(response)?;
let native_thread_id = native_thread.thread.id;
let request_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: native_thread_id.clone(),
client_user_message_id: None,
input: vec![UserInput::Text {
text: "newer native request".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let request_id = mcp
.send_raw_request(
"externalAgentConfig/detect",
@@ -727,37 +825,58 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> {
.expect("session success should include imported thread id")
.to_string();
let request_id = mcp
.send_thread_list_request(ThreadListParams {
cursor: None,
limit: None,
sort_key: None,
sort_direction: None,
model_providers: None,
source_kinds: None,
archived: None,
cwd: None,
use_state_db_only: false,
search_term: None,
parent_thread_id: None,
ancestor_thread_id: None,
})
.await?;
let response: JSONRPCResponse = timeout(
DEFAULT_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
let response = list_threads(
&mut mcp,
ThreadSortKey::CreatedAt,
/*use_state_db_only*/ false,
)
.await??;
let response: ThreadListResponse = to_response(response)?;
.await?;
assert_eq!(
response
.data
.iter()
.map(|thread| thread.id.as_str())
.collect::<Vec<_>>(),
vec![native_thread_id.as_str(), imported_thread_id.as_str()]
);
let thread = response
.data
.first()
.iter()
.find(|thread| thread.id == imported_thread_id)
.expect("expected imported thread")
.clone();
assert_eq!(imported_thread_id, thread.id.to_string());
let native_thread = response
.data
.iter()
.find(|thread| thread.id == native_thread_id)
.expect("expected native thread");
assert_thread_timestamps(&thread, source_created_at, source_updated_at);
assert!(native_thread.created_at > source_updated_at);
assert!(native_thread.updated_at > source_updated_at);
assert!(
native_thread
.recency_at
.is_some_and(|value| value > source_updated_at)
);
assert_eq!(thread.preview, control_request);
assert_eq!(thread.name.as_deref(), Some("Fix auth flow"));
for sort_key in [ThreadSortKey::UpdatedAt, ThreadSortKey::RecencyAt] {
let response = list_threads(&mut mcp, sort_key, /*use_state_db_only*/ false).await?;
assert_eq!(
response.data.first().map(|thread| thread.id.as_str()),
Some(native_thread_id.as_str())
);
}
let rollout_path = thread.path.clone().expect("imported rollout path");
let session_meta = codex_rollout::read_session_meta_line(&rollout_path).await?;
let rollout_modified_at = std::fs::metadata(&rollout_path)?
.modified()
.map(chrono::DateTime::<chrono::Utc>::from)?;
assert_eq!(session_meta.meta.timestamp, "2024-01-02T03:04:05.000Z");
assert_eq!(rollout_modified_at.timestamp(), source_updated_at);
let request_id = mcp
.send_thread_read_request(ThreadReadParams {
thread_id: thread.id.clone(),
@@ -809,9 +928,43 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> {
})
);
drop(mcp);
remove_state_db_files(codex_home.path())?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.with_env_overrides(&[("HOME", Some(home_dir.as_str()))])
.build()
.await?;
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
list_threads(
&mut mcp,
ThreadSortKey::CreatedAt,
/*use_state_db_only*/ false,
)
.await?;
for sort_key in [
ThreadSortKey::CreatedAt,
ThreadSortKey::UpdatedAt,
ThreadSortKey::RecencyAt,
] {
let rebuilt = list_threads(&mut mcp, sort_key, /*use_state_db_only*/ true).await?;
assert_eq!(
rebuilt.data.first().map(|thread| thread.id.as_str()),
Some(native_thread_id.as_str())
);
let rebuilt_thread = rebuilt
.data
.iter()
.find(|thread| thread.id == imported_thread_id)
.expect("rebuilt imported thread");
assert_thread_timestamps(rebuilt_thread, source_created_at, source_updated_at);
}
let request_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id.clone(),
thread_id: imported_thread_id.clone(),
..Default::default()
})
.await?;
@@ -824,7 +977,7 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> {
let request_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
thread_id: imported_thread_id.clone(),
client_user_message_id: None,
input: vec![UserInput::Text {
text: "follow up".to_string(),
@@ -844,9 +997,36 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> {
)
.await??;
let response = list_threads(
&mut mcp,
ThreadSortKey::RecencyAt,
/*use_state_db_only*/ true,
)
.await?;
let updated_thread = response
.data
.iter()
.find(|thread| thread.id == imported_thread_id)
.expect("updated imported thread");
assert_eq!(updated_thread.created_at, source_created_at);
assert!(updated_thread.updated_at > source_updated_at);
assert!(
updated_thread
.recency_at
.is_some_and(|value| value > source_updated_at)
);
assert_eq!(
response.data.first().map(|thread| thread.id.as_str()),
Some(imported_thread_id.as_str())
);
let resumed_rollout_modified_at = std::fs::metadata(&rollout_path)?
.modified()
.map(chrono::DateTime::<chrono::Utc>::from)?;
assert!(resumed_rollout_modified_at.timestamp() > source_updated_at);
let request_id = mcp
.send_thread_read_request(ThreadReadParams {
thread_id: thread.id,
thread_id: imported_thread_id.clone(),
include_turns: true,
})
.await?;
@@ -862,6 +1042,40 @@ async fn external_agent_config_import_creates_session_rollouts() -> Result<()> {
other => panic!("expected agent message item, got {other:?}"),
}
drop(mcp);
remove_state_db_files(codex_home.path())?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.with_env_overrides(&[("HOME", Some(home_dir.as_str()))])
.build()
.await?;
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
list_threads(
&mut mcp,
ThreadSortKey::RecencyAt,
/*use_state_db_only*/ false,
)
.await?;
let repaired_after_resume = list_threads(
&mut mcp,
ThreadSortKey::RecencyAt,
/*use_state_db_only*/ true,
)
.await?;
let repaired_thread = repaired_after_resume
.data
.iter()
.find(|thread| thread.id == imported_thread_id)
.expect("repaired resumed imported thread");
assert_eq!(repaired_thread.created_at, source_created_at);
assert!(repaired_thread.updated_at > source_updated_at);
assert!(
repaired_thread
.recency_at
.is_some_and(|value| value > source_updated_at)
);
Ok(())
}

117
codex-rs/external-agent-sessions/src/export.rs
View File

@@ -39,6 +39,9 @@ pub(crate) fn load_session_for_import_with_content_sha256(
return Ok(None);
};
let messages = parsed.messages;
let (source_created_at, source_updated_at) = source_chronology(&messages)
.map(|(created_at, updated_at)| (Some(created_at), Some(updated_at)))
.unwrap_or_default();
let first_user_message_text = messages
.iter()
.find(|message| message.role == MessageRole::User)
@@ -64,12 +67,27 @@ pub(crate) fn load_session_for_import_with_content_sha256(
cwd,
title,
first_user_message,
source_created_at,
source_updated_at,
rollout_items,
},
parsed.content_sha256,
)))
}
fn source_chronology(messages: &[ConversationMessage]) -> Option<(i64, i64)> {
let mut timestamps = messages
.iter()
.skip_while(|message| message.role != MessageRole::User)
.filter_map(|message| message.timestamp);
let first = timestamps.next()?;
Some(
timestamps.fold((first, first), |(created_at, updated_at), timestamp| {
(created_at.min(timestamp), updated_at.max(timestamp))
}),
)
}
fn rollout_items_from_messages(messages: Vec<ConversationMessage>) -> Vec<RolloutItem> {
let mut items = Vec::new();
let mut current_turn = None;
@@ -504,8 +522,107 @@ mod tests {
assert_eq!(token_count.total_token_usage, token_count.last_token_usage);
}
#[test]
fn derives_source_chronology_from_retained_valid_message_timestamps() {
let root = TempDir::new().expect("tempdir");
let project_root = root.path().join("repo");
std::fs::create_dir_all(&project_root).expect("project root");
let path = root.path().join("session.jsonl");
std::fs::write(
&path,
jsonl(&[
record_at(
"assistant",
"discarded before first user",
&project_root,
Some("2020-01-01T00:00:00Z"),
),
record_at(
"user",
"first retained message",
&project_root,
Some("2025-03-04T05:06:07Z"),
),
record_at(
"assistant",
"invalid timestamp",
&project_root,
Some("not-a-timestamp"),
),
record_at(
"assistant",
"earliest retained timestamp",
&project_root,
Some("2024-01-02T03:04:05Z"),
),
record_at(
"assistant",
"latest retained timestamp",
&project_root,
Some("2026-05-06T07:08:09Z"),
),
]),
)
.expect("session");
let imported = load_session_for_import(&path)
.expect("load")
.expect("session");
assert_eq!(
(imported.source_created_at, imported.source_updated_at),
(
chrono::DateTime::parse_from_rfc3339("2024-01-02T03:04:05Z")
.ok()
.map(|timestamp| timestamp.timestamp()),
chrono::DateTime::parse_from_rfc3339("2026-05-06T07:08:09Z")
.ok()
.map(|timestamp| timestamp.timestamp()),
)
);
}
#[test]
fn leaves_source_chronology_empty_without_valid_timestamps() {
let root = TempDir::new().expect("tempdir");
let project_root = root.path().join("repo");
std::fs::create_dir_all(&project_root).expect("project root");
let path = root.path().join("session.jsonl");
std::fs::write(
&path,
jsonl(&[
record_at(
"user",
"missing timestamp",
&project_root,
/*timestamp*/ None,
),
record_at(
"assistant",
"invalid timestamp",
&project_root,
Some("invalid"),
),
]),
)
.expect("session");
let imported = load_session_for_import(&path)
.expect("load")
.expect("session");
assert_eq!(
(imported.source_created_at, imported.source_updated_at),
(None, None)
);
}
fn record(role: &str, text: &str, cwd: &Path) -> JsonValue {
let timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
record_at(role, text, cwd, Some(timestamp.as_str()))
}
fn record_at(role: &str, text: &str, cwd: &Path, timestamp: Option<&str>) -> JsonValue {
serde_json::json!({
"type": role,
"cwd": cwd,

2
codex-rs/external-agent-sessions/src/lib.rs
View File

@@ -33,6 +33,8 @@ pub struct ImportedExternalAgentSession {
pub cwd: PathBuf,
pub title: Option<String>,
pub first_user_message: Option<String>,
pub source_created_at: Option<i64>,
pub source_updated_at: Option<i64>,
pub rollout_items: Vec<RolloutItem>,
}

67
codex-rs/rollout/src/recorder.rs
View File

@@ -8,8 +8,11 @@ use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::SystemTime;
use chrono::DateTime;
use chrono::SecondsFormat;
use chrono::Utc;
use codex_protocol::SessionId;
use codex_protocol::ThreadId;
use codex_protocol::capabilities::SelectedCapabilityRoot;
@@ -82,6 +85,7 @@ pub struct RolloutRecorder {
}
#[derive(Clone)]
#[allow(clippy::large_enum_variant)]
pub enum RolloutRecorderParams {
Create {
session_id: SessionId,
@@ -97,6 +101,8 @@ pub enum RolloutRecorderParams {
multi_agent_version: Option<MultiAgentVersion>,
history_mode: ThreadHistoryMode,
initial_window_id: Option<String>,
initial_created_at: Option<DateTime<Utc>>,
initial_updated_at: Option<DateTime<Utc>>,
},
Resume {
path: PathBuf,
@@ -190,6 +196,8 @@ impl RolloutRecorderParams {
multi_agent_version: None,
history_mode: Default::default(),
initial_window_id: None,
initial_created_at: None,
initial_updated_at: None,
}
}
@@ -249,6 +257,23 @@ impl RolloutRecorderParams {
self
}
pub fn with_initial_timestamps(
mut self,
created_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
) -> Self {
if let Self::Create {
initial_created_at,
initial_updated_at,
..
} = &mut self
{
*initial_created_at = Some(created_at);
*initial_updated_at = Some(updated_at);
}
self
}
pub fn resume(path: PathBuf) -> Self {
Self::Resume { path }
}
@@ -766,8 +791,15 @@ impl RolloutRecorder {
multi_agent_version,
history_mode,
initial_window_id,
initial_created_at,
initial_updated_at,
} => {
let log_file_info = precompute_log_file_info(config, conversation_id)?;
let log_file_info = precompute_log_file_info(
config,
conversation_id,
initial_created_at,
initial_updated_at,
)?;
let path = log_file_info.path.clone();
let thread_id = log_file_info.conversation_id;
let started_at = log_file_info.timestamp;
@@ -1493,15 +1525,31 @@ struct LogFileInfo {
/// Timestamp for the start of the session.
timestamp: OffsetDateTime,
/// Initial file modification time used when reconstructing imported chronology.
initial_modified_at: Option<SystemTime>,
}
fn precompute_log_file_info(
config: &impl RolloutConfigView,
conversation_id: ThreadId,
initial_created_at: Option<DateTime<Utc>>,
initial_updated_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 initial_created_at {
Some(created_at) => {
let timestamp = OffsetDateTime::from(SystemTime::from(created_at));
let local_offset = time::UtcOffset::local_offset_at(timestamp).map_err(|e| {
IoError::other(format!(
"failed to get local offset for initial session time: {e}"
))
})?;
timestamp.to_offset(local_offset)
}
None => OffsetDateTime::now_local()
.map_err(|e| IoError::other(format!("failed to get local time: {e}")))?,
};
let mut dir = config.codex_home().to_path_buf();
dir.push(SESSIONS_SUBDIR);
dir.push(timestamp.year().to_string());
@@ -1524,6 +1572,7 @@ fn precompute_log_file_info(
path,
conversation_id,
timestamp,
initial_modified_at: initial_updated_at.map(SystemTime::from),
})
}
@@ -1554,6 +1603,7 @@ struct RolloutWriterState {
meta: Option<SessionMeta>,
cwd: PathBuf,
rollout_path: PathBuf,
initial_modified_at: Option<SystemTime>,
last_logged_error: Option<String>,
}
@@ -1565,6 +1615,9 @@ impl RolloutWriterState {
cwd: PathBuf,
rollout_path: PathBuf,
) -> Self {
let initial_modified_at = deferred_log_file_info
.as_ref()
.and_then(|info| info.initial_modified_at);
Self {
writer: file.map(|file| JsonlWriter { file }),
deferred_log_file_info,
@@ -1572,6 +1625,7 @@ impl RolloutWriterState {
meta,
cwd,
rollout_path,
initial_modified_at,
last_logged_error: None,
}
}
@@ -1689,6 +1743,13 @@ impl RolloutWriterState {
if let Some(writer) = self.writer.as_mut() {
writer.file.flush().await?;
}
if let Some(modified_at) = self.initial_modified_at {
fs::OpenOptions::new()
.append(true)
.open(&self.rollout_path)?
.set_times(fs::FileTimes::new().set_modified(modified_at))?;
self.initial_modified_at = None;
}
Ok(())
}

9
codex-rs/thread-store/src/in_memory.rs
View File

@@ -22,6 +22,7 @@ use crate::AppendThreadItemsParams;
use crate::ArchiveThreadParams;
use crate::CreateThreadParams;
use crate::DeleteThreadParams;
use crate::InitialThreadTimestamps;
use crate::ListThreadsParams;
use crate::LoadThreadHistoryParams;
use crate::ReadThreadByRolloutPathParams;
@@ -612,6 +613,14 @@ impl ThreadStore for InMemoryThreadStore {
Box::pin(InMemoryThreadStore::create_thread(self, params))
}
fn create_thread_with_initial_timestamps(
&self,
params: CreateThreadParams,
_timestamps: InitialThreadTimestamps,
) -> ThreadStoreFuture<'_, ()> {
Box::pin(InMemoryThreadStore::create_thread(self, params))
}
fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreFuture<'_, ()> {
Box::pin(InMemoryThreadStore::resume_thread(self, params))
}

1
codex-rs/thread-store/src/lib.rs
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::InitialThreadTimestamps;
pub use types::ItemPage;
pub use types::ListItemsParams;
pub use types::ListThreadsParams;

63
codex-rs/thread-store/src/local/create_thread.rs
View File

@@ -1,5 +1,6 @@
use super::LocalThreadStore;
use crate::CreateThreadParams;
use crate::InitialThreadTimestamps;
use crate::ThreadStoreError;
use crate::ThreadStoreResult;
use crate::error::reject_paginated_history_mode;
@@ -11,6 +12,22 @@ use codex_rollout::RolloutRecorderParams;
pub(super) async fn create_thread(
store: &LocalThreadStore,
params: CreateThreadParams,
) -> ThreadStoreResult<RolloutRecorder> {
create_thread_recorder(store, params, /*initial_timestamps*/ None).await
}
pub(super) async fn create_thread_with_initial_timestamps(
store: &LocalThreadStore,
params: CreateThreadParams,
initial_timestamps: InitialThreadTimestamps,
) -> ThreadStoreResult<RolloutRecorder> {
create_thread_recorder(store, params, Some(initial_timestamps)).await
}
async fn create_thread_recorder(
store: &LocalThreadStore,
params: CreateThreadParams,
initial_timestamps: Option<InitialThreadTimestamps>,
) -> ThreadStoreResult<RolloutRecorder> {
reject_paginated_history_mode(params.history_mode)?;
let cwd = params
@@ -27,26 +44,30 @@ pub(super) async fn create_thread(
model_provider_id: params.metadata.model_provider.clone(),
generate_memories: matches!(params.metadata.memory_mode, ThreadMemoryMode::Enabled),
};
RolloutRecorder::new(
&config,
RolloutRecorderParams::new(
params.thread_id,
params.forked_from_id,
params.parent_thread_id,
params.source,
params.thread_source,
params.originator,
params.base_instructions,
params.dynamic_tools,
)
.with_session_id(params.session_id)
.with_selected_capability_roots(params.selected_capability_roots)
.with_multi_agent_version(params.multi_agent_version)
.with_history_mode(params.history_mode)
.with_initial_window_id(params.initial_window_id),
let recorder_params = RolloutRecorderParams::new(
params.thread_id,
params.forked_from_id,
params.parent_thread_id,
params.source,
params.thread_source,
params.originator,
params.base_instructions,
params.dynamic_tools,
)
.await
.map_err(|err| ThreadStoreError::Internal {
message: format!("failed to initialize local thread recorder: {err}"),
})
.with_session_id(params.session_id)
.with_selected_capability_roots(params.selected_capability_roots)
.with_multi_agent_version(params.multi_agent_version)
.with_history_mode(params.history_mode)
.with_initial_window_id(params.initial_window_id);
let recorder_params = match initial_timestamps {
Some(timestamps) => {
recorder_params.with_initial_timestamps(timestamps.created_at, timestamps.updated_at)
}
None => recorder_params,
};
RolloutRecorder::new(&config, recorder_params)
.await
.map_err(|err| ThreadStoreError::Internal {
message: format!("failed to initialize local thread recorder: {err}"),
})
}

16
codex-rs/thread-store/src/local/live_writer.rs
View File

@@ -12,6 +12,7 @@ use super::LocalThreadStore;
use super::create_thread;
use crate::AppendThreadItemsParams;
use crate::CreateThreadParams;
use crate::InitialThreadTimestamps;
use crate::ReadThreadParams;
use crate::ResumeThreadParams;
use crate::ThreadStoreError;
@@ -34,6 +35,21 @@ pub(super) async fn create_thread(
.await
}
pub(super) async fn create_thread_with_initial_timestamps(
store: &LocalThreadStore,
params: CreateThreadParams,
timestamps: InitialThreadTimestamps,
) -> ThreadStoreResult<()> {
let thread_id = params.thread_id;
let history_mode = params.history_mode;
store.ensure_live_recorder_absent(thread_id).await?;
let recorder =
create_thread::create_thread_with_initial_timestamps(store, params, timestamps).await?;
store
.insert_live_recorder(thread_id, recorder, history_mode)
.await
}
pub(super) async fn resume_thread(
store: &LocalThreadStore,
params: ResumeThreadParams,

132
codex-rs/thread-store/src/local/mod.rs
View File

@@ -26,6 +26,7 @@ use crate::AppendThreadItemsParams;
use crate::ArchiveThreadParams;
use crate::CreateThreadParams;
use crate::DeleteThreadParams;
use crate::InitialThreadTimestamps;
use crate::ListThreadsParams;
use crate::LoadThreadHistoryParams;
use crate::ReadThreadByRolloutPathParams;
@@ -246,6 +247,16 @@ impl ThreadStore for LocalThreadStore {
Box::pin(async move { live_writer::create_thread(self, params).await })
}
fn create_thread_with_initial_timestamps(
&self,
params: CreateThreadParams,
timestamps: InitialThreadTimestamps,
) -> ThreadStoreFuture<'_, ()> {
Box::pin(async move {
live_writer::create_thread_with_initial_timestamps(self, params, timestamps).await
})
}
fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreFuture<'_, ()> {
Box::pin(async move { live_writer::resume_thread(self, params).await })
}
@@ -325,6 +336,8 @@ impl ThreadStore for LocalThreadStore {
mod tests {
use std::sync::Arc;
use chrono::DateTime;
use chrono::Utc;
use codex_protocol::ThreadId;
use codex_protocol::models::BaseInstructions;
use codex_protocol::models::FunctionCallOutputPayload;
@@ -342,7 +355,9 @@ mod tests {
use tempfile::TempDir;
use super::*;
use crate::InitialThreadTimestamps;
use crate::LiveThread;
use crate::ThreadMetadataPatch;
use crate::ThreadPersistenceMetadata;
use crate::local::test_support::test_config;
use crate::local::test_support::write_archived_session_file;
@@ -398,6 +413,123 @@ mod tests {
);
}
#[tokio::test]
async fn initial_timestamps_survive_read_repair() {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(),
config.default_model_provider_id.clone(),
)
.await
.expect("state db should initialize");
let store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
let thread_id = ThreadId::new();
let created_at = DateTime::parse_from_rfc3339("2024-01-02T03:04:05Z")
.expect("created timestamp")
.with_timezone(&Utc);
let updated_at = DateTime::parse_from_rfc3339("2024-02-03T04:05:06Z")
.expect("updated timestamp")
.with_timezone(&Utc);
store
.create_thread_with_initial_timestamps(
create_thread_params(thread_id),
InitialThreadTimestamps {
created_at,
updated_at,
},
)
.await
.expect("create historical thread");
store
.append_items(AppendThreadItemsParams {
thread_id,
items: vec![user_message_item("source history")],
})
.await
.expect("append source history");
store
.update_thread_metadata(UpdateThreadMetadataParams {
thread_id,
patch: ThreadMetadataPatch {
created_at: Some(created_at),
updated_at: Some(updated_at),
advance_recency_at: Some(updated_at),
..Default::default()
},
include_archived: false,
})
.await
.expect("seed source chronology");
let rollout_path = store
.live_rollout_path(thread_id)
.await
.expect("read rollout path");
store
.persist_thread(thread_id)
.await
.expect("persist historical thread");
store
.shutdown_thread(thread_id)
.await
.expect("shutdown historical thread");
let session_meta = codex_rollout::read_session_meta_line(&rollout_path)
.await
.expect("read session metadata");
let rollout_modified_at = std::fs::metadata(&rollout_path)
.expect("rollout metadata")
.modified()
.map(DateTime::<Utc>::from)
.expect("rollout modified timestamp");
assert_eq!(session_meta.meta.timestamp, "2024-01-02T03:04:05.000Z");
assert_eq!(rollout_modified_at, updated_at);
let local_day = created_at
.with_timezone(&chrono::Local)
.format("%Y/%m/%d")
.to_string();
assert!(rollout_path.starts_with(home.path().join("sessions").join(local_day)));
assert_eq!(
runtime
.delete_thread(thread_id)
.await
.expect("delete sqlite metadata"),
1
);
drop(store);
drop(runtime);
let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(),
config.default_model_provider_id.clone(),
)
.await
.expect("state db should reopen");
let store = LocalThreadStore::new(config, Some(runtime.clone()));
store
.read_thread(ReadThreadParams {
thread_id,
include_archived: false,
include_history: false,
})
.await
.expect("read and repair historical thread");
let repaired = runtime
.get_thread(thread_id)
.await
.expect("read repaired sqlite metadata")
.expect("repaired sqlite metadata");
assert_eq!(
(
repaired.created_at,
repaired.updated_at,
repaired.recency_at,
),
(created_at, updated_at, updated_at)
);
}
#[tokio::test]
async fn raw_append_items_does_not_update_sqlite_metadata() {
// This pins the ThreadStore contract: raw appends are history-only. Callers that need

11
codex-rs/thread-store/src/store.rs
View File

@@ -8,6 +8,7 @@ use crate::AppendThreadItemsParams;
use crate::ArchiveThreadParams;
use crate::CreateThreadParams;
use crate::DeleteThreadParams;
use crate::InitialThreadTimestamps;
use crate::ItemPage;
use crate::ListItemsParams;
use crate::ListThreadsParams;
@@ -45,6 +46,16 @@ pub trait ThreadStore: Any + Send + Sync {
/// Creates a new live thread.
fn create_thread(&self, params: CreateThreadParams) -> ThreadStoreFuture<'_, ()>;
/// Creates a thread from pre-existing history using its original chronology.
///
/// This is a creation-only initialization hook. Normal appends after creation must advance
/// update and recency timestamps through the usual live-thread path.
fn create_thread_with_initial_timestamps(
&self,
params: CreateThreadParams,
timestamps: InitialThreadTimestamps,
) -> ThreadStoreFuture<'_, ()>;
/// Reopens an existing thread for live appends.
fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreFuture<'_, ()>;

9
codex-rs/thread-store/src/types.rs
View File

@@ -60,6 +60,15 @@ pub struct ThreadPersistenceMetadata {
pub memory_mode: MemoryMode,
}
/// Source chronology used only while creating a thread from pre-existing history.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct InitialThreadTimestamps {
/// Original creation time of the imported thread.
pub created_at: DateTime<Utc>,
/// Original last-activity time of the imported thread.
pub updated_at: DateTime<Utc>,
}
/// Extra configuration fields for a thread.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ExtraConfig {}