codex: carry source thread id in forked history

codex-rs/app-server/src/request_processors.rs

@@ -384,6 +384,7 @@ use codex_protocol::protocol::ConversationStartParams;
 use codex_protocol::protocol::ConversationStartTransport;
 use codex_protocol::protocol::ConversationTextParams;
 use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::ForkedHistory;
 #[cfg(test)]
 use codex_protocol::protocol::GitInfo as CoreGitInfo;
 use codex_protocol::protocol::InitialHistory;

codex-rs/app-server/src/request_processors/external_agent_config_processor.rs

@@ -35,6 +35,7 @@ use codex_external_agent_sessions::PendingSessionImport;
 use codex_external_agent_sessions::prepare_validated_session_imports;
 use codex_external_agent_sessions::record_imported_session;
 use codex_protocol::ThreadId;
+use codex_protocol::protocol::ForkedHistory;
 use codex_protocol::protocol::InitialHistory;
 use codex_thread_store::ThreadMetadataPatch;
 use std::collections::HashSet;
@@ -305,7 +306,10 @@ impl ExternalAgentConfigRequestProcessor {
             .thread_manager
             .start_thread_with_options(StartThreadOptions {
                 config,
-                initial_history: InitialHistory::Forked(rollout_items),
+                initial_history: InitialHistory::Forked(ForkedHistory {
+                    source_thread_id: None,
+                    history: rollout_items,
+                }),
                 session_source: None,
                 thread_source: None,
                 dynamic_tools: Vec::new(),

codex-rs/app-server/src/request_processors/thread_processor.rs

@@ -2501,7 +2501,7 @@ impl ThreadRequestProcessor {
         let include_turns = !exclude_turns;
 
         let (thread_history, resume_source_thread) = match if let Some(history) = history {
-            self.resume_thread_from_history(history.as_slice())
+            self.resume_thread_from_history(&thread_id, history.as_slice())
                 .await
                 .map(|thread_history| (thread_history, None))
         } else {
@@ -2928,18 +2928,20 @@ impl ThreadRequestProcessor {
 
     async fn resume_thread_from_history(
         &self,
+        thread_id: &str,
         history: &[ResponseItem],
     ) -> Result<InitialHistory, JSONRPCErrorError> {
         if history.is_empty() {
             return Err(invalid_request("history must not be empty"));
         }
-        Ok(InitialHistory::Forked(
-            history
+        Ok(InitialHistory::Forked(ForkedHistory {
+            source_thread_id: ThreadId::from_string(thread_id).ok(),
+            history: history
                 .iter()
                 .cloned()
                 .map(RolloutItem::ResponseItem)
                 .collect(),
-        ))
+        }))
     }
 
     async fn resume_thread_from_rollout(
@@ -3119,14 +3121,14 @@ impl ThreadRequestProcessor {
                     }
                 }
             }
-            InitialHistory::Forked(items) => {
+            InitialHistory::Forked(forked) => {
                 let mut thread = build_thread_from_snapshot(
                     thread_id,
                     session_id.clone(),
                     &config_snapshot,
                     Some(rollout_path.into()),
                 );
-                thread.preview = preview_from_rollout_items(items);
+                thread.preview = preview_from_rollout_items(&forked.history);
                 Ok(thread)
             }
             InitialHistory::New | InitialHistory::Cleared => Err(format!(

codex-rs/core/src/agent/control.rs

@@ -21,6 +21,7 @@ use codex_protocol::error::Result as CodexResult;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::MessagePhase;
 use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::ForkedHistory;
 use codex_protocol::protocol::InitialHistory;
 use codex_protocol::protocol::InterAgentCommunication;
 use codex_protocol::protocol::Op;
@@ -489,7 +490,10 @@ impl AgentControl {
         state
             .fork_thread_with_source(
                 config.clone(),
-                InitialHistory::Forked(forked_rollout_items),
+                InitialHistory::Forked(ForkedHistory {
+                    source_thread_id: Some(parent_thread_id),
+                    history: forked_rollout_items,
+                }),
                 self.clone(),
                 session_source,
                 /*thread_source*/ Some(ThreadSource::Subagent),

codex-rs/core/src/guardian/review_session.rs

@@ -17,6 +17,7 @@ use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
 use codex_protocol::protocol::AskForApproval;
 use codex_protocol::protocol::Event;
 use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::ForkedHistory;
 use codex_protocol::protocol::InitialHistory;
 use codex_protocol::protocol::Op;
 use codex_protocol::protocol::RolloutItem;
@@ -223,7 +224,10 @@ impl GuardianReviewSession {
                 let prior_review_count = state.prior_review_count;
                 let last_reviewed_transcript_cursor = state.last_reviewed_transcript_cursor;
                 state.last_committed_fork_snapshot = Some(GuardianReviewForkSnapshot {
-                    initial_history: InitialHistory::Forked(items),
+                    initial_history: InitialHistory::Forked(ForkedHistory {
+                        source_thread_id: Some(self.codex.session.conversation_id),
+                        history: items,
+                    }),
                     prior_review_count,
                     last_reviewed_transcript_cursor,
                 });
@@ -476,7 +480,7 @@ impl GuardianReviewSessionManager {
         let state = trunk.state.lock().await;
         let snapshot = state.last_committed_fork_snapshot.as_ref()?;
         match &snapshot.initial_history {
-            InitialHistory::Forked(items) => Some(items.clone()),
+            InitialHistory::Forked(forked) => Some(forked.history.clone()),
             InitialHistory::New | InitialHistory::Cleared | InitialHistory::Resumed(_) => None,
         }
     }

codex-rs/core/src/session/mod.rs

@@ -1230,20 +1230,20 @@ impl Session {
                     let _ = self.flush_rollout().await;
                 }
             }
-            InitialHistory::Forked(rollout_items) => {
-                self.apply_rollout_reconstruction(&turn_context, &rollout_items)
+            InitialHistory::Forked(forked) => {
+                self.apply_rollout_reconstruction(&turn_context, &forked.history)
                     .await;
 
                 // Seed usage info from the recorded rollout so UIs can show token counts
                 // immediately on resume/fork.
-                if let Some(info) = Self::last_token_info_from_rollout(&rollout_items) {
+                if let Some(info) = Self::last_token_info_from_rollout(&forked.history) {
                     let mut state = self.state.lock().await;
                     state.set_token_info(Some(info));
                 }
 
                 // If persisting, persist all rollout items as-is (the store filters).
-                if !rollout_items.is_empty() {
-                    self.persist_rollout_items(&rollout_items).await;
+                if !forked.history.is_empty() {
+                    self.persist_rollout_items(&forked.history).await;
                 }
 
                 // Forked threads should remain file-backed immediately after startup.

codex-rs/core/src/session/tests.rs

@@ -109,6 +109,7 @@ use codex_protocol::protocol::CodexErrorInfo;
 use codex_protocol::protocol::CompactedItem;
 use codex_protocol::protocol::ConversationAudioParams;
 use codex_protocol::protocol::CreditsSnapshot;
+use codex_protocol::protocol::ForkedHistory;
 use codex_protocol::protocol::GranularApprovalConfig;
 use codex_protocol::protocol::InitialHistory;
 use codex_protocol::protocol::InterAgentCommunication;
@@ -2256,7 +2257,10 @@ async fn record_initial_history_reconstructs_forked_transcript() {
     let (rollout_items, expected) = sample_rollout(&session, &turn_context).await;
 
     session
-        .record_initial_history(InitialHistory::Forked(rollout_items))
+        .record_initial_history(InitialHistory::Forked(ForkedHistory {
+            source_thread_id: None,
+            history: rollout_items,
+        }))
         .await;
 
     let history = session.state.lock().await.clone_history();
@@ -2513,7 +2517,10 @@ async fn record_initial_history_forked_hydrates_previous_turn_settings() {
     ];
 
     session
-        .record_initial_history(InitialHistory::Forked(rollout_items))
+        .record_initial_history(InitialHistory::Forked(ForkedHistory {
+            source_thread_id: None,
+            history: rollout_items,
+        }))
         .await;
 
     let history = session.clone_history().await;

codex-rs/core/src/thread_manager.rs

@@ -37,6 +37,7 @@ use codex_protocol::error::Result as CodexResult;
 use codex_protocol::openai_models::ModelPreset;
 use codex_protocol::protocol::Event;
 use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::ForkedHistory;
 use codex_protocol::protocol::InitialHistory;
 use codex_protocol::protocol::Op;
 use codex_protocol::protocol::ResumedHistory;
@@ -1439,6 +1440,7 @@ fn truncate_before_nth_user_message(
     n: usize,
     snapshot_state: &SnapshotTurnState,
 ) -> InitialHistory {
+    let source_thread_id = history.source_thread_id();
     let items: Vec<RolloutItem> = history.get_rollout_items();
     let user_positions = truncation::user_message_positions_in_rollout(&items);
     let rolled = if snapshot_state.ends_mid_turn && n >= user_positions.len() {
@@ -1457,7 +1459,10 @@ fn truncate_before_nth_user_message(
     if rolled.is_empty() {
         InitialHistory::New
     } else {
-        InitialHistory::Forked(rolled)
+        InitialHistory::Forked(ForkedHistory {
+            source_thread_id,
+            history: rolled,
+        })
     }
 }
 
@@ -1536,7 +1541,10 @@ fn fork_history_from_snapshot(
                 InitialHistory::New => InitialHistory::New,
                 InitialHistory::Cleared => InitialHistory::Cleared,
                 InitialHistory::Forked(history) => InitialHistory::Forked(history),
-                InitialHistory::Resumed(resumed) => InitialHistory::Forked(resumed.history),
+                InitialHistory::Resumed(resumed) => InitialHistory::Forked(ForkedHistory {
+                    source_thread_id: Some(resumed.conversation_id),
+                    history: resumed.history,
+                }),
             };
             if snapshot_state.ends_mid_turn {
                 append_interrupted_boundary(
@@ -1573,21 +1581,27 @@ fn append_interrupted_boundary(
                 history.push(RolloutItem::ResponseItem(marker));
             }
             history.push(aborted_event);
-            InitialHistory::Forked(history)
+            InitialHistory::Forked(ForkedHistory {
+                source_thread_id: None,
+                history,
+            })
         }
-        InitialHistory::Forked(mut history) => {
+        InitialHistory::Forked(mut forked) => {
             if let Some(marker) = interrupted_turn_history_marker(interrupted_marker) {
-                history.push(RolloutItem::ResponseItem(marker));
+                forked.history.push(RolloutItem::ResponseItem(marker));
             }
-            history.push(aborted_event);
-            InitialHistory::Forked(history)
+            forked.history.push(aborted_event);
+            InitialHistory::Forked(forked)
         }
         InitialHistory::Resumed(mut resumed) => {
             if let Some(marker) = interrupted_turn_history_marker(interrupted_marker) {
                 resumed.history.push(RolloutItem::ResponseItem(marker));
             }
             resumed.history.push(aborted_event);
-            InitialHistory::Forked(resumed.history)
+            InitialHistory::Forked(ForkedHistory {
+                source_thread_id: Some(resumed.conversation_id),
+                history: resumed.history,
+            })
         }
     }
 }

codex-rs/core/src/thread_manager_tests.rs

@@ -33,6 +33,13 @@ use wiremock::MockServer;
 
 const TEST_INSTALLATION_ID: &str = "11111111-1111-4111-8111-111111111111";
 
+fn forked_history(history: Vec<RolloutItem>) -> InitialHistory {
+    InitialHistory::Forked(ForkedHistory {
+        source_thread_id: None,
+        history,
+    })
+}
+
 fn user_msg(text: &str) -> ResponseItem {
     ResponseItem::Message {
         id: None,
@@ -96,7 +103,7 @@ fn truncates_before_requested_user_message() {
         .map(RolloutItem::ResponseItem)
         .collect();
     let truncated = truncate_before_nth_user_message(
-        InitialHistory::Forked(initial),
+        forked_history(initial),
         /*n*/ 1,
         &SnapshotTurnState {
             ends_mid_turn: false,
@@ -121,7 +128,7 @@ fn truncates_before_requested_user_message() {
         .map(RolloutItem::ResponseItem)
         .collect();
     let truncated2 = truncate_before_nth_user_message(
-        InitialHistory::Forked(initial2.clone()),
+        forked_history(initial2.clone()),
         /*n*/ 2,
         &SnapshotTurnState {
             ends_mid_turn: false,
@@ -145,7 +152,7 @@ fn out_of_range_truncation_drops_only_unfinished_suffix_mid_turn() {
     ];
 
     let truncated = truncate_before_nth_user_message(
-        InitialHistory::Forked(items.clone()),
+        forked_history(items.clone()),
         usize::MAX,
         &SnapshotTurnState {
             ends_mid_turn: true,
@@ -196,7 +203,7 @@ fn out_of_range_truncation_drops_pre_user_active_turn_prefix() {
         RolloutItem::ResponseItem(assistant_msg("partial")),
     ];
 
-    let snapshot_state = snapshot_turn_state(&InitialHistory::Forked(items.clone()));
+    let snapshot_state = snapshot_turn_state(&forked_history(items.clone()));
     assert_eq!(
         snapshot_state,
         SnapshotTurnState {
@@ -207,7 +214,7 @@ fn out_of_range_truncation_drops_pre_user_active_turn_prefix() {
     );
 
     let truncated = truncate_before_nth_user_message(
-        InitialHistory::Forked(items.clone()),
+        forked_history(items.clone()),
         usize::MAX,
         &snapshot_state,
     );
@@ -234,7 +241,7 @@ async fn ignores_session_prefix_messages_when_truncating() {
         .collect();
 
     let truncated = truncate_before_nth_user_message(
-        InitialHistory::Forked(rollout_items),
+        forked_history(rollout_items),
         /*n*/ 1,
         &SnapshotTurnState {
             ends_mid_turn: false,
@@ -982,8 +989,7 @@ async fn new_uses_active_provider_for_model_refresh() {
 
 #[test]
 fn interrupted_fork_snapshot_appends_interrupt_boundary() {
-    let committed_history =
-        InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello"))]);
+    let committed_history = forked_history(vec![RolloutItem::ResponseItem(user_msg("hello"))]);
 
     assert_eq!(
         serde_json::to_value(
@@ -1032,8 +1038,7 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() {
 
 #[test]
 fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
-    let committed_history =
-        InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello"))]);
+    let committed_history = forked_history(vec![RolloutItem::ResponseItem(user_msg("hello"))]);
 
     assert_eq!(
         serde_json::to_value(
@@ -1080,7 +1085,7 @@ fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() {
 
 #[test]
 fn interrupted_snapshot_is_not_mid_turn() {
-    let interrupted_history = InitialHistory::Forked(vec![
+    let interrupted_history = forked_history(vec![
         RolloutItem::ResponseItem(user_msg("hello")),
         RolloutItem::ResponseItem(assistant_msg("partial")),
         RolloutItem::ResponseItem(contextual_user_interrupted_marker()),
@@ -1122,7 +1127,7 @@ fn multi_agent_v2_interrupted_marker_uses_developer_input_message() {
 
 #[test]
 fn completed_legacy_event_history_is_not_mid_turn() {
-    let completed_history = InitialHistory::Forked(vec![
+    let completed_history = forked_history(vec![
         RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
             client_id: None,
             message: "hello".to_string(),
@@ -1150,7 +1155,7 @@ fn completed_legacy_event_history_is_not_mid_turn() {
 
 #[test]
 fn mixed_response_and_legacy_user_event_history_is_mid_turn() {
-    let mixed_history = InitialHistory::Forked(vec![
+    let mixed_history = forked_history(vec![
         RolloutItem::ResponseItem(user_msg("hello")),
         RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
             client_id: None,
@@ -1199,7 +1204,7 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor
     let source = manager
         .resume_thread_with_history(
             config.clone(),
-            InitialHistory::Forked(vec![
+            forked_history(vec![
                 RolloutItem::ResponseItem(user_msg("hello")),
                 RolloutItem::ResponseItem(assistant_msg("partial")),
             ]),
@@ -1307,7 +1312,7 @@ async fn interrupted_fork_snapshot_preserves_explicit_turn_id() {
     let source = manager
         .resume_thread_with_history(
             config.clone(),
-            InitialHistory::Forked(vec![
+            forked_history(vec![
                 RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
                     turn_id: "turn-explicit".to_string(),
                     trace_id: None,
@@ -1405,7 +1410,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_
     let source = manager
         .resume_thread_with_history(
             config.clone(),
-            InitialHistory::Forked(vec![
+            forked_history(vec![
                 RolloutItem::ResponseItem(user_msg("hello")),
                 RolloutItem::ResponseItem(assistant_msg("partial")),
             ]),
@@ -1548,7 +1553,7 @@ async fn resumed_thread_keeps_paused_goal_paused() -> anyhow::Result<()> {
     let source = manager
         .resume_thread_with_history(
             config.clone(),
-            InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("keep working"))]),
+            forked_history(vec![RolloutItem::ResponseItem(user_msg("keep working"))]),
             auth_manager.clone(),
             /*persist_extended_history*/ false,
             /*parent_trace*/ None,

codex-rs/core/src/tools/handlers/multi_agents_tests.rs

@@ -40,6 +40,7 @@ use codex_protocol::protocol::FileSystemAccessMode;
 use codex_protocol::protocol::FileSystemPath;
 use codex_protocol::protocol::FileSystemSandboxEntry;
 use codex_protocol::protocol::FileSystemSandboxPolicy;
+use codex_protocol::protocol::ForkedHistory;
 use codex_protocol::protocol::InitialHistory;
 use codex_protocol::protocol::InterAgentCommunication;
 use codex_protocol::protocol::NetworkSandboxPolicy;
@@ -2696,14 +2697,17 @@ async fn resume_agent_restores_closed_agent_and_accepts_send_input() {
     let thread = manager
         .resume_thread_with_history(
             config.clone(),
-            InitialHistory::Forked(vec![RolloutItem::ResponseItem(ResponseItem::Message {
-                id: None,
-                role: "user".to_string(),
-                content: vec![ContentItem::InputText {
-                    text: "materialized".to_string(),
-                }],
-                phase: None,
-            })]),
+            InitialHistory::Forked(ForkedHistory {
+                source_thread_id: None,
+                history: vec![RolloutItem::ResponseItem(ResponseItem::Message {
+                    id: None,
+                    role: "user".to_string(),
+                    content: vec![ContentItem::InputText {
+                        text: "materialized".to_string(),
+                    }],
+                    phase: None,
+                })],
+            }),
             AuthManager::from_auth_for_testing(CodexAuth::from_api_key("dummy")),
             /*persist_extended_history*/ false,
             /*parent_trace*/ None,

codex-rs/core/tests/suite/realtime_conversation.rs

@@ -15,6 +15,7 @@ use codex_protocol::protocol::ConversationStartTransport;
 use codex_protocol::protocol::ConversationTextParams;
 use codex_protocol::protocol::ErrorEvent;
 use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::ForkedHistory;
 use codex_protocol::protocol::InitialHistory;
 use codex_protocol::protocol::Op;
 use codex_protocol::protocol::RealtimeAudioFrame;
@@ -1884,7 +1885,10 @@ async fn conversation_startup_context_current_thread_selects_many_turns_by_budge
         .thread_manager
         .resume_thread_with_history(
             test.config.clone(),
-            InitialHistory::Forked(history),
+            InitialHistory::Forked(ForkedHistory {
+                source_thread_id: None,
+                history,
+            }),
             auth_manager_from_auth(CodexAuth::from_api_key("dummy")),
             /*persist_extended_history*/ false,
             /*parent_trace*/ None,

codex-rs/protocol/src/protocol.rs

@@ -2334,12 +2334,18 @@ pub struct ResumedHistory {
     pub rollout_path: Option<PathBuf>,
 }
 
+#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
+pub struct ForkedHistory {
+    pub source_thread_id: Option<ThreadId>,
+    pub history: Vec<RolloutItem>,
+}
+
 #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
 pub enum InitialHistory {
     New,
     Cleared,
     Resumed(ResumedHistory),
-    Forked(Vec<RolloutItem>),
+    Forked(ForkedHistory),
 }
 
 impl InitialHistory {
@@ -2347,7 +2353,7 @@ impl InitialHistory {
         match self {
             InitialHistory::New | InitialHistory::Cleared => false,
             InitialHistory::Resumed(resumed) => resumed.history.iter().any(&mut predicate),
-            InitialHistory::Forked(items) => items.iter().any(predicate),
+            InitialHistory::Forked(forked) => forked.history.iter().any(predicate),
         }
     }
 
@@ -2360,10 +2366,15 @@ impl InitialHistory {
                     _ => None,
                 })
             }
-            InitialHistory::Forked(items) => items.iter().find_map(|item| match item {
-                RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.id),
-                _ => None,
-            }),
+            InitialHistory::Forked(forked) => forked.source_thread_id,
+        }
+    }
+
+    pub fn source_thread_id(&self) -> Option<ThreadId> {
+        match self {
+            InitialHistory::New | InitialHistory::Cleared => None,
+            InitialHistory::Resumed(resumed) => Some(resumed.conversation_id),
+            InitialHistory::Forked(forked) => forked.source_thread_id,
         }
     }
 
@@ -2371,7 +2382,7 @@ impl InitialHistory {
         match self {
             InitialHistory::New | InitialHistory::Cleared => None,
             InitialHistory::Resumed(resumed) => session_cwd_from_items(&resumed.history),
-            InitialHistory::Forked(items) => session_cwd_from_items(items),
+            InitialHistory::Forked(forked) => session_cwd_from_items(&forked.history),
         }
     }
 
@@ -2379,7 +2390,7 @@ impl InitialHistory {
         match self {
             InitialHistory::New | InitialHistory::Cleared => Vec::new(),
             InitialHistory::Resumed(resumed) => resumed.history.clone(),
-            InitialHistory::Forked(items) => items.clone(),
+            InitialHistory::Forked(forked) => forked.history.clone(),
         }
     }
 
@@ -2396,8 +2407,9 @@ impl InitialHistory {
                     })
                     .collect(),
             ),
-            InitialHistory::Forked(items) => Some(
-                items
+            InitialHistory::Forked(forked) => Some(
+                forked
+                    .history
                     .iter()
                     .filter_map(|ri| match ri {
                         RolloutItem::EventMsg(ev) => Some(ev.clone()),
@@ -2418,7 +2430,7 @@ impl InitialHistory {
                     _ => None,
                 })
             }
-            InitialHistory::Forked(items) => items.iter().find_map(|item| match item {
+            InitialHistory::Forked(forked) => forked.history.iter().find_map(|item| match item {
                 RolloutItem::SessionMeta(meta_line) => meta_line.meta.base_instructions.clone(),
                 _ => None,
             }),
@@ -2434,7 +2446,7 @@ impl InitialHistory {
                     _ => None,
                 })
             }
-            InitialHistory::Forked(items) => items.iter().find_map(|item| match item {
+            InitialHistory::Forked(forked) => forked.history.iter().find_map(|item| match item {
                 RolloutItem::SessionMeta(meta_line) => meta_line.meta.dynamic_tools.clone(),
                 _ => None,
             }),