Add fork snapshot modes

Introduce ForkSnapshotMode so callers can choose between a committed fork and one that appends the same in-distribution interrupt marker used by the live interrupt path.

Co-authored-by: Codex <noreply@openai.com>

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

@@ -183,6 +183,7 @@ use codex_core::AuthManager;
 use codex_core::CodexAuth;
 use codex_core::CodexThread;
 use codex_core::Cursor as RolloutCursor;
+use codex_core::ForkSnapshotMode;
 use codex_core::NewThread;
 use codex_core::RolloutRecorder;
 use codex_core::SessionMeta;
@@ -4044,6 +4045,7 @@ impl CodexMessageProcessor {
                 rollout_path.clone(),
                 persist_extended_history,
                 self.request_trace_context(&request_id).await,
+                ForkSnapshotMode::Committed,
             )
             .await
         {
@@ -6513,6 +6515,7 @@ impl CodexMessageProcessor {
                 rollout_path,
                 /*persist_extended_history*/ false,
                 self.request_trace_context(request_id).await,
+                ForkSnapshotMode::Committed,
             )
             .await
             .map_err(|err| JSONRPCErrorError {

codex-rs/core/src/lib.rs

@@ -97,6 +97,7 @@ mod seatbelt_permissions;
 mod thread_manager;
 pub mod web_search;
 pub mod windows_sandbox_read_grants;
+pub use thread_manager::ForkSnapshotMode;
 pub use thread_manager::NewThread;
 pub use thread_manager::ThreadManager;
 #[deprecated(note = "use ThreadManager")]

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

@@ -22,6 +22,7 @@ use tracing::warn;
 use crate::AuthManager;
 use crate::codex::Session;
 use crate::codex::TurnContext;
+use crate::contextual_user_message::TURN_ABORTED_CLOSE_TAG;
 use crate::contextual_user_message::TURN_ABORTED_OPEN_TAG;
 use crate::hook_runtime::PendingInputHookDisposition;
 use crate::hook_runtime::inspect_pending_input;
@@ -60,6 +61,22 @@ pub(crate) use user_shell::execute_user_shell_command;
 const GRACEFULL_INTERRUPTION_TIMEOUT_MS: u64 = 100;
 const TURN_ABORTED_INTERRUPTED_GUIDANCE: &str = "The user interrupted the previous turn on purpose. Any running unified exec processes may still be running in the background. If any tools/commands were aborted, they may have partially executed; verify current state before retrying.";
 
+/// Shared model-visible marker used by both the real interrupt path and
+/// interrupted fork snapshots.
+pub(crate) fn interrupted_turn_history_marker() -> ResponseItem {
+    ResponseItem::Message {
+        id: None,
+        role: "user".to_string(),
+        content: vec![ContentItem::InputText {
+            text: format!(
+                "{TURN_ABORTED_OPEN_TAG}\n{TURN_ABORTED_INTERRUPTED_GUIDANCE}\n{TURN_ABORTED_CLOSE_TAG}"
+            ),
+        }],
+        end_turn: None,
+        phase: None,
+    }
+}
+
 fn emit_turn_network_proxy_metric(
     session_telemetry: &SessionTelemetry,
     network_proxy_active: bool,
@@ -457,17 +474,7 @@ impl Session {
         if reason == TurnAbortReason::Interrupted {
             self.cleanup_after_interrupt(&task.turn_context).await;
 
-            let marker = ResponseItem::Message {
-                id: None,
-                role: "user".to_string(),
-                content: vec![ContentItem::InputText {
-                    text: format!(
-                        "{TURN_ABORTED_OPEN_TAG}\n{TURN_ABORTED_INTERRUPTED_GUIDANCE}\n</turn_aborted>"
-                    ),
-                }],
-                end_turn: None,
-                phase: None,
-            };
+            let marker = interrupted_turn_history_marker();
             self.record_into_history(std::slice::from_ref(&marker), task.turn_context.as_ref())
                 .await;
             self.persist_rollout_items(&[RolloutItem::ResponseItem(marker)])

codex-rs/core/src/thread_manager.rs

@@ -24,6 +24,7 @@ use crate::rollout::RolloutRecorder;
 use crate::rollout::truncation;
 use crate::shell_snapshot::ShellSnapshot;
 use crate::skills::SkillsManager;
+use crate::tasks::interrupted_turn_history_marker;
 use codex_protocol::ThreadId;
 use codex_protocol::config_types::CollaborationModeMask;
 #[cfg(test)]
@@ -126,6 +127,12 @@ pub struct NewThread {
     pub session_configured: SessionConfiguredEvent,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ForkSnapshotMode {
+    Committed,
+    Interrupted,
+}
+
 #[derive(Debug, Default, PartialEq, Eq)]
 pub struct ThreadShutdownReport {
     pub completed: Vec<ThreadId>,
@@ -552,9 +559,15 @@ impl ThreadManager {
         path: PathBuf,
         persist_extended_history: bool,
         parent_trace: Option<W3cTraceContext>,
+        snapshot_mode: ForkSnapshotMode,
     ) -> CodexResult<NewThread> {
-        let history = RolloutRecorder::get_rollout_history(&path).await?;
-        let history = truncate_before_nth_user_message(history, nth_user_message);
+        let history = snapshot_fork_history(
+            truncate_before_nth_user_message(
+                RolloutRecorder::get_rollout_history(&path).await?,
+                nth_user_message,
+            ),
+            snapshot_mode,
+        );
         Box::pin(self.state.spawn_thread(
             config,
             history,
@@ -851,6 +864,28 @@ fn truncate_before_nth_user_message(history: InitialHistory, n: usize) -> Initia
     }
 }
 
+fn snapshot_fork_history(
+    history: InitialHistory,
+    snapshot_mode: ForkSnapshotMode,
+) -> InitialHistory {
+    match (history, snapshot_mode) {
+        (InitialHistory::New, ForkSnapshotMode::Committed) => InitialHistory::New,
+        (InitialHistory::New, ForkSnapshotMode::Interrupted) => {
+            InitialHistory::Forked(vec![RolloutItem::ResponseItem(
+                interrupted_turn_history_marker(),
+            )])
+        }
+        (InitialHistory::Forked(history), ForkSnapshotMode::Committed) => {
+            InitialHistory::Forked(history)
+        }
+        (InitialHistory::Forked(mut history), ForkSnapshotMode::Interrupted) => {
+            history.push(RolloutItem::ResponseItem(interrupted_turn_history_marker()));
+            InitialHistory::Forked(history)
+        }
+        (InitialHistory::Resumed(_), _) => unreachable!("truncate_before_nth_user_message"),
+    }
+}
+
 #[cfg(test)]
 #[path = "thread_manager_tests.rs"]
 mod tests;

codex-rs/core/src/thread_manager_tests.rs

@@ -3,6 +3,7 @@ use crate::codex::make_session_and_context;
 use crate::config::test_config;
 use crate::models_manager::collaboration_mode_presets::CollaborationModesConfig;
 use crate::models_manager::manager::RefreshStrategy;
+use crate::tasks::interrupted_turn_history_marker;
 use assert_matches::assert_matches;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::ReasoningItemReasoningSummary;
@@ -185,3 +186,33 @@ async fn new_uses_configured_openai_provider_for_model_refresh() {
     let _ = manager.list_models(RefreshStrategy::Online).await;
     assert_eq!(models_mock.requests().len(), 1);
 }
+
+#[test]
+fn interrupted_fork_snapshot_appends_interrupt_marker() {
+    let committed_history =
+        InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello"))]);
+
+    assert_eq!(
+        serde_json::to_value(
+            snapshot_fork_history(committed_history, ForkSnapshotMode::Interrupted)
+                .get_rollout_items()
+        )
+        .expect("serialize interrupted fork history"),
+        serde_json::to_value(vec![
+            RolloutItem::ResponseItem(user_msg("hello")),
+            RolloutItem::ResponseItem(interrupted_turn_history_marker()),
+        ])
+        .expect("serialize expected interrupted fork history"),
+    );
+    assert_eq!(
+        serde_json::to_value(
+            snapshot_fork_history(InitialHistory::New, ForkSnapshotMode::Interrupted)
+                .get_rollout_items()
+        )
+        .expect("serialize interrupted empty fork history"),
+        serde_json::to_value(vec![RolloutItem::ResponseItem(
+            interrupted_turn_history_marker(),
+        )])
+        .expect("serialize expected interrupted empty fork history"),
+    );
+}

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

@@ -12,6 +12,7 @@ use super::compact::FIRST_REPLY;
 use super::compact::SUMMARY_TEXT;
 use anyhow::Result;
 use codex_core::CodexThread;
+use codex_core::ForkSnapshotMode;
 use codex_core::ThreadManager;
 use codex_core::compact::SUMMARIZATION_PROMPT;
 use codex_core::config::Config;
@@ -841,8 +842,15 @@ async fn fork_thread(
     path: std::path::PathBuf,
     nth_user_message: usize,
 ) -> Arc<CodexThread> {
-    Box::pin(manager.fork_thread(nth_user_message, config.clone(), path, false, None))
-        .await
-        .expect("fork conversation")
-        .thread
+    Box::pin(manager.fork_thread(
+        nth_user_message,
+        config.clone(),
+        path,
+        /*persist_extended_history*/ false,
+        /*parent_trace*/ None,
+        ForkSnapshotMode::Committed,
+    ))
+    .await
+    .expect("fork conversation")
+    .thread
 }

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

@@ -1,3 +1,4 @@
+use codex_core::ForkSnapshotMode;
 use codex_core::NewThread;
 use codex_core::parse_turn_item;
 use codex_protocol::items::TurnItem;
@@ -110,7 +111,14 @@ async fn fork_thread_twice_drops_to_first_message() {
         thread: codex_fork1,
         ..
     } = thread_manager
-        .fork_thread(1, config_for_fork.clone(), base_path.clone(), false, None)
+        .fork_thread(
+            1,
+            config_for_fork.clone(),
+            base_path.clone(),
+            /*persist_extended_history*/ false,
+            /*parent_trace*/ None,
+            ForkSnapshotMode::Committed,
+        )
         .await
         .expect("fork 1");
 
@@ -129,7 +137,14 @@ async fn fork_thread_twice_drops_to_first_message() {
         thread: codex_fork2,
         ..
     } = thread_manager
-        .fork_thread(0, config_for_fork.clone(), fork1_path.clone(), false, None)
+        .fork_thread(
+            0,
+            config_for_fork.clone(),
+            fork1_path.clone(),
+            /*persist_extended_history*/ false,
+            /*parent_trace*/ None,
+            ForkSnapshotMode::Committed,
+        )
         .await
         .expect("fork 2");
 

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

@@ -1,4 +1,5 @@
 use anyhow::Result;
+use codex_core::ForkSnapshotMode;
 use codex_core::config::Constrained;
 use codex_execpolicy::Policy;
 use codex_protocol::models::DeveloperInstructions;
@@ -419,7 +420,14 @@ async fn resume_and_fork_append_permissions_messages() -> Result<()> {
     fork_config.permissions.approval_policy = Constrained::allow_any(AskForApproval::UnlessTrusted);
     let forked = initial
         .thread_manager
-        .fork_thread(usize::MAX, fork_config, rollout_path, false, None)
+        .fork_thread(
+            usize::MAX,
+            fork_config,
+            rollout_path,
+            /*persist_extended_history*/ false,
+            /*parent_trace*/ None,
+            ForkSnapshotMode::Committed,
+        )
         .await?;
     forked
         .thread

codex-rs/tui/src/app.rs

@@ -57,6 +57,7 @@ use codex_app_server_protocol::RequestId;
 use codex_arg0::Arg0DispatchPaths;
 use codex_core::AuthManager;
 use codex_core::CodexAuth;
+use codex_core::ForkSnapshotMode;
 use codex_core::ThreadManager;
 use codex_core::config::Config;
 use codex_core::config::ConfigBuilder;
@@ -2507,6 +2508,7 @@ impl App {
                         target_session.path.clone(),
                         /*persist_extended_history*/ false,
                         /*parent_trace*/ None,
+                        ForkSnapshotMode::Committed,
                     )
                     .await
                     .wrap_err_with(|| {
@@ -2930,6 +2932,7 @@ impl App {
                                 path.clone(),
                                 /*persist_extended_history*/ false,
                                 /*parent_trace*/ None,
+                                ForkSnapshotMode::Committed,
                             )
                             .await
                         {