Sync updates to imported external agent sessions (#36356)

## Why

External agent session files can gain messages after their initial import. Re-importing those files should extend the existing Codex thread instead of creating a duplicate.

## What changed

- Map a changed source session back to its uniquely imported thread and append only the missing transcript suffix.
- Update the import ledger after verifying that the source and destination transcripts match.
- Defer the update when the target is active, archived, ambiguous, diverged, or otherwise unsafe to modify.

## Testing

Added unit and app-server coverage for suffix planning, ledger checkpointing, concurrent updates, and unsafe targets that must be deferred.

GitOrigin-RevId: 3d9e71cd66e8b31cf5128e8869063868bfb3eb05

codex-rs/external-agent-migration/src/sessions/append.rs

@@ -0,0 +1,316 @@
+use std::path::Path;
+
+use super::export::EXTERNAL_SESSION_IMPORTED_MARKER;
+use super::ledger::checkpoint_existing_session_import;
+use codex_core::ThreadManager;
+use codex_protocol::ThreadId;
+use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::ThreadMemoryMode;
+use codex_thread_store::AppendThreadItemsParams;
+use codex_thread_store::ReadThreadParams;
+use codex_thread_store::ResumeThreadParams;
+use codex_thread_store::ThreadPersistenceMetadata;
+use codex_thread_store::ThreadStore;
+use tokio::sync::Semaphore;
+
+/// A changed external session and the existing native thread it may extend.
+pub struct ExistingSessionAppend<'a> {
+    pub source_path: &'a Path,
+    pub source_content_sha256: &'a str,
+    pub expected_source_content_sha256: &'a str,
+    pub thread_id: ThreadId,
+    pub source_items: &'a [RolloutItem],
+}
+
+/// Appends an exact missing suffix to one cold imported thread and checkpoints the source.
+///
+/// Any unavailable, active, archived, malformed, or diverged destination fails closed.
+pub async fn append_existing_session(
+    codex_home: &Path,
+    checkpoint_permits: &Semaphore,
+    thread_manager: &ThreadManager,
+    thread_store: &dyn ThreadStore,
+    request: ExistingSessionAppend<'_>,
+) -> bool {
+    let ExistingSessionAppend {
+        source_path,
+        source_content_sha256,
+        expected_source_content_sha256,
+        thread_id,
+        source_items,
+    } = request;
+    let Ok(mut thread) = thread_store
+        .read_thread(ReadThreadParams {
+            thread_id,
+            include_archived: true,
+            include_history: true,
+        })
+        .await
+    else {
+        return false;
+    };
+    if thread.thread_id != thread_id || thread.archived_at.is_some() {
+        return false;
+    }
+    let Some(history) = thread.history.take() else {
+        return false;
+    };
+    if history.thread_id != thread_id || thread_manager.get_thread(thread_id).await.is_ok() {
+        return false;
+    }
+    let Some(metadata) = persistence_metadata(&history.items, thread_id) else {
+        return false;
+    };
+    let Some(rollout_path) = thread.rollout_path else {
+        return false;
+    };
+    if plan_append(source_items, &history.items).is_none() {
+        return false;
+    }
+    if thread_store
+        .resume_thread(ResumeThreadParams {
+            thread_id,
+            rollout_path: Some(rollout_path),
+            history: None,
+            include_archived: false,
+            metadata,
+        })
+        .await
+        .is_err()
+    {
+        return false;
+    }
+
+    let fresh_items = match thread_store
+        .read_thread(ReadThreadParams {
+            thread_id,
+            include_archived: true,
+            include_history: true,
+        })
+        .await
+    {
+        Ok(mut thread) if thread.thread_id == thread_id && thread.archived_at.is_none() => thread
+            .history
+            .take()
+            .filter(|history| history.thread_id == thread_id)
+            .filter(|history| persistence_metadata(&history.items, thread_id).is_some())
+            .and_then(|history| plan_append(source_items, &history.items)),
+        Ok(_) | Err(_) => None,
+    };
+    let Some(items) = fresh_items else {
+        let _ = thread_store.discard_thread(thread_id).await;
+        return false;
+    };
+    if thread_store
+        .append_items(AppendThreadItemsParams { thread_id, items })
+        .await
+        .is_err()
+    {
+        let _ = thread_store.discard_thread(thread_id).await;
+        return false;
+    }
+    if thread_store.shutdown_thread(thread_id).await.is_err() {
+        let _ = thread_store.discard_thread(thread_id).await;
+        return false;
+    }
+
+    let Ok(_checkpoint_permit) = checkpoint_permits.acquire().await else {
+        return false;
+    };
+    let target_matches_source = match thread_store
+        .read_thread(ReadThreadParams {
+            thread_id,
+            include_archived: true,
+            include_history: true,
+        })
+        .await
+    {
+        Ok(mut thread) if thread.thread_id == thread_id && thread.archived_at.is_none() => thread
+            .history
+            .take()
+            .filter(|history| history.thread_id == thread_id)
+            .filter(|history| persistence_metadata(&history.items, thread_id).is_some())
+            .is_some_and(|history| model_transcripts_match(source_items, &history.items)),
+        Ok(_) | Err(_) => false,
+    };
+    if !target_matches_source {
+        return false;
+    }
+
+    let codex_home = codex_home.to_path_buf();
+    let source_path = source_path.to_path_buf();
+    let expected_source_content_sha256 = expected_source_content_sha256.to_string();
+    let source_content_sha256 = source_content_sha256.to_string();
+    matches!(
+        tokio::task::spawn_blocking(move || {
+            checkpoint_existing_session_import(
+                &codex_home,
+                &source_path,
+                thread_id,
+                &expected_source_content_sha256,
+                &source_content_sha256,
+            )
+        })
+        .await,
+        Ok(Ok(true))
+    )
+}
+
+struct SourceModelItem<'a> {
+    response_item: &'a ResponseItem,
+    append_start_index: usize,
+}
+
+/// Returns the nonempty source suffix that can be safely appended to `history_items`.
+///
+/// The destination's complete model-visible transcript must be an exact prefix of the
+/// source transcript. Rollout metadata does not participate in transcript identity.
+fn plan_append(
+    source_items: &[RolloutItem],
+    history_items: &[RolloutItem],
+) -> Option<Vec<RolloutItem>> {
+    let source = source_model_items(source_items)?;
+    let history = history_model_items(history_items)?;
+    if history.len() >= source.len()
+        || history
+            .iter()
+            .zip(&source)
+            .any(|(history, source)| *history != source.response_item)
+    {
+        return None;
+    }
+
+    let append_start_index = source.get(history.len())?.append_start_index;
+    let suffix = source_items[append_start_index..]
+        .iter()
+        .filter(|item| !is_import_marker(item))
+        .cloned()
+        .collect::<Vec<_>>();
+    suffix
+        .iter()
+        .any(|item| matches!(item, RolloutItem::ResponseItem(_)))
+        .then_some(suffix)
+}
+
+/// Returns whether source and destination contain the same complete model-visible transcript.
+///
+/// Rollout metadata does not participate in transcript identity. Unsupported history shapes fail
+/// closed.
+fn model_transcripts_match(source_items: &[RolloutItem], history_items: &[RolloutItem]) -> bool {
+    let Some(source) = source_model_items(source_items) else {
+        return false;
+    };
+    let Some(history) = history_model_items(history_items) else {
+        return false;
+    };
+    history.len() == source.len()
+        && history
+            .iter()
+            .zip(&source)
+            .all(|(history, source)| *history == source.response_item)
+}
+
+fn source_model_items(items: &[RolloutItem]) -> Option<Vec<SourceModelItem<'_>>> {
+    let mut model_items = Vec::new();
+    let mut append_start_index = None;
+    for (index, item) in items.iter().enumerate() {
+        match item {
+            RolloutItem::SessionMeta(_) | RolloutItem::InterAgentCommunicationMetadata { .. } => {}
+            RolloutItem::ResponseItem(response_item) => {
+                model_items.push(SourceModelItem {
+                    response_item,
+                    append_start_index: append_start_index.take().unwrap_or(index),
+                });
+            }
+            RolloutItem::EventMsg(EventMsg::TurnStarted(_)) => {
+                append_start_index = Some(index);
+            }
+            RolloutItem::EventMsg(EventMsg::UserMessage(_)) => {
+                append_start_index.get_or_insert(index);
+            }
+            RolloutItem::EventMsg(EventMsg::AgentMessage(event))
+                if event.message != EXTERNAL_SESSION_IMPORTED_MARKER =>
+            {
+                append_start_index = Some(index);
+            }
+            RolloutItem::EventMsg(
+                EventMsg::ContextCompacted(_) | EventMsg::ThreadRolledBack(_),
+            )
+            | RolloutItem::InterAgentCommunication(_)
+            | RolloutItem::Compacted(_)
+            | RolloutItem::TurnContext(_)
+            | RolloutItem::WorldState(_) => return None,
+            RolloutItem::EventMsg(_) => {}
+        }
+    }
+    Some(model_items)
+}
+
+fn history_model_items(items: &[RolloutItem]) -> Option<Vec<&ResponseItem>> {
+    let mut model_items = Vec::new();
+    for item in items {
+        match item {
+            RolloutItem::SessionMeta(_) | RolloutItem::InterAgentCommunicationMetadata { .. } => {}
+            RolloutItem::ResponseItem(response_item) => model_items.push(response_item),
+            RolloutItem::EventMsg(
+                EventMsg::ContextCompacted(_) | EventMsg::ThreadRolledBack(_),
+            )
+            | RolloutItem::InterAgentCommunication(_)
+            | RolloutItem::Compacted(_)
+            | RolloutItem::TurnContext(_)
+            | RolloutItem::WorldState(_) => return None,
+            RolloutItem::EventMsg(_) => {}
+        }
+    }
+    Some(model_items)
+}
+
+fn is_import_marker(item: &RolloutItem) -> bool {
+    matches!(
+        item,
+        RolloutItem::EventMsg(EventMsg::AgentMessage(event))
+            if event.message == EXTERNAL_SESSION_IMPORTED_MARKER
+    )
+}
+
+fn persistence_metadata(
+    history_items: &[RolloutItem],
+    thread_id: ThreadId,
+) -> Option<ThreadPersistenceMetadata> {
+    let RolloutItem::SessionMeta(first_meta_line) = history_items.first()? else {
+        return None;
+    };
+    if first_meta_line.meta.id != thread_id {
+        return None;
+    }
+    let meta = history_items.iter().rev().find_map(|item| match item {
+        RolloutItem::SessionMeta(meta_line) if meta_line.meta.id == thread_id => {
+            Some(&meta_line.meta)
+        }
+        _ => None,
+    })?;
+    if meta.cwd.as_os_str().is_empty()
+        || meta
+            .model_provider
+            .as_deref()
+            .is_none_or(|provider| provider.trim().is_empty())
+    {
+        return None;
+    }
+    let memory_mode = match meta.memory_mode.as_deref() {
+        None | Some("enabled") => ThreadMemoryMode::Enabled,
+        Some("disabled") => ThreadMemoryMode::Disabled,
+        Some(_) => return None,
+    };
+    Some(ThreadPersistenceMetadata {
+        cwd: Some(meta.cwd.clone()),
+        model_provider: meta.model_provider.clone()?,
+        memory_mode,
+    })
+}
+
+#[cfg(test)]
+#[path = "append_tests.rs"]
+mod tests;

codex-rs/external-agent-migration/src/sessions/append_tests.rs

@@ -0,0 +1,122 @@
+use super::super::ConversationMessage;
+use super::super::MessageRole;
+use super::super::export::EXTERNAL_SESSION_IMPORTED_MARKER;
+use super::super::export::rollout_items_from_messages;
+use super::*;
+use codex_protocol::models::ContentItem;
+use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::ContextCompactedEvent;
+use codex_protocol::protocol::ThreadRolledBackEvent;
+use pretty_assertions::assert_eq;
+
+#[test]
+fn returns_the_missing_suffix_from_its_visible_boundary() {
+    let history = rollout(&[(MessageRole::User, "first request")]);
+    let source = rollout(&[
+        (MessageRole::User, "first request"),
+        (MessageRole::Assistant, "late answer"),
+        (MessageRole::User, "follow-up request"),
+    ]);
+
+    let suffix = plan_append(&source, &history).expect("exact prefix should append");
+
+    assert!(matches!(
+        suffix.first(),
+        Some(RolloutItem::EventMsg(EventMsg::AgentMessage(event)))
+            if event.message == "late answer"
+    ));
+    assert_eq!(
+        model_messages(&suffix),
+        vec![
+            (MessageRole::Assistant, "late answer"),
+            (MessageRole::User, "follow-up request"),
+        ]
+    );
+    assert!(!suffix.iter().any(|item| matches!(
+        item,
+        RolloutItem::EventMsg(EventMsg::AgentMessage(event))
+            if event.message == EXTERNAL_SESSION_IMPORTED_MARKER
+    )));
+}
+
+#[test]
+fn requires_a_strict_nonempty_model_prefix() {
+    let history = rollout(&[(MessageRole::User, "first request")]);
+    let source = rollout(&[
+        (MessageRole::User, "first request"),
+        (MessageRole::User, "follow-up request"),
+    ]);
+    assert!(plan_append(&history, &history).is_none());
+    assert!(
+        plan_append(
+            &source,
+            &rollout(&[(MessageRole::User, "rewritten request")])
+        )
+        .is_none()
+    );
+
+    let mut metadata_changed = history.clone();
+    for item in &mut metadata_changed {
+        if let RolloutItem::EventMsg(EventMsg::TurnStarted(event)) = item {
+            event.turn_id = "different-turn-id".to_string();
+            event.started_at = Some(9_999);
+        }
+    }
+    assert!(model_transcripts_match(&history, &metadata_changed));
+    assert!(!model_transcripts_match(&source, &history));
+    assert!(plan_append(&source, &metadata_changed).is_some());
+
+    for event in [
+        EventMsg::ContextCompacted(ContextCompactedEvent),
+        EventMsg::ThreadRolledBack(ThreadRolledBackEvent { num_turns: 1 }),
+    ] {
+        let mut rewritten = history.clone();
+        rewritten.push(RolloutItem::EventMsg(event));
+        assert!(plan_append(&source, &rewritten).is_none());
+    }
+    let mut with_tool_call = history;
+    with_tool_call.push(RolloutItem::ResponseItem(ResponseItem::FunctionCall {
+        id: None,
+        name: "native_tool".to_string(),
+        namespace: None,
+        arguments: "{}".to_string(),
+        encrypted_function_args: None,
+        call_id: "native-call".to_string(),
+        internal_chat_message_metadata_passthrough: None,
+    }));
+    assert!(plan_append(&source, &with_tool_call).is_none());
+}
+
+fn rollout(messages: &[(MessageRole, &str)]) -> Vec<RolloutItem> {
+    rollout_items_from_messages(
+        messages
+            .iter()
+            .enumerate()
+            .map(|(index, &(role, text))| ConversationMessage {
+                role,
+                text: text.to_string(),
+                timestamp: Some(index as i64),
+            })
+            .collect(),
+    )
+}
+
+fn model_messages(items: &[RolloutItem]) -> Vec<(MessageRole, &str)> {
+    items
+        .iter()
+        .filter_map(|item| match item {
+            RolloutItem::ResponseItem(ResponseItem::Message { role, content, .. }) => {
+                match (role.as_str(), content.as_slice()) {
+                    ("user", [ContentItem::InputText { text }]) => {
+                        Some((MessageRole::User, text.as_str()))
+                    }
+                    ("assistant", [ContentItem::OutputText { text }]) => {
+                        Some((MessageRole::Assistant, text.as_str()))
+                    }
+                    _ => None,
+                }
+            }
+            _ => None,
+        })
+        .collect()
+}

codex-rs/external-agent-migration/src/sessions/export.rs

@@ -24,7 +24,7 @@ use std::collections::BTreeSet;
 use std::io;
 use std::path::Path;
 
-const EXTERNAL_SESSION_IMPORTED_MARKER: &str = "<EXTERNAL SESSION IMPORTED>";
+pub(super) const EXTERNAL_SESSION_IMPORTED_MARKER: &str = "<EXTERNAL SESSION IMPORTED>";
 
 #[cfg(test)]
 fn load_session_for_import(path: &Path) -> io::Result<Option<ImportedExternalAgentSession>> {
@@ -82,7 +82,7 @@ pub(crate) fn load_session_for_import_with_content_sha256(
     )))
 }
 
-fn rollout_items_from_messages(messages: Vec<ConversationMessage>) -> Vec<RolloutItem> {
+pub(super) fn rollout_items_from_messages(messages: Vec<ConversationMessage>) -> Vec<RolloutItem> {
     let mut items = Vec::new();
     let mut current_turn = None;
     let mut response_item_bytes = 0i64;

codex-rs/external-agent-migration/src/sessions/ledger.rs

@@ -44,6 +44,16 @@ pub struct CompletedExternalAgentSessionImport {
     pub title: Option<String>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum SessionImportSourceMapping {
+    None,
+    Unique {
+        source_content_sha256: String,
+        imported_thread_id: ThreadId,
+    },
+    Ambiguous,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ImportedConnectorCandidate {
     pub name: String,
@@ -82,6 +92,71 @@ pub(crate) fn record_imported_session(
     )
 }
 
+pub(crate) fn find_existing_session_import(
+    codex_home: &Path,
+    source_path: &Path,
+) -> io::Result<SessionImportSourceMapping> {
+    let source_path = canonical_source_path(source_path)?;
+    let ledger = load_import_ledger(codex_home)?;
+    let mut matching_records = ledger
+        .records
+        .iter()
+        .filter(|record| record.source_path == source_path);
+    let Some(record) = matching_records.next() else {
+        return Ok(SessionImportSourceMapping::None);
+    };
+    if matching_records.next().is_some() {
+        return Ok(SessionImportSourceMapping::Ambiguous);
+    }
+    Ok(SessionImportSourceMapping::Unique {
+        source_content_sha256: record.content_sha256.clone(),
+        imported_thread_id: record.imported_thread_id,
+    })
+}
+
+pub(super) fn checkpoint_existing_session_import(
+    codex_home: &Path,
+    source_path: &Path,
+    thread_id: ThreadId,
+    expected_source_content_sha256: &str,
+    source_content_sha256: &str,
+) -> io::Result<bool> {
+    if expected_source_content_sha256 == source_content_sha256 {
+        return Ok(false);
+    }
+    let source_path = canonical_source_path(source_path)?;
+    let source_modified_at = session_modified_at(&source_path)?;
+    if session_content_sha256(&source_path)? != source_content_sha256
+        || source_modified_at != session_modified_at(&source_path)?
+    {
+        return Ok(false);
+    }
+
+    let mut ledger = load_import_ledger(codex_home)?;
+    let mut matching_indices = ledger
+        .records
+        .iter()
+        .enumerate()
+        .filter_map(|(index, record)| (record.source_path == source_path).then_some(index));
+    let Some(index) = matching_indices.next() else {
+        return Ok(false);
+    };
+    if matching_indices.next().is_some() {
+        return Ok(false);
+    }
+    let record = &mut ledger.records[index];
+    if record.imported_thread_id != thread_id
+        || record.content_sha256 != expected_source_content_sha256
+    {
+        return Ok(false);
+    }
+    record.content_sha256 = source_content_sha256.to_string();
+    record.imported_at = now_unix_seconds();
+    record.source_modified_at = source_modified_at;
+    save_import_ledger(codex_home, &ledger)?;
+    Ok(true)
+}
+
 pub fn record_completed_session_imports(
     codex_home: &Path,
     imports: Vec<CompletedExternalAgentSessionImport>,

codex-rs/external-agent-migration/src/sessions/ledger_tests.rs

@@ -1,11 +1,14 @@
 use super::CompletedExternalAgentSessionImport;
 use super::ImportedConnectorCandidate;
 use super::ImportedExternalAgentSessionLedger;
+use super::checkpoint_existing_session_import;
 use super::read_imported_connector_candidates;
 use super::record_completed_session_imports;
 use codex_protocol::ThreadId;
+use pretty_assertions::assert_eq;
 use sha2::Digest;
 use sha2::Sha256;
+use std::path::Path;
 use tempfile::TempDir;
 
 #[test]
@@ -95,6 +98,75 @@ fn completed_import_refreshes_existing_record_metadata() {
     assert_eq!(ledger.records[0].title.as_deref(), Some("Second title"));
 }
 
+#[test]
+fn checkpoint_is_compare_and_swap_and_preserves_record_metadata() {
+    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, "old").expect("initial source");
+    let source_path = std::fs::canonicalize(source_path).expect("canonical source");
+    let thread_id = ThreadId::new();
+    let old_hash = format!("{:x}", Sha256::digest("old"));
+    let new_hash = format!("{:x}", Sha256::digest("new"));
+    record_completed_session_imports(
+        &codex_home,
+        vec![CompletedExternalAgentSessionImport {
+            source_path: source_path.clone(),
+            source_content_sha256: old_hash.clone(),
+            imported_thread_id: thread_id,
+            connector_names: vec!["Gmail".to_string()],
+            title: Some("Original title".to_string()),
+        }],
+    )
+    .expect("record initial import");
+    std::fs::write(&source_path, "new").expect("updated source");
+    let ledger_before = ledger_bytes(&codex_home);
+    for (candidate_thread_id, expected_hash, candidate_hash) in [
+        (ThreadId::new(), old_hash.as_str(), new_hash.as_str()),
+        (thread_id, "stale", new_hash.as_str()),
+        (thread_id, old_hash.as_str(), "not-the-source-hash"),
+    ] {
+        assert!(
+            !checkpoint_existing_session_import(
+                &codex_home,
+                &source_path,
+                candidate_thread_id,
+                expected_hash,
+                candidate_hash,
+            )
+            .expect("rejected checkpoint")
+        );
+    }
+    assert_eq!(ledger_bytes(&codex_home), ledger_before);
+
+    assert!(
+        checkpoint_existing_session_import(
+            &codex_home,
+            &source_path,
+            thread_id,
+            &old_hash,
+            &new_hash,
+        )
+        .expect("checkpoint")
+    );
+    let ledger = super::load_import_ledger(&codex_home).expect("updated ledger");
+    let [record] = ledger.records.as_slice() else {
+        panic!("checkpoint must preserve one record");
+    };
+    assert_eq!(
+        record,
+        &super::ImportedExternalAgentSessionRecord {
+            source_path: source_path.clone(),
+            content_sha256: new_hash,
+            imported_thread_id: thread_id,
+            imported_at: record.imported_at,
+            source_modified_at: super::session_modified_at(&source_path).expect("source mtime"),
+            connector_names: vec!["Gmail".to_string()],
+            title: Some("Original title".to_string()),
+        }
+    );
+}
+
 #[test]
 fn connector_candidates_use_latest_import_for_each_source() {
     let root = TempDir::new().expect("tempdir");
@@ -144,3 +216,7 @@ fn connector_candidates_use_latest_import_for_each_source() {
         ]
     );
 }
+
+fn ledger_bytes(codex_home: &Path) -> Vec<u8> {
+    std::fs::read(super::import_ledger_path(codex_home)).expect("ledger")
+}

codex-rs/external-agent-migration/src/sessions/mod.rs

@@ -1,5 +1,6 @@
 //! Parsing and export helpers for external-agent session histories.
 
+mod append;
 mod export;
 pub(crate) mod ledger;
 pub(crate) mod records_cla;
@@ -7,6 +8,7 @@ mod records_common;
 pub(crate) mod records_cur;
 mod title;
 
+use codex_protocol::ThreadId;
 use codex_protocol::protocol::RolloutItem;
 use std::collections::BTreeSet;
 use std::io;
@@ -17,6 +19,8 @@ pub use crate::detect::sessions::ImportedSessionConnectorAttribution;
 pub use crate::detect::sessions::detect_imported_cla_session_connectors;
 pub use crate::detect::sessions::detect_recent_cla_sessions;
 pub use crate::detect::sessions::detect_recent_cur_sessions;
+pub use append::ExistingSessionAppend;
+pub use append::append_existing_session;
 use export::load_session_for_import_with_content_sha256;
 pub use ledger::CompletedExternalAgentSessionImport;
 pub use ledger::ImportedConnectorCandidate;
@@ -77,10 +81,20 @@ pub struct ImportedExternalAgentSession {
     pub rollout_items: Vec<RolloutItem>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SessionImportTarget {
+    New,
+    Existing {
+        thread_id: ThreadId,
+        expected_source_content_sha256: String,
+    },
+}
+
 #[derive(Debug, Clone)]
 pub struct PendingSessionImport {
     pub source_path: PathBuf,
     pub source_content_sha256: String,
+    pub target: SessionImportTarget,
     pub attributed_mcp_server_ids: BTreeSet<String>,
     pub session: ImportedExternalAgentSession,
 }
@@ -101,11 +115,27 @@ pub fn prepare_validated_session_import_with_metadata_mode(
     session: ExternalAgentSessionMigration,
     metadata_mode: SessionMetadataMode,
 ) -> io::Result<Option<PendingSessionImport>> {
-    let has_been_imported = has_current_session_been_imported(codex_home, &session.path)?;
-    if has_been_imported {
+    let Some(mut pending) = load_importable_session(&session.path, &session.cwd, metadata_mode)?
+    else {
         return Ok(None);
-    }
-    load_importable_session(&session.path, &session.cwd, metadata_mode)
+    };
+    pending.target = match ledger::find_existing_session_import(codex_home, &pending.source_path)? {
+        ledger::SessionImportSourceMapping::None => SessionImportTarget::New,
+        ledger::SessionImportSourceMapping::Unique {
+            source_content_sha256,
+            imported_thread_id,
+        } => {
+            if source_content_sha256 == pending.source_content_sha256 {
+                return Ok(None);
+            }
+            SessionImportTarget::Existing {
+                thread_id: imported_thread_id,
+                expected_source_content_sha256: source_content_sha256,
+            }
+        }
+        ledger::SessionImportSourceMapping::Ambiguous => return Ok(None),
+    };
+    Ok(Some(pending))
 }
 
 fn load_importable_session(
@@ -129,6 +159,7 @@ fn load_importable_session(
         .then_some(PendingSessionImport {
             source_path,
             source_content_sha256,
+            target: SessionImportTarget::New,
             attributed_mcp_server_ids,
             session: imported_session,
         }))
@@ -174,6 +205,7 @@ pub(crate) fn now_unix_seconds() -> i64 {
 mod tests {
     use super::*;
     use codex_protocol::ThreadId;
+    use pretty_assertions::assert_eq;
     use sha2::Digest;
     use sha2::Sha256;
     use tempfile::TempDir;
@@ -183,7 +215,8 @@ 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");
+        std::fs::write(&source_path, session_record(root.path(), "first request"))
+            .expect("session");
         ledger::record_imported_session(&codex_home, &source_path, ThreadId::new())
             .expect("record import");
 
@@ -194,6 +227,62 @@ mod tests {
         assert!(pending.is_none());
     }
 
+    #[test]
+    fn prepares_changed_session_for_its_unique_import() {
+        let root = TempDir::new().expect("tempdir");
+        let codex_home = root.path().join("codex-home");
+        let source_path = root.path().join("session.jsonl");
+        let initial = session_record(root.path(), "first request");
+        let expected_source_content_sha256 = format!("{:x}", Sha256::digest(&initial));
+        std::fs::write(&source_path, initial).expect("initial session");
+        let imported_thread_id = ThreadId::new();
+        ledger::record_imported_session(&codex_home, &source_path, imported_thread_id)
+            .expect("record import");
+        std::fs::write(&source_path, session_record(root.path(), "changed request"))
+            .expect("changed session");
+
+        let pending =
+            prepare_validated_session_import(&codex_home, session_migration(&source_path))
+                .expect("prepare changed session")
+                .expect("changed session should be eligible");
+
+        assert_eq!(
+            pending.target,
+            SessionImportTarget::Existing {
+                thread_id: imported_thread_id,
+                expected_source_content_sha256,
+            }
+        );
+    }
+
+    #[test]
+    fn skips_changed_session_with_ambiguous_imports() {
+        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, session_record(root.path(), "changed request"))
+            .expect("session");
+        let source_path = std::fs::canonicalize(source_path).expect("canonical source");
+        let import = |source_content_sha256: &str| ledger::CompletedExternalAgentSessionImport {
+            source_path: source_path.clone(),
+            source_content_sha256: source_content_sha256.to_string(),
+            imported_thread_id: ThreadId::new(),
+            connector_names: Vec::new(),
+            title: None,
+        };
+        ledger::record_completed_session_imports(
+            &codex_home,
+            vec![import("first hash"), import("second hash")],
+        )
+        .expect("record ambiguous imports");
+
+        let pending =
+            prepare_validated_session_import(&codex_home, session_migration(&source_path))
+                .expect("prepare ambiguous session");
+
+        assert!(pending.is_none());
+    }
+
     #[test]
     fn reports_session_preparation_errors() {
         let root = TempDir::new().expect("tempdir");
@@ -227,6 +316,7 @@ mod tests {
             pending.source_content_sha256,
             format!("{:x}", Sha256::digest(contents))
         );
+        assert_eq!(pending.target, SessionImportTarget::New);
     }
 
     #[test]
@@ -281,4 +371,14 @@ mod tests {
             title: None,
         }
     }
+
+    fn session_record(cwd: &Path, text: &str) -> String {
+        serde_json::json!({
+            "type": "user",
+            "cwd": cwd,
+            "timestamp": "2026-06-03T12:00:00Z",
+            "message": { "content": text },
+        })
+        .to_string()
+    }
 }