openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-03 14:59:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve imported session fidelity

Browse Source
This commit is contained in:
charlesgong-openai
2026-07-07 18:04:56 -04:00
parent 0bbea86a6a
commit d1243be71d
10 changed files with 602 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
codex-rs/app-server/src/request_processors.rs
View File

@@ -504,6 +504,7 @@ mod config_processor;
mod environment_processor;
mod external_agent_config_processor;
mod external_agent_session_import;
mod external_agent_session_rollout;
mod feedback_doctor_report;
mod feedback_processor;
mod fs_processor;

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

@@ -1,6 +1,9 @@
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use chrono::DateTime;
use chrono::SecondsFormat;
use chrono::Utc;
use codex_arg0::Arg0DispatchPaths;
use codex_core::ThreadManager;
@@ -15,28 +18,33 @@ use codex_models_manager::manager::RefreshStrategy;
use codex_protocol::ThreadId;
use codex_protocol::models::BaseInstructions;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::SessionContextWindow;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::ThreadHistoryMode;
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::DeleteThreadParams;
use codex_thread_store::LocalThreadStore;
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;
use crate::config::external_agent_config::record_import_error;
use crate::config_manager::ConfigManager;
use super::external_agent_session_rollout::materialize_imported_rollout;
const SESSION_IMPORT_CONCURRENCY: usize = 5;
#[derive(Clone)]
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 +62,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 +143,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,
}))
@@ -164,10 +195,18 @@ impl ExternalAgentSessionImporter {
&self,
session: ImportedExternalAgentSession,
) -> Result<ThreadId, String> {
if !self.thread_store.as_any().is::<LocalThreadStore>() {
return Err(
"external agent session import requires the local thread store".to_string(),
);
}
let ImportedExternalAgentSession {
cwd,
title,
first_user_message,
source_session_id: _,
created_at,
updated_at,
mut rollout_items,
} = session;
let config = self
@@ -204,42 +243,63 @@ impl ExternalAgentSessionImporter {
ThreadMemoryMode::Disabled
};
let now = Utc::now();
let create_params = CreateThreadParams {
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 base_instructions = BaseInstructions {
text: config
.base_instructions
.clone()
.unwrap_or_else(|| model_info.get_model_instructions(config.personality)),
};
let session_meta = SessionMeta {
session_id: thread_id.into(),
thread_id,
extra_config: None,
id: thread_id,
forked_from_id: None,
parent_thread_id: None,
timestamp: created_at.to_rfc3339_opts(SecondsFormat::Millis, true),
cwd: cwd.clone(),
originator: codex_login::default_client::originator().value,
cli_version: env!("CARGO_PKG_VERSION").to_string(),
agent_nickname: source.get_nickname(),
agent_role: source.get_agent_role(),
agent_path: source.get_agent_path().map(Into::into),
source: source.clone(),
thread_source: None,
originator: codex_login::default_client::originator().value,
base_instructions: BaseInstructions {
text: config
.base_instructions
.clone()
.unwrap_or_else(|| model_info.get_model_instructions(config.personality)),
},
dynamic_tools: Vec::new(),
model_provider: Some(model_provider.clone()),
base_instructions: Some(base_instructions),
dynamic_tools: None,
selected_capability_roots: Vec::new(),
multi_agent_version: Some(MultiAgentVersion::V1),
memory_mode: matches!(memory_mode, ThreadMemoryMode::Disabled)
.then_some("disabled".to_string()),
history_mode: ThreadHistoryMode::Legacy,
initial_window_id: uuid::Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
cwd: Some(cwd.clone()),
model_provider: model_provider.clone(),
memory_mode,
},
multi_agent_version: Some(MultiAgentVersion::V1),
context_window: Some(SessionContextWindow::new(uuid::Uuid::now_v7().to_string())),
};
rollout_items.retain(is_persisted_rollout_item);
let rollout_path = materialize_imported_rollout(
self.codex_home.as_path(),
created_at,
updated_at,
session_meta,
rollout_items,
)
.await
.map_err(|err| format!("failed to materialize imported session: {err}"))?;
let title = title
.as_deref()
.and_then(codex_core::util::normalize_thread_name);
let metadata = ThreadMetadataPatch {
rollout_path: Some(rollout_path.clone()),
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()),
@@ -252,39 +312,22 @@ impl ExternalAgentSessionImporter {
..Default::default()
};
self.thread_store
.create_thread(create_params)
.await
.map_err(|err| format!("failed to import session: {err}"))?;
if !rollout_items.is_empty()
&& let Err(err) = self
.thread_store
.append_items(AppendThreadItemsParams {
thread_id,
items: rollout_items,
})
.await
{
let _ = self.thread_store.discard_thread(thread_id).await;
return Err(format!("failed to import session: {err}"));
}
self.thread_store
if let Err(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
.map_err(|err| format!("failed to persist imported session: {err}"))?;
self.thread_store
.shutdown_thread(thread_id)
.await
.map_err(|err| format!("failed to shutdown imported session: {err}"))?;
{
let _ = self
.thread_store
.delete_thread(DeleteThreadParams { thread_id })
.await;
let _ = std::fs::remove_file(rollout_path);
return Err(format!("failed to index imported session: {err}"));
}
Ok(thread_id)
}
}

124
codex-rs/app-server/src/request_processors/external_agent_session_rollout.rs Normal file
View File

@@ -0,0 +1,124 @@
use std::fs::FileTimes;
use std::fs::OpenOptions;
use std::io;
use std::io::BufWriter;
use std::io::Write;
use std::path::Path;
use std::path::PathBuf;
use chrono::DateTime;
use chrono::SecondsFormat;
use chrono::Utc;
use codex_git_utils::collect_git_info;
use codex_protocol::protocol::GitInfo as ProtocolGitInfo;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_rollout::SESSIONS_SUBDIR;
pub(super) async fn materialize_imported_rollout(
codex_home: &Path,
created_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
session_meta: SessionMeta,
rollout_items: Vec<RolloutItem>,
) -> io::Result<PathBuf> {
let git = collect_git_info(session_meta.cwd.as_path())
.await
.map(|info| ProtocolGitInfo {
commit_hash: info.commit_hash,
branch: info.branch,
repository_url: info.repository_url,
});
let codex_home = codex_home.to_path_buf();
tokio::task::spawn_blocking(move || {
write_imported_rollout(
codex_home.as_path(),
created_at,
updated_at,
SessionMetaLine {
meta: session_meta,
git,
},
rollout_items,
)
})
.await
.map_err(|err| io::Error::other(format!("imported rollout task failed: {err}")))?
}
fn write_imported_rollout(
codex_home: &Path,
created_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
session_meta: SessionMetaLine,
rollout_items: Vec<RolloutItem>,
) -> io::Result<PathBuf> {
let rollout_dir = codex_home
.join(SESSIONS_SUBDIR)
.join(created_at.format("%Y").to_string())
.join(created_at.format("%m").to_string())
.join(created_at.format("%d").to_string());
std::fs::create_dir_all(rollout_dir.as_path())?;
let filename_timestamp = created_at.format("%Y-%m-%dT%H-%M-%S");
let rollout_path = rollout_dir.join(format!(
"rollout-{filename_timestamp}-{}.jsonl",
session_meta.meta.id
));
if rollout_path.exists() {
return Err(io::Error::new(
io::ErrorKind::AlreadyExists,
format!(
"imported rollout already exists: {}",
rollout_path.display()
),
));
}
let temporary_path = rollout_path.with_extension("jsonl.tmp");
let mut renamed = false;
let result = (|| {
let file = OpenOptions::new()
.create_new(true)
.write(true)
.open(temporary_path.as_path())?;
let mut writer = BufWriter::new(file);
write_rollout_line(
&mut writer,
created_at.to_rfc3339_opts(SecondsFormat::Millis, true),
RolloutItem::SessionMeta(session_meta),
)?;
let item_timestamp = updated_at.to_rfc3339_opts(SecondsFormat::Millis, true);
for item in rollout_items {
write_rollout_line(&mut writer, item_timestamp.clone(), item)?;
}
writer.flush()?;
writer.get_ref().sync_all()?;
drop(writer);
std::fs::rename(temporary_path.as_path(), rollout_path.as_path())?;
renamed = true;
OpenOptions::new()
.write(true)
.open(rollout_path.as_path())?
.set_times(FileTimes::new().set_modified(updated_at.into()))?;
Ok(())
})();
if let Err(err) = result {
let _ = std::fs::remove_file(temporary_path);
if renamed {
let _ = std::fs::remove_file(rollout_path);
}
return Err(err);
}
Ok(rollout_path)
}
fn write_rollout_line(
writer: &mut BufWriter<std::fs::File>,
timestamp: String,
item: RolloutItem,
) -> io::Result<()> {
serde_json::to_writer(writer.by_ref(), &RolloutLine { timestamp, item })
.map_err(io::Error::other)?;
writer.write_all(b"\n")
}

53
codex-rs/app-server/tests/suite/v2/external_agent_config.rs
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,40 @@ 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();
let rollout_path = codex_rollout::find_thread_path_by_id_str(
codex_home.path(),
&imported_thread_id,
/*state_db_ctx*/ None,
)
.await?
.expect("imported rollout path");
assert_eq!(
rollout_path.strip_prefix(codex_home.path())?,
PathBuf::from(format!(
"sessions/2026/06/03/rollout-2026-06-03T12-00-00-{imported_thread_id}.jsonl"
))
);
assert_eq!(
chrono::DateTime::<chrono::Utc>::from(std::fs::metadata(&rollout_path)?.modified()?)
.timestamp(),
source_updated_at
);
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 +794,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!(

44
codex-rs/external-agent-sessions/src/detect.rs
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()
);
}

123
codex-rs/external-agent-sessions/src/export.rs
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,

71
codex-rs/external-agent-sessions/src/ledger.rs
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> {

79
codex-rs/external-agent-sessions/src/ledger_tests.rs
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, /*source_session_id*/ 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")
);
}

44
codex-rs/external-agent-sessions/src/lib.rs
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 {

25
codex-rs/external-agent-sessions/src/records.rs
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!(