optimize codex resume and fork history

codex-rs/core/src/codex_delegate.rs

@@ -101,6 +101,7 @@ pub(crate) async fn run_codex_thread_interactive(
         conversation_history,
         session_source: SessionSource::SubAgent(subagent_source.clone()),
         forked_from_thread_id,
+        initial_rollout_copy: None,
         parent_thread_id: Some(parent_session.thread_id),
         thread_source: Some(ThreadSource::Subagent),
         agent_control: parent_session.services.agent_control.clone(),

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

@@ -423,6 +423,7 @@ pub(crate) struct CodexSpawnArgs {
     pub(crate) conversation_history: InitialHistory,
     pub(crate) session_source: SessionSource,
     pub(crate) forked_from_thread_id: Option<ThreadId>,
+    pub(crate) initial_rollout_copy: Option<PathBuf>,
     pub(crate) parent_thread_id: Option<ThreadId>,
     pub(crate) thread_source: Option<ThreadSource>,
     pub(crate) agent_control: AgentControl,
@@ -512,6 +513,7 @@ impl Codex {
             conversation_history,
             session_source,
             forked_from_thread_id,
+            initial_rollout_copy,
             parent_thread_id,
             thread_source,
             agent_control,
@@ -644,6 +646,7 @@ impl Codex {
             app_server_client_version: None,
             session_source,
             forked_from_thread_id,
+            initial_rollout_copy,
             parent_thread_id,
             thread_source,
             dynamic_tools,
@@ -1259,12 +1262,15 @@ impl Session {
     }
 
     async fn record_initial_history(&self, conversation_history: InitialHistory) {
-        let is_subagent = {
+        let (is_subagent, skip_fork_history_persistence) = {
             let state = self.state.lock().await;
-            state
-                .session_configuration
-                .session_source
-                .is_non_root_agent()
+            (
+                state
+                    .session_configuration
+                    .session_source
+                    .is_non_root_agent(),
+                state.session_configuration.initial_rollout_copy.is_some(),
+            )
         };
         let has_prior_user_turns = initial_history_has_prior_user_turns(&conversation_history);
         {
@@ -1331,7 +1337,12 @@ impl Session {
                 }
 
                 // If persisting, persist all rollout items as-is (the store filters).
-                if !rollout_items.is_empty() {
+                if skip_fork_history_persistence {
+                    let suffix = Self::fork_rollout_copy_suffix(&rollout_items);
+                    if !suffix.is_empty() {
+                        self.persist_rollout_items(suffix).await;
+                    }
+                } else if !rollout_items.is_empty() {
                     self.persist_rollout_items(&rollout_items).await;
                 }
 
@@ -1346,6 +1357,17 @@ impl Session {
         }
     }
 
+    fn fork_rollout_copy_suffix(rollout_items: &[RolloutItem]) -> &[RolloutItem] {
+        let Some(RolloutItem::EventMsg(EventMsg::TurnAborted(event))) = rollout_items.last() else {
+            return &[];
+        };
+        if event.reason == TurnAbortReason::Interrupted {
+            &rollout_items[rollout_items.len() - 1..]
+        } else {
+            &[]
+        }
+    }
+
     #[instrument(
         level = "trace",
         skip_all,

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

@@ -102,6 +102,8 @@ pub(crate) struct SessionConfiguration {
     pub(super) session_source: SessionSource,
     /// Immediate history source copied into this thread, when this thread was forked.
     pub(super) forked_from_thread_id: Option<ThreadId>,
+    /// Optional rollout JSONL copied into this thread after its own session metadata.
+    pub(super) initial_rollout_copy: Option<PathBuf>,
     /// Immediate control/spawn parent for this thread, when it has one.
     pub(super) parent_thread_id: Option<ThreadId>,
     /// Optional analytics source classification for this thread.
@@ -557,6 +559,9 @@ impl Session {
                             },
                             dynamic_tools: session_configuration.dynamic_tools.clone(),
                             multi_agent_version: initial_multi_agent_version,
+                            initial_rollout_copy: session_configuration
+                                .initial_rollout_copy
+                                .clone(),
                             metadata: ThreadPersistenceMetadata {
                                 cwd: Some(config.cwd.to_path_buf()),
                                 model_provider: config.model_provider_id.clone(),
@@ -570,10 +575,13 @@ impl Session {
                         LiveThread::create(Arc::clone(&thread_store), params).await?
                     }
                     InitialHistory::Resumed(resumed_history) => {
+                        let local_resume_with_path = resumed_history.rollout_path.is_some()
+                            && thread_store.as_any().is::<LocalThreadStore>();
                         let params = ResumeThreadParams {
                             thread_id: resumed_history.conversation_id,
                             rollout_path: resumed_history.rollout_path.clone(),
-                            history: Some(resumed_history.history.clone()),
+                            history: (!local_resume_with_path)
+                                .then(|| resumed_history.history.clone()),
                             include_archived: true,
                             metadata: ThreadPersistenceMetadata {
                                 cwd: Some(config.cwd.to_path_buf()),
@@ -585,7 +593,12 @@ impl Session {
                                 },
                             },
                         };
-                        LiveThread::resume(Arc::clone(&thread_store), params).await?
+                        LiveThread::resume_with_observed_history(
+                            Arc::clone(&thread_store),
+                            params,
+                            Some(resumed_history.history.as_slice()),
+                        )
+                        .await?
                     }
                 };
                 Ok(Some(live_thread))

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

@@ -3378,6 +3378,7 @@ async fn set_rate_limits_retains_previous_credits() {
         app_server_client_version: None,
         session_source: SessionSource::Exec,
         forked_from_thread_id: None,
+        initial_rollout_copy: None,
         parent_thread_id: None,
         thread_source: None,
         dynamic_tools: Vec::new(),
@@ -3485,6 +3486,7 @@ async fn set_rate_limits_updates_plan_type_when_present() {
         app_server_client_version: None,
         session_source: SessionSource::Exec,
         forked_from_thread_id: None,
+        initial_rollout_copy: None,
         parent_thread_id: None,
         thread_source: None,
         dynamic_tools: Vec::new(),
@@ -3738,6 +3740,7 @@ async fn attach_thread_persistence(session: &mut Session) -> PathBuf {
             base_instructions: BaseInstructions::default(),
             dynamic_tools: Vec::new(),
             multi_agent_version: None,
+            initial_rollout_copy: None,
             metadata: ThreadPersistenceMetadata {
                 cwd: Some(config.cwd.to_path_buf()),
                 model_provider: config.model_provider_id.clone(),
@@ -4013,6 +4016,7 @@ pub(crate) async fn make_session_configuration_for_tests() -> SessionConfigurati
         app_server_client_version: None,
         session_source: SessionSource::Exec,
         forked_from_thread_id: None,
+        initial_rollout_copy: None,
         parent_thread_id: None,
         thread_source: None,
         dynamic_tools: Vec::new(),
@@ -4880,6 +4884,7 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() {
         app_server_client_version: None,
         session_source: SessionSource::Exec,
         forked_from_thread_id: None,
+        initial_rollout_copy: None,
         parent_thread_id: None,
         thread_source: None,
         dynamic_tools: Vec::new(),
@@ -4993,6 +4998,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
         app_server_client_version: None,
         session_source: SessionSource::Exec,
         forked_from_thread_id: None,
+        initial_rollout_copy: None,
         parent_thread_id: None,
         thread_source: None,
         dynamic_tools: Vec::new(),
@@ -5239,6 +5245,7 @@ async fn make_session_with_config_and_rx(
         app_server_client_version: None,
         session_source: SessionSource::Exec,
         forked_from_thread_id: None,
+        initial_rollout_copy: None,
         parent_thread_id: None,
         thread_source: None,
         dynamic_tools: Vec::new(),
@@ -5346,6 +5353,7 @@ async fn make_session_with_history_source_and_agent_control_and_rx(
         app_server_client_version: None,
         session_source: session_source.clone(),
         forked_from_thread_id: None,
+        initial_rollout_copy: None,
         parent_thread_id: None,
         thread_source: None,
         dynamic_tools: Vec::new(),
@@ -6578,6 +6586,7 @@ async fn shutdown_complete_does_not_append_to_thread_store_after_shutdown() {
             base_instructions: BaseInstructions::default(),
             dynamic_tools: Vec::new(),
             multi_agent_version: None,
+            initial_rollout_copy: None,
             metadata: ThreadPersistenceMetadata {
                 cwd: Some(config.cwd.to_path_buf()),
                 model_provider: config.model_provider_id.clone(),
@@ -7053,6 +7062,7 @@ where
         app_server_client_version: None,
         session_source: SessionSource::Exec,
         forked_from_thread_id: None,
+        initial_rollout_copy: None,
         parent_thread_id: None,
         thread_source: None,
         dynamic_tools,

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

@@ -720,6 +720,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
             GUARDIAN_REVIEWER_NAME.to_string(),
         )),
         forked_from_thread_id: None,
+        initial_rollout_copy: None,
         parent_thread_id: None,
         thread_source: None,
         agent_control: AgentControl::default(),

codex-rs/core/src/thread_manager.rs

@@ -954,6 +954,32 @@ impl ThreadManager {
             thread_source,
             parent_trace,
             supports_openai_form_elicitation,
+            /*initial_rollout_copy*/ None,
+        )
+        .await
+    }
+
+    pub async fn fork_thread_from_history_with_initial_rollout_copy<S>(
+        &self,
+        snapshot: S,
+        config: Config,
+        history: InitialHistory,
+        thread_source: Option<ThreadSource>,
+        parent_trace: Option<W3cTraceContext>,
+        supports_openai_form_elicitation: bool,
+        initial_rollout_copy: PathBuf,
+    ) -> CodexResult<NewThread>
+    where
+        S: Into<ForkSnapshot>,
+    {
+        self.fork_thread_with_initial_history(
+            snapshot.into(),
+            config,
+            history,
+            thread_source,
+            parent_trace,
+            supports_openai_form_elicitation,
+            Some(initial_rollout_copy),
         )
         .await
     }
@@ -966,6 +992,7 @@ impl ThreadManager {
         thread_source: Option<ThreadSource>,
         parent_trace: Option<W3cTraceContext>,
         supports_openai_form_elicitation: bool,
+        initial_rollout_copy: Option<PathBuf>,
     ) -> CodexResult<NewThread> {
         // `forked_from_id()` describes this history's existing lineage. When
         // forking a resumed thread, the child copies the resumed thread itself.
@@ -999,23 +1026,30 @@ impl ThreadManager {
             &config.cwd,
         );
         let agent_control = self.agent_control_for_config(&config);
-        Box::pin(self.state.spawn_thread(
-            config,
-            history,
-            Arc::clone(&self.state.auth_manager),
-            agent_control,
-            /*parent_thread_id*/ None,
-            source_thread_id,
-            thread_source,
-            Vec::new(),
-            /*metrics_service_name*/ None,
-            initial_multi_agent_mode,
-            parent_trace,
-            environments,
-            /*thread_extension_init*/ ExtensionDataInit::default(),
-            supports_openai_form_elicitation,
-            /*user_shell_override*/ None,
-        ))
+        Box::pin(
+            self.state
+                .spawn_thread_with_source_and_initial_rollout_copy(
+                    config,
+                    history,
+                    Arc::clone(&self.state.auth_manager),
+                    agent_control,
+                    self.state.session_source.clone(),
+                    /*parent_thread_id*/ None,
+                    source_thread_id,
+                    thread_source,
+                    Vec::new(),
+                    /*metrics_service_name*/ None,
+                    initial_multi_agent_mode,
+                    /*inherited_environments*/ None,
+                    /*inherited_exec_policy*/ None,
+                    parent_trace,
+                    environments,
+                    /*thread_extension_init*/ ExtensionDataInit::default(),
+                    supports_openai_form_elicitation,
+                    /*user_shell_override*/ None,
+                    initial_rollout_copy,
+                ),
+        )
         .await
     }
 
@@ -1428,6 +1462,53 @@ impl ThreadManagerState {
         thread_extension_init: ExtensionDataInit,
         supports_openai_form_elicitation: bool,
         user_shell_override: Option<crate::shell::Shell>,
+    ) -> CodexResult<NewThread> {
+        Box::pin(self.spawn_thread_with_source_and_initial_rollout_copy(
+            config,
+            initial_history,
+            auth_manager,
+            agent_control,
+            session_source,
+            parent_thread_id,
+            forked_from_thread_id,
+            thread_source,
+            dynamic_tools,
+            metrics_service_name,
+            initial_multi_agent_mode,
+            inherited_environments,
+            inherited_exec_policy,
+            parent_trace,
+            environments,
+            thread_extension_init,
+            supports_openai_form_elicitation,
+            user_shell_override,
+            /*initial_rollout_copy*/ None,
+        ))
+        .await
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) async fn spawn_thread_with_source_and_initial_rollout_copy(
+        &self,
+        config: Config,
+        initial_history: InitialHistory,
+        auth_manager: Arc<AuthManager>,
+        agent_control: AgentControl,
+        session_source: SessionSource,
+        parent_thread_id: Option<ThreadId>,
+        forked_from_thread_id: Option<ThreadId>,
+        thread_source: Option<ThreadSource>,
+        dynamic_tools: Vec<codex_protocol::dynamic_tools::DynamicToolSpec>,
+        metrics_service_name: Option<String>,
+        initial_multi_agent_mode: Option<MultiAgentMode>,
+        inherited_environments: Option<TurnEnvironmentSnapshot>,
+        inherited_exec_policy: Option<Arc<crate::exec_policy::ExecPolicyManager>>,
+        parent_trace: Option<W3cTraceContext>,
+        environments: Vec<TurnEnvironmentSelection>,
+        thread_extension_init: ExtensionDataInit,
+        supports_openai_form_elicitation: bool,
+        user_shell_override: Option<crate::shell::Shell>,
+        initial_rollout_copy: Option<PathBuf>,
     ) -> CodexResult<NewThread> {
         let is_resumed_thread = matches!(&initial_history, InitialHistory::Resumed(_));
         if let InitialHistory::Resumed(resumed) = &initial_history {
@@ -1482,6 +1563,7 @@ impl ThreadManagerState {
             conversation_history: initial_history,
             session_source,
             forked_from_thread_id,
+            initial_rollout_copy,
             parent_thread_id,
             thread_source,
             agent_control,